spark

spark

目的

Apache Spark は、インメモリーコンピューティングを使用して大規模なデータセットを処理する、高速な分散処理フレームワークです。クラスター上で、効率的なバッチ処理、リアルタイム分析、機械学習、グラフ処理を実現します。

使用する場面

単一マシンのメモリーに収まらないデータセットを処理する場合に Spark を使用します。たとえば、テラバイト規模のログデータを分析したり、ETL ジョブを実行したりする場合です。機械学習アルゴリズムなど、高速な反復計算が必要なシナリオや、Hadoop などのビッグデータエコシステムと統合する場合にも適しています。Pandas などのよりシンプルなツールで十分な小規模タスクでは使用を避けてください。

主な機能

• インメモリーキャッシュにより、たとえば persist(StorageLevel.MEMORY_ONLY) を使って反復アルゴリズムを高速化します。
• RDD（Resilient Distributed Datasets）を使用したフォールトトレラントな分散コンピューティングにより、自動復旧を実現します。
• Scala、Python、Java、R など複数の言語をサポートします。たとえば、from pyspark.sql import SparkSession を使ってデータフレームを扱うには PySpark を使用します。
• 組み込みライブラリとして、構造化データのクエリに使う Spark SQL、機械学習に使う MLlib、グラフ処理に使う GraphX、リアルタイムデータに使う Structured Streaming を備えています。
• 数千のノードまでスケーリングでき、YARN または Kubernetes による動的なリソース割り当てが可能です。

使用パターン

Spark でデータを処理するには、まずコード内で SparkSession を作成します。バッチジョブの場合は spark-submit で送信し、対話形式で作業する場合は Spark のシェルを使用します。クラスターのマスターモードでは "yarn" など、必ずマスター URL を指定してください。ファイルまたはデータベースからデータソースを読み込み、DataFrame で変換して出力を書き込みます。ストリーミングでは Structured Streaming を使って、Kafka トピックをリアルタイムで処理します。

例 1：PySpark での単語数カウント。

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WordCount").getOrCreate()
words = spark.read.text("hdfs://path/to/file.txt").rdd.flatMap(lambda x: x[0].split(" "))
counts = words.map(lambda x: (x, 1)).reduceByKey(lambda a, b: a + b)
counts.saveAsTextFile("hdfs://output/path")

例 2：CSV から Parquet への ETL ジョブ。

spark = SparkSession.builder.master("local[*]").appName("ETL").getOrCreate()
df = spark.read.format("csv").option("header", "true").load("s3://bucket/data.csv")
df.write.format("parquet").mode("overwrite").save("hdfs://processed/data.parquet")

これらを実行するには、次のコマンドを使用します：spark-submit --master yarn --executor-memory 4g your_script.py。

主なコマンド／API

アプリケーションの実行には spark-submit を使用します：spark-submit --class MainClass --master yarn --deploy-mode cluster --driver-memory 2g your.jar arg1 arg2。対話型セッションの場合は、pyspark または spark-shell を実行します。主な API 呼び出しには、次のような SparkSession の作成があります：SparkSession.builder().appName("App").master("local").getOrCreate()。次のようにデータを読み込みます：spark.read.csv("path", header=True, inferSchema=True)。DataFrame API を使ってデータを変換します。たとえば、df.filter(df['age'] > 30).groupBy('department').count() です。設定には SparkConf を使用します：conf = SparkConf().set("spark.executor.cores", "2")。クラスターにアクセスするには、マスターノード用の $SPARK_MASTER_URL などの環境変数を設定します。

統合に関する注意点

Spark を Hadoop と統合するには、$HADOOP_CONF_DIR 環境変数に Hadoop の設定パスを指定し、マスターとして YARN を使用します。Kafka の場合は、spark-submit に --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2 を指定してコネクターを追加し、spark.readStream.format("kafka").option("kafka.bootstrap.servers", "host:port").load() でストリームを読み込みます。JDBC を使ってデータベースに接続します：df.write.jdbc(url="jdbc:postgresql://host/db", table="table", mode="append")。この場合、クラスパスに JDBC ドライバーが必要です。設定には spark-defaults.conf などの設定ファイルを使用します。たとえば、spark.sql.shuffle.partitions 200 のように指定します。

エラー処理

OutOfMemory エラーが発生した場合は、spark-submit に --driver-memory 4g --executor-memory 8g を追加してメモリーを増やします。タスクが失敗した場合は、Spark UI（http://driver-host:4040）でログを確認し、spark.task.maxFailures 設定で再試行の上限を指定します。一般的なシリアライズの問題（例：NotSerializableException）は、Java で Serializable を実装するなどしてクラスをシリアライズ可能にすると解決できます。データの偏りがある場合は、df.repartition(100).write... を使ってデータを再パーティション化します。API 呼び出しは必ず try-except で囲みます。たとえば、try: df = spark.read.csv("path") except Exception as e: print(e) のようにします。

グラフの関係

接続先：データエンジニアリングクラスター（例：ストレージ用の hadoop、オーケストレーション用の airflow）。関連タグ：big-data、distributed-computing。リンク：ストリーミングには kafka と統合し、入出力には hadoop のファイルシステムを使用します。