apache-spark-data-processing

Apache Spark によるデータ処理

基本的な RDD 操作から、高度なストリーミング、SQL、機械学習のワークフローまで、Apache Spark によるデータ処理を習得するための包括的なスキルです。スケーラブルな分散データパイプラインと分析システムの構築方法を学びます。

このスキルを使用する場面

次のような場合に Apache Spark を使用します。

• 大規模データの処理：単一マシンでは処理できない大きさのデータセット（TB～PB 規模）を扱う
• 分散コンピューティングの実行：クラスターのノード間で並列計算を実行する
• リアルタイムストリーム処理：連続するデータストリームを低レイテンシーで処理する
• 複雑なデータ分析：高度な分析、集計、変換を実行する
• 大規模な機械学習：膨大なデータセットで ML モデルをトレーニングする
• ETL/ELT パイプライン：堅牢なデータ変換および読み込みのワークフローを構築する
• インタラクティブなデータ分析：大規模なデータセットを探索的に分析する
• 統合データ処理：バッチ処理とストリーミングのワークロードを単一のフレームワークで組み合わせる

適さない用途：

• 単一マシンのメモリに収まる小規模なデータセット（<100 GB）
• 単純な CRUD 操作（従来型のデータベースを使用してください）
• 特殊なストリームプロセッサーが適している超低レイテンシー要件（<10ms）
• 分散データ全体で強い ACID トランザクションが必要なワークフロー

基本概念

Resilient Distributed Datasets（RDD）

RDD は Spark の基本的なデータ抽象化であり、並列処理できる、不変の分散オブジェクトコレクションです。

主な特徴：

• 耐障害性：リネージ追跡によるフォールトトレランスを備えています
• 分散性：クラスターのノード間でパーティション分割されます
• 不変性：変換によって既存の RDD を変更するのではなく、新しい RDD を作成します
• 遅延評価：変換によって計算グラフを構築し、アクションによって実行を開始します
• インメモリーコンピューティング：反復アルゴリズムの中間結果をキャッシュします

RDD 操作：

• 変換：新しい RDD を返す遅延操作（map、filter、flatMap、reduceByKey）
• アクション：計算を開始して値を返す操作（collect、count、reduce、saveAsTextFile）

RDD を使用する場面：

• データの分散とパーティション分割を低レベルで制御する場合
• 独自のパーティション分割方式が必要な場合
• 非構造化データ（テキストファイル、バイナリデータ）を扱う場合
• 古いバージョンの Spark からレガシーコードを移行する場合

可能な限り DataFrame/Dataset を優先してください。Catalyst オプティマイザーによる自動最適化が利用できます。

DataFrame と Dataset

DataFrame は、名前付きの列で構成される分散データコレクションです。データベースのテーブルや pandas の DataFrame に似ていますが、強力な最適化機能を備えています。

DataFrame：

• スキーマを持つ構造化データ
• 自動クエリー最適化（Catalyst）
• 複数言語への対応（Python、Scala、Java、R）
• SQL のような操作を行うための豊富な API

Dataset（Scala/Java のみ）：

• コンパイル時の型安全性を備えた型付き DataFrame
• JVM の最適化により Scala で最高のパフォーマンスを発揮
• RDD の型安全性と DataFrame の最適化を組み合わせたもの

RDD と比べた主な利点：

• クエリー最適化：Catalyst オプティマイザーがクエリーを書き換え、効率を高めます
• Tungsten による実行：CPU とメモリーの使用を最適化します
• 列指向ストレージ：効率的なデータ表現を実現します
• コード生成：コンパイル時のバイトコード生成により実行を高速化します

遅延評価

Spark は実行を最適化するため、遅延評価を使用します。

1. 変換によって、一連の操作を表す有向非巡回グラフ（DAG）を構築します
2. アクションによって DAG の実行を開始します
3. Spark のオプティマイザーが DAG 全体を分析し、最適化された実行計画を作成します
4. クラスターのノード間で処理を分散します

利点：

• ネットワークを介したデータ移動を最小限に抑えます
• 複数の操作を単一のステージにまとめます
• 不要な計算を排除します
• メモリー使用量を最適化します

パーティショニング

データは並列処理のためにパーティションに分割されます。

• デフォルトのパーティショニング：通常、HDFS のブロックサイズまたは入力ソースに基づきます
• ハッシュパーティショニング：キーのハッシュ値によってデータを分散します（groupByKey、reduceByKey で使用されます）
• 範囲パーティショニング：キーの範囲によってデータを分散します（ソート済みデータに便利です）
• カスタムパーティショニング：独自のパーティショニングロジックを定義します

パーティション数を決める際の考慮事項：

• パーティションが少なすぎる場合：クラスターを十分に活用できず、タスクの実行時間が長くなります
• パーティションが多すぎる場合：スケジューリングのオーバーヘッドが増え、タスクの実行時間が短くなります
• 一般的な目安：クラスター内の CPU コアあたり 2～4 個のパーティション
• repartition() または coalesce() を使用してパーティション数を調整します

キャッシュと永続化

頻繁にアクセスするデータをメモリにキャッシュして、パフォーマンスを高めます。

# Cache DataFrame in memory
df.cache()  # Shorthand for persist(StorageLevel.MEMORY_AND_DISK)

# Different storage levels
df.persist(StorageLevel.MEMORY_ONLY)       # Fast but may lose data if evicted
df.persist(StorageLevel.MEMORY_AND_DISK)   # Spill to disk if memory full
df.persist(StorageLevel.DISK_ONLY)         # Store only on disk
df.persist(StorageLevel.MEMORY_ONLY_SER)   # Serialized in memory (more compact)

# Unpersist when done
df.unpersist()

キャッシュする場合：

• ワークフロー内で複数回使用するデータ
• 反復アルゴリズム（ML のトレーニング）
• 対話型分析セッション
• 後続の処理で再利用する、コストの高い変換

キャッシュしない場合：

• 1 回しか使用しないデータ
• クラスターのメモリ容量を超える非常に大きなデータセット
• 新しいデータが継続的に追加されるストリーミングアプリケーション

Spark SQL

Spark SQL を使用すると、SQL または DataFrame API で構造化データをクエリできます。

• DataFrame やテーブルに対して SQL クエリを実行します
• DataFrame を一時ビューとして登録します
• 構造化データと半構造化データを結合します
• Hive メタストアに接続してテーブルのメタデータを取得します
• さまざまなデータソース（Parquet、ORC、JSON、CSV、JDBC）に対応しています

パフォーマンス機能：

• Catalyst オプティマイザー：ルールベースおよびコストベースのクエリ最適化
• Tungsten 実行エンジン：ステージ全体のコード生成、ベクトル化処理
• 適応型クエリ実行（AQE）：統計情報に基づく実行時の最適化
• 動的パーティションプルーニング：実行中に無関係なパーティションをスキップします

ブロードキャスト変数とアキュムレーター

分散コンピューティングを効率化する共有変数です。

ブロードキャスト変数：

• 各ノードにキャッシュされる読み取り専用変数
• 大きな読み取り専用データ（ルックアップテーブル、ML モデル）の共有に効率的です
• タスクごとに大きなデータを送信する必要がありません

# Broadcast a lookup table
lookup_table = {"key1": "value1", "key2": "value2"}
broadcast_lookup = sc.broadcast(lookup_table)

# Use in transformations
rdd.map(lambda x: broadcast_lookup.value.get(x, "default"))

アキュムレーター：

• タスク間で値を集計するための書き込み専用変数
• 分散処理でカウンターや合計値に使用します
• 集計後の値を読み取れるのはドライバーのみです

# Create accumulator
error_count = sc.accumulator(0)

# Increment in tasks
rdd.foreach(lambda x: error_count.add(1) if is_error(x) else None)

# Read final value in driver
print(f"Total errors: {error_count.value}")

Spark SQL の詳細

DataFrame の作成

さまざまなソースから DataFrame を作成します。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SparkSQLExample").getOrCreate()

# From structured data
data = [("Alice", 1), ("Bob", 2), ("Charlie", 3)]
columns = ["name", "id"]
df = spark.createDataFrame(data, columns)

# From files
df_json = spark.read.json("path/to/file.json")
df_parquet = spark.read.parquet("path/to/file.parquet")
df_csv = spark.read.option("header", "true").csv("path/to/file.csv")

# From JDBC sources
df_jdbc = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:port/database") \
    .option("dbtable", "table_name") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

DataFrame の操作

一般的な DataFrame の変換処理です。

# Select columns
df.select("name", "age").show()

# Filter rows
df.filter(df.age > 21).show()
df.where(df["age"] > 21).show()  # Alternative syntax

# Add/modify columns
from pyspark.sql.functions import col, lit
df.withColumn("age_plus_10", col("age") + 10).show()
df.withColumn("country", lit("USA")).show()

# Aggregations
df.groupBy("department").count().show()
df.groupBy("department").agg({"salary": "avg", "age": "max"}).show()

# Sorting
df.orderBy("age").show()
df.orderBy(col("age").desc()).show()

# Joins
df1.join(df2, df1.id == df2.user_id, "inner").show()
df1.join(df2, "id", "left_outer").show()

# Unions
df1.union(df2).show()

SQL クエリ

DataFrame に対して SQL を実行します。

# Register DataFrame as temporary view
df.createOrReplaceTempView("people")

# Run SQL queries
sql_result = spark.sql("SELECT name FROM people WHERE age > 21")
sql_result.show()

# Complex queries
result = spark.sql("""
    SELECT
        department,
        COUNT(*) as employee_count,
        AVG(salary) as avg_salary,
        MAX(age) as max_age
    FROM people
    WHERE age > 25
    GROUP BY department
    HAVING COUNT(*) > 5
    ORDER BY avg_salary DESC
""")
result.show()

データソース

Spark SQL は複数のデータ形式に対応しています。

Parquet（分析に推奨）：

• 列指向のストレージ形式
• 優れた圧縮率とクエリパフォーマンス
• ファイルにスキーマを埋め込みます
• 述語プッシュダウンと列プルーニングに対応しています

# Write
df.write.parquet("output/path", mode="overwrite", compression="snappy")

# Read with partition pruning
df = spark.read.parquet("output/path").filter(col("date") == "2025-01-01")

ORC（最適化された行列形式）：

• Parquet に似ていますが、圧縮率がやや優れています
• Hive との統合に適しています
• クエリを高速化するインデックスを内蔵しています

df.write.orc("output/path", mode="overwrite")
df = spark.read.orc("output/path")

JSON（半構造化データ）：

• 人が読みやすい一方、効率は低めです
• 読み込み時にスキーマを推定します
• ネストしたデータや複雑なデータに適しています

# Read with explicit schema
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])

df = spark.read.schema(schema).json("data.json")

CSV（レガシー／シンプルなデータ）：

• 幅広く互換性がありますが、処理速度は遅めです
• ヘッダーの推定または明示的なスキーマが必要です
• 圧縮によるメリットは最小限です

df.write.csv("output.csv", header=True, mode="overwrite")
df = spark.read.option("header", "true").option("inferSchema", "true").csv("data.csv")

ウィンドウ関数

ウィンドウ関数による高度な分析です。

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank, lag, lead, sum, avg

# Define window specification
window_spec = Window.partitionBy("department").orderBy(col("salary").desc())

# Ranking functions
df.withColumn("rank", rank().over(window_spec)).show()
df.withColumn("row_num", row_number().over(window_spec)).show()
df.withColumn("dense_rank", dense_rank().over(window_spec)).show()

# Aggregate functions over window
df.withColumn("dept_avg_salary", avg("salary").over(window_spec)).show()
df.withColumn("running_total", sum("salary").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow))).show()

# Offset functions
df.withColumn("prev_salary", lag("salary", 1).over(window_spec)).show()
df.withColumn("next_salary", lead("salary", 1).over(window_spec)).show()

ユーザー定義関数（UDF）

カスタム変換を作成します。

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, IntegerType

# Python UDF (slower due to serialization overhead)
def categorize_age(age):
    if age < 18:
        return "Minor"
    elif age < 65:
        return "Adult"
    else:
        return "Senior"

categorize_udf = udf(categorize_age, StringType())
df.withColumn("age_category", categorize_udf(col("age"))).show()

# Pandas UDF (vectorized, faster for large datasets)
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf(IntegerType())
def square(series: pd.Series) -> pd.Series:
    return series ** 2

df.withColumn("age_squared", square(col("age"))).show()

UDF のパフォーマンスに関するヒント：

• 可能な場合は、組み込みの Spark 関数を使用します（常に高速です）
• パフォーマンスを高めるには、Python UDF より Pandas UDF を優先します
• 最高のパフォーマンスを得るには、Scala UDF を使用します（シリアライズのオーバーヘッドがありません）
• DataFrame を複数回使用する場合は、UDF を適用する前にキャッシュします

変換とアクション

一般的な変換

map：各要素に関数を適用します

# RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])
squared = rdd.map(lambda x: x * 2)  # [2, 4, 6, 8, 10]

# DataFrame (use select with functions)
from pyspark.sql.functions import col
df.select(col("value") * 2).show()

filter：述語に一致する要素を選択します

# RDD
rdd.filter(lambda x: x > 2).collect()  # [3, 4, 5]

# DataFrame
df.filter(col("age") > 25).show()

flatMap：結果をマッピングして平坦化します

# RDD - Split text into words
lines = sc.parallelize(["hello world", "apache spark"])
words = lines.flatMap(lambda line: line.split(" "))  # ["hello", "world", "apache", "spark"]

reduceByKey：キーごとに値を集約します

# Word count example
words = sc.parallelize(["apple", "banana", "apple", "cherry", "banana", "apple"])
word_pairs = words.map(lambda word: (word, 1))
word_counts = word_pairs.reduceByKey(lambda a, b: a + b)
# Result: [("apple", 3), ("banana", 2), ("cherry", 1)]

groupByKey：キーごとに値をグループ化します（可能な限り避け、代わりにreduceByKeyを使用してください）

# Less efficient than reduceByKey
word_pairs.groupByKey().mapValues(list).collect()
# Result: [("apple", [1, 1, 1]), ("banana", [1, 1]), ("cherry", [1])]

join：キーに基づいてデータセットを結合します

# RDD join
users = sc.parallelize([("user1", "Alice"), ("user2", "Bob")])
orders = sc.parallelize([("user1", 100), ("user2", 200), ("user1", 150)])
users.join(orders).collect()
# Result: [("user1", ("Alice", 100)), ("user1", ("Alice", 150)), ("user2", ("Bob", 200))]

# DataFrame join (more efficient)
df_users.join(df_orders, "user_id", "inner").show()

distinct：重複を削除します

# RDD
rdd.distinct().collect()

# DataFrame
df.distinct().show()
df.dropDuplicates(["user_id"]).show()  # Drop based on specific columns

coalesce/repartition：パーティション数を変更します

# Reduce partitions (no shuffle, more efficient)
df.coalesce(1).write.parquet("output")

# Increase/decrease partitions (involves shuffle)
df.repartition(10).write.parquet("output")
df.repartition(10, "user_id").write.parquet("output")  # Partition by column

一般的なアクション

collect：すべてのデータをドライバーに取得します

results = rdd.collect()  # Returns list
# WARNING: Only use on small datasets that fit in driver memory

count：要素数を数えます

total = df.count()  # Number of rows

first/take：最初のN個の要素を取得します

first_elem = rdd.first()
first_five = rdd.take(5)

reduce：すべての要素を集約します

total_sum = rdd.reduce(lambda a, b: a + b)

foreach：各要素に対して関数を実行します

# Side effects only (no return value)
rdd.foreach(lambda x: print(x))

saveAsTextFile：ファイルシステムに書き込みます

rdd.saveAsTextFile("hdfs://path/to/output")

show：DataFrameの行を表示します（アクション）

df.show(20, truncate=False)  # Show 20 rows, don't truncate columns

構造化ストリーミング

DataFrame APIを使用して、連続するデータストリームを処理します。

中核となる概念

ストリーミングDataFrame：

• 継続的に増加する無制限のテーブル
• バッチDataFrameと同じ操作を使用できます
• マイクロバッチ処理（デフォルト）または連続処理

入力ソース：

• ファイルソース（JSON、Parquet、CSV、ORC、text）
• Kafka
• Socket（テスト用）
• Rateソース（テスト用）
• カスタムソース

出力モード：

• Append：結果テーブルに追加された新しい行のみを書き込みます
• Complete：トリガーごとに結果テーブル全体を書き込みます
• Update：更新された行のみを書き込みます

出力シンク：

• ファイルシンク（Parquet、ORC、JSON、CSV、text）
• Kafka
• Console（デバッグ用）
• Memory（テスト用）
• Foreach/ForeachBatch（カスタムロジック）

基本的なストリーミングの例

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName("StreamingExample").getOrCreate()

# Read stream from JSON files
input_stream = spark.readStream \
    .format("json") \
    .schema(schema) \
    .option("maxFilesPerTrigger", 1) \
    .load("input/directory")

# Transform streaming data
processed = input_stream \
    .filter(col("value") > 10) \
    .select("id", "value", "timestamp")

# Write stream to Parquet
query = processed.writeStream \
    .format("parquet") \
    .option("path", "output/directory") \
    .option("checkpointLocation", "checkpoint/directory") \
    .outputMode("append") \
    .start()

# Wait for termination
query.awaitTermination()

ストリーミングデータと静的データの結合

ストリーミングデータを静的な参照データと結合します：

# Static DataFrame (loaded once)
static_df = spark.read.parquet("reference/data")

# Streaming DataFrame
streaming_df = spark.readStream.format("kafka").load()

# Inner join (supported)
joined = streaming_df.join(static_df, "type")

# Left outer join (supported)
joined = streaming_df.join(static_df, "type", "left_outer")

# Write result
joined.writeStream \
    .format("parquet") \
    .option("path", "output") \
    .option("checkpointLocation", "checkpoint") \
    .start()

ウィンドウ集計

時間ウィンドウごとにデータを集計します：

from pyspark.sql.functions import window, col, count

# 10-minute tumbling window
windowed_counts = streaming_df \
    .groupBy(
        window(col("timestamp"), "10 minutes"),
        col("word")
    ) \
    .count()

# 10-minute sliding window with 5-minute slide
windowed_counts = streaming_df \
    .groupBy(
        window(col("timestamp"), "10 minutes", "5 minutes"),
        col("word")
    ) \
    .count()

# Write to console for debugging
query = windowed_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("truncate", "false") \
    .start()

遅延データのウォーターマーク処理

ウォーターマークを使用して、遅れて到着したデータを処理します：

from pyspark.sql.functions import window

# Define watermark (10 minutes tolerance for late data)
windowed_counts = streaming_df \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(col("timestamp"), "10 minutes"),
        col("word")
    ) \
    .count()

# Data arriving more than 10 minutes late will be dropped

ウォーターマークの利点：

• 古い集計状態を破棄して状態のサイズを制限します
• 許容ウィンドウ内の遅延データを処理します
• 無限の状態を維持しないことでパフォーマンスを向上させます

セッションウィンドウ

非アクティブな間隔に基づいてイベントをセッションにグループ化します：

from pyspark.sql.functions import session_window, when

# Dynamic session window based on user
session_window_spec = session_window(
    col("timestamp"),
    when(col("userId") == "user1", "5 seconds")
    .when(col("userId") == "user2", "20 seconds")
    .otherwise("5 minutes")
)

sessionized_counts = streaming_df \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(session_window_spec, col("userId")) \
    .count()

ステートフルなストリーム処理

マイクロバッチ間で状態を維持します：

from pyspark.sql.functions import expr

# Deduplication using state
deduplicated = streaming_df \
    .withWatermark("timestamp", "1 hour") \
    .dropDuplicates(["user_id", "event_id"])

# Stream-stream joins (stateful)
stream1 = spark.readStream.format("kafka").option("subscribe", "topic1").load()
stream2 = spark.readStream.format("kafka").option("subscribe", "topic2").load()

joined = stream1 \
    .withWatermark("timestamp", "10 minutes") \
    .join(
        stream2.withWatermark("timestamp", "20 minutes"),
        expr("stream1.user_id = stream2.user_id AND stream1.timestamp >= stream2.timestamp AND stream1.timestamp <= stream2.timestamp + interval 15 minutes"),
        "inner"
    )

チェックポイント

チェックポイントを使用してフォールトトレランスを確保します：

# Checkpoint location stores:
# - Stream metadata (offsets, configuration)
# - State information (for stateful operations)
# - Write-ahead logs

query = streaming_df.writeStream \
    .format("parquet") \
    .option("path", "output") \
    .option("checkpointLocation", "checkpoint/dir")  # REQUIRED for production \
    .start()

# Recovery: Restart query with same checkpoint location
# Spark will resume from last committed offset

チェックポイントのベストプラクティス：

• 本番ストリームでは必ずcheckpointLocationを設定してください
• チェックポイントには信頼性の高い分散ストレージ（HDFS、S3）を使用してください
• ストリームの実行中にチェックポイントディレクトリを削除しないでください
• 災害復旧に備えてチェックポイントをバックアップしてください

MLlibによる機械学習

Sparkのスケーラブルな機械学習ライブラリです。

中核となるコンポーネント

MLlibの機能：

• MLパイプライン：変換とモデルを連結します
• 特徴量化：ベクトルアセンブラー、スケーラー、エンコーダー
• 分類と回帰：線形モデル、ツリーベースのモデル、ニューラルネットワーク
• クラスタリング：K-means、Gaussian Mixture、LDA
• 協調フィルタリング：ALS（Alternating Least Squares）
• 次元削減：PCA、SVD
• モデル選択：交差検証、訓練データとテストデータの分割、パラメーター調整

MLパイプライン

変換と推定器を連結します：

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression

# Load data
df = spark.read.format("libsvm").load("data/sample_libsvm_data.txt")

# Define pipeline stages
assembler = VectorAssembler(
    inputCols=["feature1", "feature2", "feature3"],
    outputCol="features"
)

scaler = StandardScaler(
    inputCol="features",
    outputCol="scaled_features",
    withStd=True,
    withMean=True
)

lr = LogisticRegression(
    featuresCol="scaled_features",
    labelCol="label",
    maxIter=10,
    regParam=0.01
)

# Create pipeline
pipeline = Pipeline(stages=[assembler, scaler, lr])

# Split data
train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)

# Train model
model = pipeline.fit(train_df)

# Make predictions
predictions = model.transform(test_df)
predictions.select("label", "prediction", "probability").show()

特徴量エンジニアリング

生データを特徴量に変換します：

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler, MinMaxScaler

# Categorical encoding
indexer = StringIndexer(inputCol="category", outputCol="category_index")
encoder = OneHotEncoder(inputCol="category_index", outputCol="category_vec")

# Numerical scaling
scaler = MinMaxScaler(inputCol="features", outputCol="scaled_features")

# Assemble features
assembler = VectorAssembler(
    inputCols=["category_vec", "numeric_feature1", "numeric_feature2"],
    outputCol="features"
)

# Text processing
from pyspark.ml.feature import Tokenizer, HashingTF, IDF

tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashing_tf = HashingTF(inputCol="words", outputCol="raw_features", numFeatures=10000)
idf = IDF(inputCol="raw_features", outputCol="features")

ストリーミング線形回帰

ストリーミングデータでモデルを訓練します：

from pyspark.mllib.regression import LabeledPoint
from pyspark.streaming import StreamingContext
from pyspark.streaming.ml import StreamingLinearRegressionWithSGD

# Create StreamingContext
ssc = StreamingContext(sc, batchDuration=1)

# Define data streams
training_stream = ssc.textFileStream("training/data/path")
testing_stream = ssc.textFileStream("testing/data/path")

# Parse streams into LabeledPoint objects
def parse_point(line):
    values = [float(x) for x in line.strip().split(',')]
    return LabeledPoint(values[0], values[1:])

parsed_training = training_stream.map(parse_point)
parsed_testing = testing_stream.map(parse_point)

# Initialize model
num_features = 3
model = StreamingLinearRegressionWithSGD(initialWeights=[0.0] * num_features)

# Train and predict
model.trainOn(parsed_training)
predictions = model.predictOnValues(parsed_testing.map(lambda lp: (lp.label, lp.features)))

# Print predictions
predictions.pprint()

# Start streaming
ssc.start()
ssc.awaitTermination()

モデル評価

モデルのパフォーマンスを評価します：

from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator, RegressionEvaluator

# Binary classification
binary_evaluator = BinaryClassificationEvaluator(
    labelCol="label",
    rawPredictionCol="rawPrediction",
    metricName="areaUnderROC"
)
auc = binary_evaluator.evaluate(predictions)
print(f"AUC: {auc}")

# Multiclass classification
multi_evaluator = MulticlassClassificationEvaluator(
    labelCol="label",
    predictionCol="prediction",
    metricName="accuracy"
)
accuracy = multi_evaluator.evaluate(predictions)
print(f"Accuracy: {accuracy}")

# Regression
regression_evaluator = RegressionEvaluator(
    labelCol="label",
    predictionCol="prediction",
    metricName="rmse"
)
rmse = regression_evaluator.evaluate(predictions)
print(f"RMSE: {rmse}")

ハイパーパラメーター調整

交差検証でモデルパラメーターを最適化します：

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Define model
rf = RandomForestClassifier(labelCol="label", featuresCol="features")

# Build parameter grid
param_grid = ParamGridBuilder() \
    .addGrid(rf.numTrees, [10, 20, 50]) \
    .addGrid(rf.maxDepth, [5, 10, 15]) \
    .addGrid(rf.minInstancesPerNode, [1, 5, 10]) \
    .build()

# Define evaluator
evaluator = MulticlassClassificationEvaluator(metricName="accuracy")

# Cross-validation
cv = CrossValidator(
    estimator=rf,
    estimatorParamMaps=param_grid,
    evaluator=evaluator,
    numFolds=5,
    parallelism=4
)

# Train
cv_model = cv.fit(train_df)

# Best model
best_model = cv_model.bestModel
print(f"Best numTrees: {best_model.getNumTrees}")
print(f"Best maxDepth: {best_model.getMaxDepth()}")

# Evaluate on test set
predictions = cv_model.transform(test_df)
accuracy = evaluator.evaluate(predictions)
print(f"Test Accuracy: {accuracy}")

分散行列演算

MLlibは分散行列の表現を提供します：

from pyspark.mllib.linalg.distributed import RowMatrix, IndexedRowMatrix, CoordinateMatrix
from pyspark.mllib.linalg import Vectors

# RowMatrix: Distributed matrix without row indices
rows = sc.parallelize([
    Vectors.dense([1.0, 2.0, 3.0]),
    Vectors.dense([4.0, 5.0, 6.0]),
    Vectors.dense([7.0, 8.0, 9.0])
])
row_matrix = RowMatrix(rows)

# Compute statistics
print(f"Rows: {row_matrix.numRows()}")
print(f"Cols: {row_matrix.numCols()}")
print(f"Column means: {row_matrix.computeColumnSummaryStatistics().mean()}")

# IndexedRowMatrix: Matrix with row indices
from pyspark.mllib.linalg.distributed import IndexedRow
indexed_rows = sc.parallelize([
    IndexedRow(0, Vectors.dense([1.0, 2.0, 3.0])),
    IndexedRow(1, Vectors.dense([4.0, 5.0, 6.0]))
])
indexed_matrix = IndexedRowMatrix(indexed_rows)

# CoordinateMatrix: Sparse matrix using (row, col, value) entries
from pyspark.mllib.linalg.distributed import MatrixEntry
entries = sc.parallelize([
    MatrixEntry(0, 0, 1.0),
    MatrixEntry(0, 2, 3.0),
    MatrixEntry(1, 1, 5.0)
])
coord_matrix = CoordinateMatrix(entries)

層化サンプリング

クラス分布を維持しながらデータをサンプリングします：

# Scala/Java approach
data = [("a", 1), ("b", 2), ("a", 3), ("b", 4), ("a", 5), ("c", 6)]
rdd = sc.parallelize(data)

# Define sampling fractions per key
fractions = {"a": 0.5, "b": 0.5, "c": 0.5}

# Approximate sample (faster, one pass)
sampled_rdd = rdd.sampleByKey(withReplacement=False, fractions=fractions)

# Exact sample (slower, guaranteed exact counts)
exact_sampled = rdd.sampleByKeyExact(withReplacement=False, fractions=fractions)

print(sampled_rdd.collect())

パフォーマンスチューニング

メモリ管理

メモリの内訳：

• 実行メモリ：シャッフル、結合、ソート、集約に使用します
• ストレージメモリ：キャッシュとブロードキャスト変数に使用します
• ユーザーメモリ：ユーザーデータ構造と UDF に使用します
• 予約メモリ：Spark の内部処理用に予約されています

設定：

spark = SparkSession.builder \
    .appName("MemoryTuning") \
    .config("spark.executor.memory", "4g") \
    .config("spark.driver.memory", "2g") \
    .config("spark.memory.fraction", "0.6")  # Fraction for execution + storage \
    .config("spark.memory.storageFraction", "0.5")  # Fraction of above for storage \
    .getOrCreate()

メモリ管理のベストプラクティス：

• Spark UI でメモリ使用量を監視します
• キャッシュには適切なストレージレベルを使用します
• 大規模なデータセットをドライバーに収集しないようにします
• メモリを多く消費する処理では、エグゼキューターのメモリを増やします
• メモリ効率を高めるため、Kryo シリアル化を使用します

シャッフルの最適化

シャッフルはコストの高い処理です。できるだけ減らしましょう。

シャッフルの原因：

• groupByKey、reduceByKey、aggregateByKey
• join、cogroup
• repartition、coalesce（増加させる場合）
• distinct、intersection
• sortByKey

最適化戦略：

# 1. Use reduceByKey instead of groupByKey
# Bad: groupByKey shuffles all data
word_pairs.groupByKey().mapValues(sum)

# Good: reduceByKey combines locally before shuffle
word_pairs.reduceByKey(lambda a, b: a + b)

# 2. Broadcast small tables in joins
from pyspark.sql.functions import broadcast
large_df.join(broadcast(small_df), "key")

# 3. Partition data appropriately
df.repartition(200, "user_id")  # Partition by key for subsequent aggregations

# 4. Coalesce instead of repartition when reducing partitions
df.coalesce(10)  # No shuffle, just merge partitions

# 5. Tune shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", 200)  # Default is 200

シャッフルの設定：

spark = SparkSession.builder \
    .config("spark.sql.shuffle.partitions", 200) \
    .config("spark.default.parallelism", 200) \
    .config("spark.sql.adaptive.enabled", "true")  # Enable AQE \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

パーティション分割の戦略

パーティション数の目安：

• 少なすぎる場合：クラスターが十分に活用されず、OOM エラーが発生します
• 多すぎる場合：タスクのスケジューリングにかかるオーバーヘッドが増加します
• 適切な範囲：CPU コア数の 2～4 倍
• 大規模なシャッフルの場合：100～200 以上のパーティション

列によるパーティション分割：

# Partition writes by date for easy filtering
df.write.partitionBy("date", "country").parquet("output")

# Read with partition pruning (only reads relevant partitions)
spark.read.parquet("output").filter(col("date") == "2025-01-15").show()

カスタムパーティション分割：

from pyspark.rdd import portable_hash

# Custom partitioner for RDD
def custom_partitioner(key):
    return portable_hash(key) % 100

rdd.partitionBy(100, custom_partitioner)

キャッシュ戦略

キャッシュするタイミング：

# Iterative algorithms (ML)
training_data.cache()
for i in range(num_iterations):
    model = train_model(training_data)

# Multiple aggregations on same data
base_df.cache()
result1 = base_df.groupBy("country").count()
result2 = base_df.groupBy("city").avg("sales")

# Interactive analysis
df.cache()
df.filter(condition1).show()
df.filter(condition2).show()
df.groupBy("category").count().show()

ストレージレベル：

from pyspark import StorageLevel

# Memory only (fastest, but may lose data)
df.persist(StorageLevel.MEMORY_ONLY)

# Memory and disk (spill to disk if needed)
df.persist(StorageLevel.MEMORY_AND_DISK)

# Serialized in memory (more compact, slower access)
df.persist(StorageLevel.MEMORY_ONLY_SER)

# Disk only (slowest, but always available)
df.persist(StorageLevel.DISK_ONLY)

# Replicated (fault tolerance)
df.persist(StorageLevel.MEMORY_AND_DISK_2)  # 2 replicas

ブロードキャスト結合

小さなテーブルとの結合を最適化します。

from pyspark.sql.functions import broadcast

# Automatic broadcast (tables < spark.sql.autoBroadcastJoinThreshold)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10 * 1024 * 1024)  # 10 MB

# Explicit broadcast hint
large_df.join(broadcast(small_df), "key")

# Benefits:
# - No shuffle of large table
# - Small table sent to all executors once
# - Much faster for small dimension tables

アダプティブクエリー実行（AQE）

実行時のクエリー最適化を有効にします。

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# AQE Benefits:
# - Dynamically coalesce partitions after shuffle
# - Handle skewed joins by splitting large partitions
# - Optimize join strategy at runtime

データ形式の選択

パフォーマンスの比較：

1. Parquet（分析に最適）：列指向で圧縮され、クエリーが高速です
2. ORC（Hive に最適）：Parquet と同様で、圧縮率がやや高くなります
3. Avro（行指向に最適）：書き込みの多いワークロードに適しています
4. JSON（最も低速）：人間が読みやすい形式ですが、効率はよくありません
5. CSV（レガシー）：互換性がありますが、低速でスキーマもありません

推奨事項：

• ほとんどの分析ワークロードでは Parquet を使用します
• 圧縮（snappy、gzip、lzo）を有効にします
• よくフィルターに使用する列でパーティション分割します
• 読み取りの多いワークロードでは列指向形式を使用します

Catalyst オプティマイザー

クエリーの最適化について理解します。

# View physical plan
df.explain(mode="extended")

# Optimizations include:
# - Predicate pushdown: Push filters to data source
# - Column pruning: Read only required columns
# - Constant folding: Evaluate constants at compile time
# - Join reordering: Optimize join order
# - Partition pruning: Skip irrelevant partitions

本番環境へのデプロイ

クラスターマネージャー

Standalone：

• シンプルな組み込みのクラスターマネージャーです
• 開発環境や小規模なクラスターに簡単にセットアップできます
• 他のフレームワークとリソースを共有しません

# Start master
$SPARK_HOME/sbin/start-master.sh

# Start workers
$SPARK_HOME/sbin/start-worker.sh spark://master:7077

# Submit application
spark-submit --master spark://master:7077 app.py

YARN：

• Hadoop のリソースマネージャーです
• MapReduce、Hive などとクラスターリソースを共有します
• 2 つのモードがあります：cluster（YARN 上でドライバーを実行）と client（ローカルマシン上でドライバーを実行）

# Cluster mode (driver runs on YARN)
spark-submit --master yarn --deploy-mode cluster app.py

# Client mode (driver runs locally)
spark-submit --master yarn --deploy-mode client app.py

Kubernetes：

• 最新のコンテナーオーケストレーションです
• リソースを動的に割り当てます
• クラウドネイティブなデプロイに対応します

spark-submit \
  --master k8s://https://k8s-master:443 \
  --deploy-mode cluster \
  --name spark-app \
  --conf spark.executor.instances=5 \
  --conf spark.kubernetes.container.image=spark:latest \
  app.py

Mesos：

• 汎用クラスターマネージャーです
• リソースを細粒度または粗粒度で共有します

アプリケーションの送信

基本的な spark-submit：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 4g \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 10 \
  --conf spark.sql.shuffle.partitions=200 \
  --py-files dependencies.zip \
  --files config.json \
  application.py

設定オプション：

• --master：クラスターマネージャーの URL
• --deploy-mode：ドライバーの実行場所（client または cluster）
• --driver-memory：ドライバープロセスのメモリ
• --executor-memory：エグゼキューターあたりのメモリ
• --executor-cores：エグゼキューターあたりのコア数
• --num-executors：エグゼキューターの数
• --conf：Spark の設定プロパティ
• --py-files：Python の依存関係
• --files：配布する追加ファイル

リソース割り当て

一般的な目安：

• ドライバーのメモリ：1～4 GB（大規模な結果を収集する場合を除く）
• エグゼキューターのメモリ：エグゼキューターあたり 4～16 GB
• エグゼキューターのコア数：エグゼキューターあたり 4～5 コア（5 を超えると効果が薄れます）
• エグゼキューターの数：クラスターの容量を使い切りつつ、OS や他のサービス用のリソースを残します
• 並列度：合計コア数の 2～4 倍

計算例：

Cluster: 10 nodes, 32 cores each, 128 GB RAM each

Option 1: Many small executors
  - 30 executors (3 per node)
  - 10 cores per executor
  - 40 GB memory per executor
  - Total: 300 cores

Option 2: Fewer large executors (RECOMMENDED)
  - 50 executors (5 per node)
  - 5 cores per executor
  - 24 GB memory per executor
  - Total: 250 cores

動的割り当て

ワークロードに応じてエグゼキューターを自動的にスケールします。

spark = SparkSession.builder \
    .appName("DynamicAllocation") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", 2) \
    .config("spark.dynamicAllocation.maxExecutors", 100) \
    .config("spark.dynamicAllocation.initialExecutors", 10) \
    .config("spark.dynamicAllocation.executorIdleTimeout", "60s") \
    .getOrCreate()

メリット：

• リソースをより効率的に活用できます
• 変動するワークロードに合わせて自動的にスケールします
• クラウド環境でコストを削減できます

監視とログ記録

Spark UI：

• http://driver:4040 の Web UI
• ステージ、タスク、ストレージ、環境、エグゼキューター
• SQL クエリプランと実行の詳細
• ボトルネックやパフォーマンスの問題を特定

History Server：

# Start history server
$SPARK_HOME/sbin/start-history-server.sh

# Configure event logging
spark.conf.set("spark.eventLog.enabled", "true")
spark.conf.set("spark.eventLog.dir", "hdfs://namenode/spark-logs")

メトリクス：

# Enable metrics collection
spark.conf.set("spark.metrics.conf.*.sink.console.class", "org.apache.spark.metrics.sink.ConsoleSink")
spark.conf.set("spark.metrics.conf.*.sink.console.period", 10)

ログ記録：

# Configure log level
spark.sparkContext.setLogLevel("WARN")  # ERROR, WARN, INFO, DEBUG

# Custom logging
import logging
logger = logging.getLogger(__name__)
logger.info("Custom log message")

フォールトトレランス

自動復旧：

• タスクの失敗：失敗したタスクを自動的に再試行します
• エグゼキューターの失敗：ほかのエグゼキューターでタスクを再スケジュールします
• ドライバーの失敗：チェックポイントから復元します（ストリーミング）
• ノードの失敗：リネージから失われたパーティションを再計算します

チェックポイント：

# Set checkpoint directory
spark.sparkContext.setCheckpointDir("hdfs://namenode/checkpoints")

# Checkpoint RDD (breaks lineage for very long chains)
rdd.checkpoint()

# Streaming checkpoint (required for production)
query = streaming_df.writeStream \
    .option("checkpointLocation", "hdfs://namenode/streaming-checkpoint") \
    .start()

投機的実行：

# Enable speculative execution for slow tasks
spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", 1.5)
spark.conf.set("spark.speculation.quantile", 0.75)

データローカリティ

パフォーマンス向上のためにデータ配置を最適化します。

ローカリティレベル：

1. PROCESS_LOCAL：タスクと同じ JVM 内にあるデータ（最速）
2. NODE_LOCAL：同じノード上の別プロセスにあるデータ
3. RACK_LOCAL：同じラック内にあるデータ
4. ANY：別のラックにあるデータ（最遅）

ローカリティの改善：

# Increase locality wait time
spark.conf.set("spark.locality.wait", "10s")
spark.conf.set("spark.locality.wait.node", "5s")
spark.conf.set("spark.locality.wait.rack", "3s")

# Partition data to match cluster topology
df.repartition(num_nodes * cores_per_node)

ベストプラクティス

コード構成

1. モジュール設計：データの読み込み、変換、出力のロジックを分離します
2. 構成管理：構成を外部化します（構成ファイルを使用）
3. エラー処理：堅牢なエラー処理とログ記録を実装します
4. テスト：変換処理をユニットテストし、パイプラインを統合テストします
5. ドキュメント：複雑な変換処理とビジネスロジックを文書化します

パフォーマンス

1. シャッフルを避ける：groupByKey の代わりに reduceByKey を使用します
2. キャッシュを賢く使う：複数回再利用されるデータだけをキャッシュします
3. 小さなテーブルをブロードキャストする：小さな参照データにはブロードキャスト結合を使用します
4. 適切にパーティション分割する：CPU コア数の 2～4 倍にし、頻繁にフィルターされる列でパーティション分割します
5. Parquet を使用する：分析ワークロードには列指向形式を使用します
6. AQE を有効にする：実行時最適化には適応型クエリ実行を活用します
7. メモリを調整する：エグゼキューターのメモリとコアのバランスを取ります
8. 監視する：Spark UI を使用してボトルネックを特定します

開発ワークフロー

1. 小さく始める：ローカルでサンプルデータを使って開発します
2. 早期にプロファイルする：最初からパフォーマンスを監視します
3. 反復する：メトリクスに基づいて段階的に最適化します
4. 大規模にテストする：デプロイ前に本番環境相当のデータで検証します
5. バージョン管理：コード、構成、スキーマを追跡します

データ品質

1. スキーマ検証：読み込み時と書き込み時にスキーマを適用します
2. NULL の処理：NULL 値を明示的に処理します
3. データ検証：想定される範囲、形式、制約を確認します
4. 重複排除：ビジネスロジックに基づいて重複を削除します
5. 監査ログ：データリネージと変換処理を追跡します

セキュリティ

1. 認証：YARN/HDFS で Kerberos を有効にします
2. 認可：データアクセス制御に ACL を使用します
3. 暗号化：保存中と転送中のデータを暗号化します
4. シークレット管理：安全な認証情報プロバイダーを使用します
5. 監査証跡：データへのアクセスと変更を記録します

コスト最適化

1. 適切なサイズのリソースを選ぶ：エグゼキューターを過剰に割り当てないようにします
2. 動的割り当て：ワークロードに応じてエグゼキューターをスケーリングします
3. スポットインスタンス：クラウドでスポット／プリエンプティブルインスタンスを使用します
4. データ圧縮：効率的な形式（Parquet、ORC）を使用します
5. パーティション分割：不要なデータ読み込みを枝刈りします
6. 自動シャットダウン：アイドル状態のクラスターを終了します

よくあるパターン

ETL パイプラインパターン

def etl_pipeline(spark, input_path, output_path):
    # Extract
    raw_df = spark.read.parquet(input_path)

    # Transform
    cleaned_df = raw_df \
        .dropDuplicates(["id"]) \
        .filter(col("value").isNotNull()) \
        .withColumn("processed_date", current_date())

    # Enrich
    enriched_df = cleaned_df.join(broadcast(reference_df), "key")

    # Aggregate
    aggregated_df = enriched_df \
        .groupBy("category", "date") \
        .agg(
            count("*").alias("count"),
            sum("amount").alias("total_amount"),
            avg("value").alias("avg_value")
        )

    # Load
    aggregated_df.write \
        .partitionBy("date") \
        .mode("overwrite") \
        .parquet(output_path)

増分処理パターン

def incremental_process(spark, input_path, output_path, checkpoint_path):
    # Read last processed timestamp
    last_timestamp = read_checkpoint(checkpoint_path)

    # Read new data
    new_data = spark.read.parquet(input_path) \
        .filter(col("timestamp") > last_timestamp)

    # Process
    processed = transform(new_data)

    # Write
    processed.write.mode("append").parquet(output_path)

    # Update checkpoint
    max_timestamp = new_data.agg(max("timestamp")).collect()[0][0]
    write_checkpoint(checkpoint_path, max_timestamp)

緩やかに変化するディメンション（SCD）パターン

def scd_type2_upsert(spark, dimension_df, updates_df):
    # Mark existing records as inactive if updated
    inactive_records = dimension_df \
        .join(updates_df, "business_key") \
        .select(
            dimension_df["*"],
            lit(False).alias("is_active"),
            current_date().alias("end_date")
        )

    # Add new records
    new_records = updates_df \
        .withColumn("is_active", lit(True)) \
        .withColumn("start_date", current_date()) \
        .withColumn("end_date", lit(None))

    # Union unchanged, inactive, and new records
    result = dimension_df \
        .join(updates_df, "business_key", "left_anti") \
        .union(inactive_records) \
        .union(new_records)

    return result

ウィンドウ分析パターン

def calculate_running_metrics(df):
    from pyspark.sql.window import Window
    from pyspark.sql.functions import row_number, lag, sum, avg

    # Define window
    window_spec = Window.partitionBy("user_id").orderBy("timestamp")

    # Calculate metrics
    result = df \
        .withColumn("row_num", row_number().over(window_spec)) \
        .withColumn("prev_value", lag("value", 1).over(window_spec)) \
        .withColumn("running_total", sum("value").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow))) \
        .withColumn("moving_avg", avg("value").over(window_spec.rowsBetween(-2, 0)))

    return result

トラブルシューティング

メモリ不足エラー

症状：

• java.lang.OutOfMemoryError
• エグゼキューターの失敗
• ガベージコレクションの遅延

解決策：

# Increase executor memory
spark.conf.set("spark.executor.memory", "8g")

# Increase driver memory (if collecting data)
spark.conf.set("spark.driver.memory", "4g")

# Reduce memory pressure
df.persist(StorageLevel.MEMORY_AND_DISK)  # Spill to disk
df.coalesce(100)  # Reduce partition count
spark.conf.set("spark.sql.shuffle.partitions", 400)  # Increase shuffle partitions

# Avoid collect() on large datasets
# Use take() or limit() instead
df.take(100)

シャッフルのパフォーマンス問題

症状：

• シャッフルの読み込み／書き込み時間が長い
• パーティションサイズに偏りがある
• タスクの遅延

解決策：

# Increase shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", 400)

# Handle skew with salting
df_salted = df.withColumn("salt", (rand() * 10).cast("int"))
result = df_salted.groupBy("key", "salt").agg(...)

# Use broadcast for small tables
large_df.join(broadcast(small_df), "key")

# Enable AQE for automatic optimization
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

ストリーミングジョブの失敗

症状：

• ストリーミングクエリが停止した
• チェックポイントが破損している
• 処理の遅延が増加している

解決策：

# Increase executor memory for stateful operations
spark.conf.set("spark.executor.memory", "8g")

# Tune watermark for late data
.withWatermark("timestamp", "15 minutes")

# Increase trigger interval to reduce micro-batch overhead
.trigger(processingTime="30 seconds")

# Monitor lag and adjust parallelism
spark.conf.set("spark.sql.shuffle.partitions", 200)

# Recover from checkpoint corruption
# Delete checkpoint directory and restart (data loss possible)
# Or implement custom state recovery logic

データスキュー

症状：

• 一部のタスクがほかのタスクより大幅に時間がかかる
• パーティションサイズが不均衡
• Executor の OOM エラー

解決策：

# 1. Salting technique (add random prefix to keys)
from pyspark.sql.functions import concat, lit, rand

df_salted = df.withColumn("salted_key", concat(col("key"), lit("_"), (rand() * 10).cast("int")))
result = df_salted.groupBy("salted_key").agg(...)

# 2. Repartition by skewed column
df.repartition(200, "skewed_column")

# 3. Isolate skewed keys
skewed_keys = df.groupBy("key").count().filter(col("count") > threshold).select("key")
skewed_df = df.join(broadcast(skewed_keys), "key")
normal_df = df.join(broadcast(skewed_keys), "key", "left_anti")

# Process separately
skewed_result = process_with_salting(skewed_df)
normal_result = process_normally(normal_df)
final = skewed_result.union(normal_result)

Context7 コード統合

このスキルは、Apache Spark の公式リポジトリにある実際のコード例を統合します。EXAMPLES.md ファイル内のすべてのコードスニペットは、Context7 の Apache Spark ライブラリドキュメントから取得されており、本番環境ですぐに使えるパターンとベストプラクティスを提供します。

バージョンと互換性

• Apache Spark バージョン：3.x（2.4 以降と互換性あり）
• Python：3.7+
• Scala：2.12+
• Java：8+
• R：3.5+

参考資料

• 公式ドキュメント：https://spark.apache.org/docs/latest/
• API リファレンス：https://spark.apache.org/docs/latest/api.html
• GitHub リポジトリ：https://github.com/apache/spark
• Databricks ブログ：https://databricks.com/blog
• Context7 ライブラリ：/apache/spark

---

スキルのバージョン：1.0.0 最終更新日：2025年10月 スキルのカテゴリ：ビッグデータ、分散コンピューティング、データエンジニアリング、機械学習 Context7 統合：/apache/spark（ドキュメントのトークン数：8000）