Spark 4.0.0 は、Python 3.9+ で動作します。標準の CPython インタープリタを使用できるため、NumPy などの C ライブラリを使用できます。PyPy 7.3.6+ でも動作します。

Python の Spark アプリケーションは、実行時に Spark を含んでいる bin/spark-submit スクリプトで実行するか、または setup.py に以下のように含めることで実行できます。

    install_requires=[
        'pyspark==4.0.0'
    ]

pip で PySpark をインストールせずに Python で Spark アプリケーションを実行するには、Spark ディレクトリにある bin/spark-submit スクリプトを使用します。このスクリプトは Spark の Java/Scala ライブラリをロードし、クラスターにアプリケーションを送信できます。対話型 Python シェルを起動するには bin/pyspark を使用することもできます。

HDFS データにアクセスしたい場合は、HDFS のバージョンにリンクされた PySpark のビルドを使用する必要があります。事前ビルドされたパッケージ も Spark ホームページで一般的な HDFS バージョン向けに提供されています。

最後に、Spark クラスの一部をプログラムにインポートする必要があります。次の行を追加します。

from pyspark import SparkContext, SparkConf

PySpark は、ドライバーとワーカーの両方で同じマイナーバージョンの Python を必要とします。PATH のデフォルトの Python バージョンを使用します。PYSPARK_PYTHON を設定することで、使用したい Python のバージョンを指定できます。例：

$ PYSPARK_PYTHON=python3.8 bin/pyspark
$ PYSPARK_PYTHON=/path-to-your-pypy/pypy bin/spark-submit examples/src/main/python/pi.py

Spark 4.0.0 は、デフォルトで Scala 2.13 用にビルドおよび配布されています。(Spark は他の Scala バージョンでもビルドできます。) Scala でアプリケーションを記述するには、互換性のある Scala バージョン（例：2.13.X）を使用する必要があります。

Spark アプリケーションを記述するには、Spark への Maven 依存関係を追加する必要があります。Spark は Maven Central で入手できます。

groupId = org.apache.spark
artifactId = spark-core_2.13
version = 4.0.0

さらに、HDFS クラスターにアクセスしたい場合は、HDFS のバージョン用の hadoop-client への依存関係を追加する必要があります。

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

最後に、Spark クラスの一部をプログラムにインポートする必要があります。次の行を追加します。

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

(Spark 1.3.0 より前は、不可欠な暗黙の変換を有効にするために、明示的に import org.apache.spark.SparkContext._ をインポートする必要がありました。)

Spark 4.0.0 は、関数を簡潔に記述するための ラムダ式 をサポートしています。それ以外の場合は、org.apache.spark.api.java.function パッケージのクラスを使用できます。

注意：Spark 2.2.0 で Java 7 のサポートが削除されました。

Java で Spark アプリケーションを記述するには、Spark への依存関係を追加する必要があります。Spark は Maven Central で入手できます。

groupId = org.apache.spark
artifactId = spark-core_2.13
version = 4.0.0

さらに、HDFS クラスターにアクセスしたい場合は、HDFS のバージョン用の hadoop-client への依存関係を追加する必要があります。

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

最後に、Spark クラスの一部をプログラムにインポートする必要があります。次の行を追加します。

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.SparkConf;

Spark プログラムが最初に行うべきことは、SparkContext オブジェクトを作成することです。これは、Spark がクラスターにどのようにアクセスするかを伝えます。SparkContext を作成するには、まずアプリケーションに関する情報を含む SparkConf オブジェクトを構築する必要があります。

conf = SparkConf().setAppName(appName).setMaster(master)
sc = SparkContext(conf=conf)

Spark プログラムが最初に行うべきことは、SparkContext オブジェクトを作成することです。これは、Spark がクラスターにどのようにアクセスするかを伝えます。SparkContext を作成するには、まずアプリケーションに関する情報を含む SparkConf オブジェクトを構築する必要があります。

JVM ごとにアクティブな SparkContext は 1 つだけである必要があります。新しい SparkContext を作成する前に、アクティブな SparkContext を stop() する必要があります。

val conf = new SparkConf().setAppName(appName).setMaster(master)
new SparkContext(conf)

Spark プログラムが最初に行うべきことは、JavaSparkContext オブジェクトを作成することです。これは、Spark がクラスターにどのようにアクセスするかを伝えます。SparkContext を作成するには、まずアプリケーションに関する情報を含む SparkConf オブジェクトを構築する必要があります。

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
JavaSparkContext sc = new JavaSparkContext(conf);

PySpark シェルでは、sc という名前の特別なインタープリタ対応 SparkContext がすでに作成されています。独自の SparkContext を作成しても機能しません。--master 引数を使用してコンテキストが接続するマスターを設定でき、--py-files にカンマ区切りのリストを渡すことで、実行時のパスに Python .zip、.egg、または .py ファイルを追加できます。サードパーティの Python 依存関係については、Python パッケージ管理 を参照してください。--packages 引数に Maven 座標のカンマ区切りのリストを提供することで、シェルセッションに依存関係（例：Spark Packages）を追加することもできます。依存関係が存在する可能性のある追加のリポジトリ（例：Sonatype）は、--repositories 引数に渡すことができます。たとえば、4 つのコアで bin/pyspark を実行するには、次のようにします。

$ ./bin/pyspark --master "local[4]"

または、code.py を検索パスに追加するには (後で import code できるようにするため)、次のようにします。

$ ./bin/pyspark --master "local[4]" --py-files code.py

オプションの完全なリストについては、pyspark --help を実行してください。内部的には、pyspark はより一般的な spark-submit スクリプト を呼び出します。

IPython、拡張 Python インタープリタで PySpark シェルを起動することも可能です。PySpark は IPython 1.0.0 以降で動作します。IPython を使用するには、bin/pyspark を実行する際に PYSPARK_DRIVER_PYTHON 変数を ipython に設定します。

$ PYSPARK_DRIVER_PYTHON=ipython ./bin/pyspark

Jupyter Notebook (旧称 IPython Notebook) を使用するには、

$ PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=notebook ./bin/pyspark

PYSPARK_DRIVER_PYTHON_OPTS を設定することで、ipython または jupyter コマンドをカスタマイズできます。

Jupyter Notebook サーバーが起動したら、「Files」タブから新しいノートブックを作成できます。ノートブック内では、Jupyter Notebook から Spark を試す前に、%pylab inline というコマンドをノートブックの一部として入力できます。

Spark シェルでは、sc という名前の特別なインタープリタ対応 SparkContext がすでに作成されています。独自の SparkContext を作成しても機能しません。--master 引数を使用してコンテキストが接続するマスターを設定でき、--jars にカンマ区切りのリストを渡すことで、JAR をクラスパスに追加できます。--packages 引数に Maven 座標のカンマ区切りのリストを提供することで、シェルセッションに依存関係（例：Spark Packages）を追加することもできます。依存関係が存在する可能性のある追加のリポジトリ（例：Sonatype）は、--repositories 引数に渡すことができます。たとえば、4 つのコアで bin/spark-shell を実行するには、次のようにします。

$ ./bin/spark-shell --master "local[4]"

または、code.jar をクラスパスに追加するには、次のようにします。

$ ./bin/spark-shell --master "local[4]" --jars code.jar

Maven 座標を使用して依存関係を含めるには

$ ./bin/spark-shell --master "local[4]" --packages "org.example:example:0.1"

オプションの完全なリストについては、spark-shell --help を実行してください。内部的には、spark-shell はより一般的な spark-submit スクリプト を呼び出します。

並列化されたコレクションは、ドライバープログラムの既存のイテラブルまたはコレクションに対して SparkContext の parallelize メソッドを呼び出すことによって作成されます。コレクションの要素はコピーされ、並列に操作できる分散データセットを形成します。たとえば、1 から 5 までの数値を保持する並列化されたコレクションを作成する方法は次のとおりです。

data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)

作成後、分散データセット (distData) は並列に操作できます。たとえば、リストの要素を合計するために distData.reduce(lambda a, b: a + b) を呼び出すことができます。分散データセットの操作については後で説明します。

並列化されたコレクションは、ドライバープログラムの既存のコレクション (Scala Seq) に対して SparkContext の parallelize メソッドを呼び出すことによって作成されます。コレクションの要素はコピーされ、並列に操作できる分散データセットを形成します。たとえば、1 から 5 までの数値を保持する並列化されたコレクションを作成する方法は次のとおりです。

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

作成後、分散データセット (distData) は並列に操作できます。たとえば、配列の要素を合計するために distData.reduce((a, b) => a + b) を呼び出すことができます。分散データセットの操作については後で説明します。

並列化されたコレクションは、ドライバープログラムの既存の Collection に対して JavaSparkContext の parallelize メソッドを呼び出すことによって作成されます。コレクションの要素はコピーされ、並列に操作できる分散データセットを形成します。たとえば、1 から 5 までの数値を保持する並列化されたコレクションを作成する方法は次のとおりです。

List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
JavaRDD<Integer> distData = sc.parallelize(data);

作成後、分散データセット (distData) は並列に操作できます。たとえば、リストの要素を合計するために distData.reduce((a, b) -> a + b) を呼び出すことができます。分散データセットの操作については後で説明します。

PySpark は、ローカルファイルシステム、HDFS、Cassandra、HBase、Amazon S3 など、Hadoop がサポートするあらゆるストレージソースから分散データセットを作成できます。Spark はテキストファイル、SequenceFiles、およびその他の Hadoop InputFormat をサポートしています。

テキストファイル RDD は、SparkContext の textFile メソッドを使用して作成できます。このメソッドは、ファイルの URI (ローカルパスまたは hdfs://、s3a:// などの URI) を受け取り、それを行のコレクションとして読み取ります。例：

>>> distFile = sc.textFile("data.txt")

作成後、distFile はデータセット操作で処理できます。たとえば、map および reduce 操作を使用して、すべての行のサイズを合計できます。例：distFile.map(lambda s: len(s)).reduce(lambda a, b: a + b)。

Spark でファイルを読み取ることに関する注意点

• ローカルファイルシステム上のパスを使用する場合、そのファイルはワーカーノードでも同じパスでアクセス可能である必要があります。ファイルをすべてのワーカーにコピーするか、ネットワークでマウントされた共有ファイルシステムを使用してください。

• textFile を含む Spark のすべてのファイルベース入力メソッドは、ディレクトリ、圧縮ファイル、およびワイルドカードもサポートしています。たとえば、textFile("/my/directory")、textFile("/my/directory/*.txt")、および textFile("/my/directory/*.gz") を使用できます。

• textFile メソッドは、ファイルのパーティション数を制御するためのオプションの 2 番目の引数も取ります。デフォルトでは、Spark はファイルごとに 1 つのパーティションを作成します (HDFS ではデフォルトで 128MB)。ただし、より大きな値を渡して、より多くのパーティションを要求することもできます。注意：ブロック数より少ないパーティションを持つことはできません。

テキストファイル以外に、Spark の Python API は他のいくつかのデータ形式もサポートしています。

• SparkContext.wholeTextFiles は、複数の小さなテキストファイルを含むディレクトリを読み込み、各ファイルを (ファイル名、コンテンツ) のペアとして返します。これは、textFile が各ファイルの各行を 1 つのレコードとして返すのとは対照的です。

• RDD.saveAsPickleFile および SparkContext.pickleFile は、ピクル化された Python オブジェクトの単純な形式で RDD を保存することをサポートしています。ピクルシリアライズではバッチ処理が使用され、デフォルトのバッチサイズは 10 です。

• SequenceFile および Hadoop Input/Output Formats

注意 この機能は現在 Experimental としてマークされており、高度なユーザーを対象としています。将来的に Spark SQL に基づいた読み取り/書き込みサポートに置き換えられる可能性があり、その場合は Spark SQL が推奨されるアプローチです。

Writable サポート

PySpark SequenceFile サポートは、Java 内のキー・バリューペアの RDD をロードし、Writable を基本 Java 型に変換し、結果の Java オブジェクトを pickle を使用してピクル化します。キー・バリューペアの RDD を SequenceFile に保存する場合、PySpark は逆を行います。Python オブジェクトを Java オブジェクトにアンピクル化し、それを Writable に変換します。以下の Writable は自動的に変換されます。

配列は標準では扱われません。ユーザーは、読み取りまたは書き込み時にカスタム ArrayWritable サブタイプを指定する必要があります。書き込み時には、ユーザーは配列をカスタム ArrayWritable サブタイプに変換するカスタムコンバータも指定する必要があります。読み取り時には、デフォルトのコンバータはカスタム ArrayWritable サブタイプを Java Object[] に変換し、それが Python タプルにピクル化されます。プリミティブ型の配列の Python array.array を取得するには、ユーザーがカスタムコンバータを指定する必要があります。

SequenceFiles の保存と読み込み

テキストファイルと同様に、SequenceFile はパスを指定して保存および読み込むことができます。キーと値のクラスを指定できますが、標準の Writable では不要です。

>>> rdd = sc.parallelize(range(1, 4)).map(lambda x: (x, "a" * x))
>>> rdd.saveAsSequenceFile("path/to/file")
>>> sorted(sc.sequenceFile("path/to/file").collect())
[(1, u'a'), (2, u'aa'), (3, u'aaa')]

その他の Hadoop Input/Output Format の保存と読み込み

PySpark は、新しい Hadoop MapReduce API および古い Hadoop MapReduce API の両方について、任意の Hadoop InputFormat を読み取ったり、任意の Hadoop OutputFormat を書き込んだりすることもできます。必要に応じて、Python dict として Hadoop 設定を渡すことができます。Elasticsearch ESInputFormat を使用した例を次に示します。

$ ./bin/pyspark --jars /path/to/elasticsearch-hadoop.jar
>>> conf = {"es.resource" : "index/type"}  # assume Elasticsearch is running on localhost defaults
>>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",
                             "org.apache.hadoop.io.NullWritable",
                             "org.elasticsearch.hadoop.mr.LinkedMapWritable",
                             conf=conf)
>>> rdd.first()  # the result is a MapWritable that is converted to a Python dict
(u'Elasticsearch ID',
 {u'field1': True,
  u'field2': u'Some Text',
  u'field3': 12345})

注意：InputFormat が Hadoop 設定および/または入力パスにのみ依存しており、キーと値のクラスが上記の表に従って簡単に変換できる場合、このアプローチはこのようなケースでうまく機能するはずです。

カスタムシリアライズされたバイナリデータ (Cassandra/HBase からのデータ読み込みなど) がある場合は、まず Scala/Java 側で pickle の pickler で処理できるものにデータを変換する必要があります。Converter トレイトが提供されています。このトレイトを拡張し、convert メソッドに変換コードを実装するだけです。このクラスと InputFormat にアクセスするために必要な依存関係が Spark ジョブ jar にパッケージ化され、PySpark クラスパスに含まれていることを確認してください。

Python の例 と Converter の例 を参照して、Cassandra / HBase InputFormat および OutputFormat をカスタムコンバータと共に使用する例を確認してください。

Spark は、ローカルファイルシステム、HDFS、Cassandra、HBase、Amazon S3 など、Hadoop がサポートするあらゆるストレージソースから分散データセットを作成できます。Spark はテキストファイル、SequenceFiles、およびその他の Hadoop InputFormat をサポートしています。

テキストファイル RDD は、SparkContext の textFile メソッドを使用して作成できます。このメソッドは、ファイルの URI (ローカルパスまたは hdfs://、s3a:// などの URI) を受け取り、それを行のコレクションとして読み取ります。例：

scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

作成後、distFile はデータセット操作で処理できます。たとえば、map および reduce 操作を使用して、すべての行のサイズを合計できます。例：distFile.map(s => s.length).reduce((a, b) => a + b)。

Spark でファイルを読み取ることに関する注意点

• ローカルファイルシステム上のパスを使用する場合、そのファイルはワーカーノードでも同じパスでアクセス可能である必要があります。ファイルをすべてのワーカーにコピーするか、ネットワークでマウントされた共有ファイルシステムを使用してください。

• textFile を含む Spark のすべてのファイルベース入力メソッドは、ディレクトリ、圧縮ファイル、およびワイルドカードもサポートしています。たとえば、textFile("/my/directory")、textFile("/my/directory/*.txt")、および textFile("/my/directory/*.gz") を使用できます。複数のファイルが読み取られる場合、パーティションの順序はファイルシステムから返される順序によって決まります。たとえば、パスによるファイルの辞書順ソートに従う場合もあれば、そうでない場合もあります。パーティション内では、要素は基になるファイル内の順序に従って並べられます。

• textFile メソッドは、ファイルのパーティション数を制御するためのオプションの 2 番目の引数も取ります。デフォルトでは、Spark はファイルごとに 1 つのパーティションを作成します (HDFS ではデフォルトで 128MB)。ただし、より大きな値を渡して、より多くのパーティションを要求することもできます。注意：ブロック数より少ないパーティションを持つことはできません。

テキストファイル以外に、Spark の Scala API は他のいくつかのデータ形式もサポートしています。

• SparkContext.wholeTextFiles は、複数の小さなテキストファイルを含むディレクトリを読み込み、各ファイルを (ファイル名、コンテンツ) のペアとして返します。これは、textFile が各ファイルの各行を 1 つのレコードとして返すのとは対照的です。パーティショニングはデータローカリティによって決定されるため、場合によってはパーティションが少なすぎる可能性があります。そのような場合、wholeTextFiles は最小パーティション数を制御するためのオプションの 2 番目の引数を提供します。

• SequenceFiles の場合、SparkContext の sequenceFile[K, V] メソッドを使用します。ここで K と V はファイル内のキーと値の型です。これらは、Hadoop の Writable インターフェースのサブクラスである必要があります。たとえば、IntWritable や Text です。さらに、Spark は一般的な Writable のいくつかに対してネイティブ型を指定できます。たとえば、sequenceFile[Int, String] は IntWritable と Text を自動的に読み取ります。

• その他の Hadoop InputFormats については、SparkContext.hadoopRDD メソッドを使用できます。これは、任意の JobConf と入力フォーマットクラス、キークラス、値クラスを取ります。これらは、入力ソースを持つ Hadoop ジョブの場合と同様に設定します。また、「新しい」MapReduce API (org.apache.hadoop.mapreduce) に基づく InputFormats の場合は SparkContext.newAPIHadoopRDD を使用することもできます。

• RDD.saveAsObjectFile および SparkContext.objectFile は、Java オブジェクトのシリアライズされた単純な形式で RDD を保存することをサポートしています。これは Avro のような専門化された形式ほど効率的ではありませんが、任意の RDD を簡単に保存できます。

Spark は、ローカルファイルシステム、HDFS、Cassandra、HBase、Amazon S3 など、Hadoop がサポートするあらゆるストレージソースから分散データセットを作成できます。Spark はテキストファイル、SequenceFiles、およびその他の Hadoop InputFormat をサポートしています。

テキストファイル RDD は、SparkContext の textFile メソッドを使用して作成できます。このメソッドは、ファイルの URI (ローカルパスまたは hdfs://、s3a:// などの URI) を受け取り、それを行のコレクションとして読み取ります。例：

JavaRDD<String> distFile = sc.textFile("data.txt");

作成後、distFile はデータセット操作で処理できます。たとえば、map および reduce 操作を使用して、すべての行のサイズを合計できます。例：distFile.map(s -> s.length()).reduce((a, b) -> a + b)。

Spark でファイルを読み取ることに関する注意点

• ローカルファイルシステム上のパスを使用する場合、そのファイルはワーカーノードでも同じパスでアクセス可能である必要があります。ファイルをすべてのワーカーにコピーするか、ネットワークでマウントされた共有ファイルシステムを使用してください。

• textFile を含む Spark のすべてのファイルベース入力メソッドは、ディレクトリ、圧縮ファイル、およびワイルドカードもサポートしています。たとえば、textFile("/my/directory")、textFile("/my/directory/*.txt")、および textFile("/my/directory/*.gz") を使用できます。

• textFile メソッドは、ファイルのパーティション数を制御するためのオプションの 2 番目の引数も取ります。デフォルトでは、Spark はファイルごとに 1 つのパーティションを作成します (HDFS ではデフォルトで 128MB)。ただし、より大きな値を渡して、より多くのパーティションを要求することもできます。注意：ブロック数より少ないパーティションを持つことはできません。

テキストファイル以外に、Spark の Java API は他のいくつかのデータ形式もサポートしています。

• JavaSparkContext.wholeTextFiles は、複数の小さなテキストファイルを含むディレクトリを読み込み、各ファイルを (ファイル名、コンテンツ) のペアとして返します。これは、textFile が各ファイルの各行を 1 つのレコードとして返すのとは対照的です。

• SequenceFiles の場合、SparkContext の sequenceFile[K, V] メソッドを使用します。ここで K と V はファイル内のキーと値の型です。これらは、Hadoop の Writable インターフェースのサブクラスである必要があります。たとえば、IntWritable や Text です。

• その他の Hadoop InputFormats については、JavaSparkContext.hadoopRDD メソッドを使用できます。これは、任意の JobConf と入力フォーマットクラス、キークラス、値クラスを取ります。これらは、入力ソースを持つ Hadoop ジョブの場合と同様に設定します。また、「新しい」MapReduce API (org.apache.hadoop.mapreduce) に基づく InputFormats の場合は JavaSparkContext.newAPIHadoopRDD を使用することもできます。

• JavaRDD.saveAsObjectFile および JavaSparkContext.objectFile は、Java オブジェクトのシリアライズされた単純な形式で RDD を保存することをサポートしています。これは Avro のような専門化された形式ほど効率的ではありませんが、任意の RDD を簡単に保存できます。

RDD の基本を説明するために、次の簡単なプログラムを検討してください。

lines = sc.textFile("data.txt")
lineLengths = lines.map(lambda s: len(s))
totalLength = lineLengths.reduce(lambda a, b: a + b)

最初の行は、外部ファイルから基になる RDD を定義します。このデータセットはメモリにロードされたり、その他の方法で処理されたりしません。lines は単にファイルへのポインタです。2 番目の行は、map 変換の結果として lineLengths を定義します。ここでも、遅延のため、lineLengths はすぐに計算されません。最後に、アクションである reduce を実行します。この時点で Spark は、計算を別々のマシンで実行されるタスクに分割し、各マシンはマップとその部分のローカル削減の両方を実行し、その結果のみをドライバープログラムに返します。

後で lineLengths を再度使用したい場合は、次を追加できます。

lineLengths.persist()

reduce の前に、これにより lineLengths が初めて計算された後にメモリに保存されます。

RDD の基本を説明するために、次の簡単なプログラムを検討してください。

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

最初の行は、外部ファイルから基になる RDD を定義します。このデータセットはメモリにロードされたり、その他の方法で処理されたりしません。lines は単にファイルへのポインタです。2 番目の行は、map 変換の結果として lineLengths を定義します。ここでも、遅延のため、lineLengths はすぐに計算されません。最後に、アクションである reduce を実行します。この時点で Spark は、計算を別々のマシンで実行されるタスクに分割し、各マシンはマップとその部分のローカル削減の両方を実行し、その結果のみをドライバープログラムに返します。

後で lineLengths を再度使用したい場合は、次を追加できます。

lineLengths.persist()

reduce の前に、これにより lineLengths が初めて計算された後にメモリに保存されます。

RDD の基本を説明するために、次の簡単なプログラムを検討してください。

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a + b);

最初の行は、外部ファイルから基になる RDD を定義します。このデータセットはメモリにロードされたり、その他の方法で処理されたりしません。lines は単にファイルへのポインタです。2 番目の行は、map 変換の結果として lineLengths を定義します。ここでも、遅延のため、lineLengths はすぐに計算されません。最後に、アクションである reduce を実行します。この時点で Spark は、計算を別々のマシンで実行されるタスクに分割し、各マシンはマップとその部分のローカル削減の両方を実行し、その結果のみをドライバープログラムに返します。

後で lineLengths を再度使用したい場合は、次を追加できます。

lineLengths.persist(StorageLevel.MEMORY_ONLY());

reduce の前に、これにより lineLengths が初めて計算された後にメモリに保存されます。

Spark の API は、クラスターで実行するためにドライバープログラムに関数を渡すことに大きく依存しています。これを行うための 3 つの推奨される方法があります。

• ラムダ式: 式として記述できる簡単な関数の場合。(ラムダは複数ステートメントの関数や値を返さないステートメントはサポートしません。)
• Spark を呼び出す関数内のローカル def: より長いコードの場合。
• モジュールのトップレベル関数。

たとえば、lambda ではサポートできないより長い関数を渡すには、次のコードを検討してください。

"""MyScript.py"""
if __name__ == "__main__":
    def myFunc(s):
        words = s.split(" ")
        return len(words)

    sc = SparkContext(...)
    sc.textFile("file.txt").map(myFunc)

メソッドへの参照をクラスインスタンス (シングルトンオブジェクトとは対照的) に渡すことも可能ですが、そのメソッドを含むオブジェクトも一緒に送信する必要があります。たとえば、次を検討してください。

class MyClass(object):
    def func(self, s):
        return s
    def doStuff(self, rdd):
        return rdd.map(self.func)

ここで、new MyClass を作成し、それに対して doStuff を呼び出すと、その中の map は その MyClass インスタンスの func メソッドを参照するため、オブジェクト全体をクラスターに送信する必要があります。

同様に、外側のオブジェクトのフィールドにアクセスすると、オブジェクト全体が参照されます。

class MyClass(object):
    def __init__(self):
        self.field = "Hello"
    def doStuff(self, rdd):
        return rdd.map(lambda s: self.field + s)

この問題を回避するには、最も簡単な方法は、外部からアクセスするのではなく、field をローカル変数にコピーすることです。

def doStuff(self, rdd):
    field = self.field
    return rdd.map(lambda s: field + s)

Spark の API は、クラスターで実行するためにドライバープログラムに関数を渡すことに大きく依存しています。これを行うための 2 つの推奨される方法があります。

• 匿名関数構文: 短いコード片に使用できます。
• グローバルシングルトンオブジェクト内の静的メソッド。たとえば、MyFunctions オブジェクトを定義してから、MyFunctions.func1 を次のように渡すことができます。

object MyFunctions {
  def func1(s: String): String = { ... }
}

myRdd.map(MyFunctions.func1)

メソッドへの参照をクラスインスタンス (シングルトンオブジェクトとは対照的) に渡すことも可能ですが、そのメソッドを含むオブジェクトも一緒に送信する必要があります。たとえば、次を検討してください。

class MyClass {
  def func1(s: String): String = { ... }
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}

ここで、新しい MyClass インスタンスを作成し、それに対して doStuff を呼び出すと、その中の map は その MyClass インスタンスの func1 メソッドを参照するため、オブジェクト全体をクラスターに送信する必要があります。これは、rdd.map(x => this.func1(x)) と書くのと似ています。

同様に、外側のオブジェクトのフィールドにアクセスすると、オブジェクト全体が参照されます。

class MyClass {
  val field = "Hello"
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}

は、rdd.map(x => this.field + x) と書くのと同じであり、これは this のすべてを参照します。この問題を回避するには、外部からアクセスするのではなく、field をローカル変数にコピーするのが最も簡単な方法です。

def doStuff(rdd: RDD[String]): RDD[String] = {
  val field_ = this.field
  rdd.map(x => field_ + x)
}

Spark の API は、クラスターで実行するためにドライバープログラムに関数を渡すことに大きく依存しています。Java では、関数は org.apache.spark.api.java.function パッケージのインターフェースを実装するクラスによって表されます。これらの関数を作成するには 2 つの方法があります。

• 独自のクラスで Function インターフェースを実装します (匿名内部クラスまたは名前付きクラスのいずれか)。そして、そのインスタンスを Spark に渡します。
• ラムダ式 を使用して、実装を簡潔に定義します。

このガイドの多くは簡潔さのためにラムダ構文を使用していますが、すべての API を長形式で簡単に使用できます。たとえば、上記のコードを次のように記述できます。

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
  public Integer call(String s) { return s.length(); }
});
int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
  public Integer call(Integer a, Integer b) { return a + b; }
});

または、関数をインラインで記述するのが煩雑な場合。

class GetLength implements Function<String, Integer> {
  public Integer call(String s) { return s.length(); }
}
class Sum implements Function2<Integer, Integer, Integer> {
  public Integer call(Integer a, Integer b) { return a + b; }
}

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new GetLength());
int totalLength = lineLengths.reduce(new Sum());

Java の匿名内部クラスは、final としてマークされている限り、エンクロージングスコープの変数にアクセスすることもできることに注意してください。Spark は、他の言語と同様に、これらの変数のコピーを各ワーカーノードに送信します。

Spark のほとんどの操作は任意のオブジェクト型の RDD で動作しますが、いくつかの特別な操作はキー・バリューペアの RDD でのみ利用可能です。最も一般的なものは、キーによる要素のグループ化や集約などの分散 "シャッフル" 操作です。

Python では、これらの操作は (1, 2) のような組み込み Python タプルを含む RDD で機能します。これらのタプルを作成し、目的の操作を呼び出すだけです。

たとえば、次のコードはキー・バリューペアの reduceByKey 操作を使用して、ファイル内の各行が何回出現するかをカウントします。

lines = sc.textFile("data.txt")
pairs = lines.map(lambda s: (s, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)

たとえば、counts.sortByKey() を使用してペアをアルファベット順にソートし、最後に counts.collect() を使用してオブジェクトのリストとしてドライバープログラムに戻すこともできます。

Spark のほとんどの操作は任意のオブジェクト型の RDD で動作しますが、いくつかの特別な操作はキー・バリューペアの RDD でのみ利用可能です。最も一般的なものは、キーによる要素のグループ化や集約などの分散 "シャッフル" 操作です。

Scala では、これらの操作は Tuple2 オブジェクト (言語の組み込みタプル、単に (a, b) と記述して作成) を含む RDD で自動的に利用可能になります。キー・バリューペア操作は、PairRDDFunctions クラスで利用可能であり、これはタプルの RDD を自動的にラップします。

たとえば、次のコードはキー・バリューペアの reduceByKey 操作を使用して、ファイル内の各行が何回出現するかをカウントします。

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

たとえば、counts.sortByKey() を使用してペアをアルファベット順にソートし、最後に counts.collect() を使用してオブジェクトの配列としてドライバープログラムに戻すこともできます。

注意: キー・バリューペア操作でカスタムオブジェクトをキーとして使用する場合、カスタム equals() メソッドに一致する hashCode() メソッドが伴っていることを確認する必要があります。詳細については、Object.hashCode() ドキュメント に概説されている契約を参照してください。

Spark のほとんどの操作は任意のオブジェクト型の RDD で動作しますが、いくつかの特別な操作はキー・バリューペアの RDD でのみ利用可能です。最も一般的なものは、キーによる要素のグループ化や集約などの分散 "シャッフル" 操作です。

Java では、キー・バリューペアは Scala 標準ライブラリの scala.Tuple2 クラスを使用して表現されます。タプルを作成するには new Tuple2(a, b) を呼び出すだけで、後で tuple._1() と tuple._2() でフィールドにアクセスできます。

キー・バリューペアの RDD は、JavaPairRDD クラスによって表されます。mapToPair や flatMapToPair のような map 操作の特別なバージョンを使用して、JavaRDD から JavaPairRDD を構築できます。JavaPairRDD には、標準の RDD 関数と特別なキー・バリュー関数があります。

たとえば、次のコードはキー・バリューペアの reduceByKey 操作を使用して、ファイル内の各行が何回出現するかをカウントします。

JavaRDD<String> lines = sc.textFile("data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);

たとえば、counts.sortByKey() を使用してペアをアルファベット順にソートし、最後に counts.collect() を使用してオブジェクトの配列としてドライバープログラムに戻すこともできます。

注意: キー・バリューペア操作でカスタムオブジェクトをキーとして使用する場合、カスタム equals() メソッドに一致する hashCode() メソッドが伴っていることを確認する必要があります。詳細については、Object.hashCode() ドキュメント に概説されている契約を参照してください。

アキュムレータは、初期値 v から SparkContext.accumulator(v) を呼び出すことによって作成されます。クラスターで実行されるタスクは、add メソッドまたは += 演算子を使用してそれに加算できます。ただし、その値は読み取れません。ドライバープログラムのみが、value メソッドを使用してアキュムレータの値にアクセスできます。

次のコードは、配列の要素を合計するためにアキュムレータが使用されていることを示しています。

>>> accum = sc.accumulator(0)
>>> accum
Accumulator<id=0, value=0>

>>> sc.parallelize([1, 2, 3, 4]).foreach(lambda x: accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

>>> accum.value
10

このコードは Int 型のアキュムレータの組み込みサポートを使用しましたが、プログラマは AccumulatorParam をサブクラス化することによって独自の型を作成することもできます。AccumulatorParam インターフェースには 2 つのメソッドがあります。データ型の「ゼロ値」を提供する zero と、2 つの値を結合する addInPlace です。たとえば、数学ベクトルを表す Vector クラスがある場合、次のように記述できます。

class VectorAccumulatorParam(AccumulatorParam):
    def zero(self, initialValue):
        return Vector.zeros(initialValue.size)

    def addInPlace(self, v1, v2):
        v1 += v2
        return v1

# Then, create an Accumulator of this type:
vecAccum = sc.accumulator(Vector(...), VectorAccumulatorParam())

数値アキュムレータは、SparkContext.longAccumulator() または SparkContext.doubleAccumulator() を呼び出すことで作成できます。これは、それぞれ Long 型または Double 型の値を集計します。クラスターで実行されるタスクは、add メソッドを使用してそれに加算できます。ただし、その値は読み取れません。ドライバープログラムのみが、value メソッドを使用してアキュムレータの値にアクセスできます。

次のコードは、配列の要素を合計するためにアキュムレータが使用されていることを示しています。

scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
res2: Long = 10

このコードでは、組み込みのLong型のアキュムレータサポートを使用しましたが、プログラマはAccumulatorV2をサブクラス化することで独自の型を作成することもできます。AccumulatorV2抽象クラスには、オーバーライドする必要があるいくつかのメソッドがあります。アキュムレータをゼロにリセットするためのreset、アキュムレータに別の値を追加するためのadd、同じ型の別のアキュムレータをこのアキュムレータにマージするためのmergeです。オーバーライドする必要があるその他のメソッドは、APIドキュメントに含まれています。たとえば、数学的なベクトルを表すMyVectorクラスがあると仮定すると、次のように記述できます。

class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {

  private val myVector: MyVector = MyVector.createZeroVector

  def reset(): Unit = {
    myVector.reset()
  }

  def add(v: MyVector): Unit = {
    myVector.add(v)
  }
  ...
}

// Then, create an Accumulator of this type:
val myVectorAcc = new VectorAccumulatorV2
// Then, register it into spark context:
sc.register(myVectorAcc, "MyVectorAcc1")

注意: プログラマが独自のAccumulatorV2型を定義する場合、結果の型は追加される要素の型と異なる場合があります。

数値アキュムレータは、SparkContext.longAccumulator() または SparkContext.doubleAccumulator() を呼び出すことで作成できます。これは、それぞれ Long 型または Double 型の値を集計します。クラスターで実行されるタスクは、add メソッドを使用してそれに加算できます。ただし、その値は読み取れません。ドライバープログラムのみが、value メソッドを使用してアキュムレータの値にアクセスできます。

次のコードは、配列の要素を合計するためにアキュムレータが使用されていることを示しています。

LongAccumulator accum = jsc.sc().longAccumulator();

sc.parallelize(Arrays.asList(1, 2, 3, 4)).foreach(x -> accum.add(x));
// ...
// 10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

accum.value();
// returns 10

このコードでは、組み込みのLong型のアキュムレータサポートを使用しましたが、プログラマはAccumulatorV2をサブクラス化することで独自の型を作成することもできます。AccumulatorV2抽象クラスには、オーバーライドする必要があるいくつかのメソッドがあります。アキュムレータをゼロにリセットするためのreset、アキュムレータに別の値を追加するためのadd、同じ型の別のアキュムレータをこのアキュムレータにマージするためのmergeです。オーバーライドする必要があるその他のメソッドは、APIドキュメントに含まれています。たとえば、数学的なベクトルを表すMyVectorクラスがあると仮定すると、次のように記述できます。

class VectorAccumulatorV2 implements AccumulatorV2<MyVector, MyVector> {

  private MyVector myVector = MyVector.createZeroVector();

  public void reset() {
    myVector.reset();
  }

  public void add(MyVector v) {
    myVector.add(v);
  }
  ...
}

// Then, create an Accumulator of this type:
VectorAccumulatorV2 myVectorAcc = new VectorAccumulatorV2();
// Then, register it into spark context:
jsc.sc().register(myVectorAcc, "MyVectorAcc1");

注意: プログラマが独自のAccumulatorV2型を定義する場合、結果の型は追加される要素の型と異なる場合があります。

警告: Sparkタスクが終了すると、Sparkはそのタスクでの累積された更新をアキュムレータにマージしようとします。失敗した場合、Sparkは失敗を無視し、タスクを成功とマークして他のタスクの実行を続行します。したがって、バグのあるアキュムレータはSparkジョブに影響を与えませんが、Sparkジョブが成功しても正しく更新されない可能性があります。