Spark 3.5.1はPython 3.8以降で動作します。標準のCPythonインタープリターを使用できるため、NumPyのようなCライブラリを使用できます。また、PyPy 7.3.6以降でも動作します。

PythonでのSparkアプリケーションは、ランタイムにSparkを含むbin/spark-submitスクリプトを使用して実行するか、setup.pyに含めることで実行できます。

    install_requires=[
        'pyspark=={site.SPARK_VERSION}'
    ]

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

HDFSデータにアクセスする場合は、HDFSのバージョンにリンクするPySparkのビルドを使用する必要があります。プリビルドパッケージは、一般的なHDFSバージョン向けにSparkのホームページでも入手できます。

最後に、いくつかの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 3.5.1は、デフォルトでScala 2.12と連携するようにビルドおよび配布されています。（Sparkは、他のバージョンのScalaと連携するようにビルドすることもできます。）Scalaでアプリケーションを作成するには、互換性のあるScalaバージョン（例：2.12.X）を使用する必要があります。

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

groupId = org.apache.spark
artifactId = spark-core_2.12
version = 3.5.1

さらに、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 3.5.1は、関数を簡潔に記述するためのラムダ式をサポートしています。そうでない場合は、org.apache.spark.api.java.functionパッケージのクラスを使用できます。

Java 7のサポートは、Spark 2.2.0で削除されたことに注意してください。

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

groupId = org.apache.spark
artifactId = spark-core_2.12
version = 3.5.1

さらに、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がクラスタにどのようにアクセスするかをSparkに伝えます。SparkContextを作成するには、最初にアプリケーションに関する情報を含むSparkConfオブジェクトをビルドする必要があります。

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

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

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

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

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

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

PySparkシェルでは、特別なインタープリター対応のSparkContextが、scという変数に既に作成されています。独自のSparkContextを作成しても機能しません。コンテキストが接続するマスターは、--master引数を使用して設定できます。また、コンマ区切りのリストを--py-filesに渡すことで、Pythonの.zip、.egg、または.pyファイルをランタイムパスに追加できます。サードパーティのPython依存関係については、Pythonパッケージ管理を参照してください。また、コンマ区切りのMaven座標のリストを--packages引数に指定することにより、シェルセッションに依存関係（例：Sparkパッケージ）を追加することもできます。依存関係が存在する可能性のある追加のリポジトリ（例：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ノートブック（以前はIPythonノートブックとして知られていました）を使用するには、

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

ipython または jupyter コマンドをカスタマイズするには、PYSPARK_DRIVER_PYTHON_OPTS を設定します。

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

Sparkシェルでは、特別なインタープリター対応のSparkContextがすでに変数 sc に作成されています。独自のSparkContextを作成しても機能しません。コンテキストが接続するマスターは、--master 引数を使用して設定できます。また、コンマ区切りのリストを --jars 引数に渡すことで、JARをクラスパスに追加できます。依存関係（例えばSparkパッケージ）をシェルセッションに追加するには、Maven座標のコンマ区切りのリストを --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はファイルの各ブロック（HDFSではデフォルトで128MBのブロック）に対して1つのパーティションを作成しますが、より大きな値を渡すことで、より多くのパーティションを要求することもできます。ブロック数よりも少ないパーティションを持つことはできないことに注意してください。

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

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

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

• SequenceFileおよびHadoop Input/Output形式

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

Writableサポート

PySpark SequenceFileサポートは、Java内のキーと値のペアのRDDをロードし、Writablesを基本のJava型に変換し、pickle を使用して結果のJavaオブジェクトをpickle化します。キーと値のペアのRDDをSequenceFileに保存する場合、PySparkは逆のことを行います。PythonオブジェクトをJavaオブジェクトにアンピックルしてから、Writablesに変換します。次のWritablesは自動的に変換されます。

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

SequenceFileの保存とロード

テキストファイルと同様に、SequenceFileはパスを指定することで保存およびロードできます。キーと値のクラスを指定できますが、標準のWritablesにはこれは必要ありません。

>>> 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形式の保存とロード

PySparkは、「新しい」および「古い」Hadoop MapReduce APIの両方について、任意のHadoop InputFormatを読み取り、任意のHadoop OutputFormatを書き込むこともできます。必要に応じて、Hadoop構成をPython dictとして渡すことができます。以下に、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クラスパスに含まれていることを確認してください。

Cassandra/HBase InputFormat と OutputFormat をカスタムコンバーターで使用する例については、Pythonの例とConverterの例を参照してください。

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でファイルを読み取る際の注意点

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

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

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

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

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

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

• 他のHadoop InputFormatの場合は、SparkContext.hadoopRDDメソッドを使用できます。このメソッドは、任意のJobConfと入力形式クラス、キー・クラス、値クラスを取ります。これらは、入力ソースを持つHadoopジョブと同じ方法で設定します。また、「新しい」MapReduce API（org.apache.hadoop.mapreduce）に基づくInputFormatには、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はファイルの各ブロック（HDFSではデフォルトで128MBのブロック）に対して1つのパーティションを作成しますが、より大きな値を渡すことで、より多くのパーティションを要求することもできます。ブロック数よりも少ないパーティションを持つことはできないことに注意してください。

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

• JavaSparkContext.wholeTextFilesを使用すると、複数の小さなテキストファイルを含むディレクトリを読み取り、それらをそれぞれ（ファイル名、内容）のペアとして返します。これは、各ファイルの行ごとに1つのレコードを返すtextFileとは対照的です。

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

• 他のHadoop InputFormatの場合は、JavaSparkContext.hadoopRDDメソッドを使用できます。このメソッドは、任意のJobConfと入力形式クラス、キー・クラス、値クラスを取ります。これらは、入力ソースを持つHadoopジョブと同じ方法で設定します。また、「新しい」MapReduce API（org.apache.hadoop.mapreduce）に基づくInputFormatには、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つの推奨される方法があります。

• 無名関数構文。短いコードで使用できます。
• グローバルシングルトンオブジェクト内の静的メソッド。例えば、object 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で自動的に利用できます。キーと値のペア操作は、タプルのRDDを自動的にラップするPairRDDFunctionsクラスで利用できます。

たとえば、次のコードでは、キーと値のペアに対して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クラスで表されます。JavaRDDからJavaPairRDDを構築するには、mapToPairやflatMapToPairのようなmap操作の特別なバージョンを使用します。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()ドキュメントに記載されているコントラクトを参照してください。

アキュムレータは、SparkContext.accumulator(v)を呼び出すことにより、初期値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インターフェースには、データ型の「ゼロ値」を提供するためのzeroと、2つの値を加算するためのaddInPlaceの2つのメソッドがあります。たとえば、数学ベクトルを表す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())

数値アキュムレータは、それぞれLong型またはDouble型の値を累積するために、SparkContext.longAccumulator()またはSparkContext.doubleAccumulator()を呼び出すことで作成できます。クラスタで実行されているタスクは、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型を定義する場合、結果の型が追加された要素の型と異なる場合があることに注意してください。

数値アキュムレータは、それぞれLong型またはDouble型の値を累積するために、SparkContext.longAccumulator()またはSparkContext.doubleAccumulator()を呼び出すことで作成できます。クラスタで実行されているタスクは、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ジョブが成功しても正しく更新されない可能性があります。