Spark 2.0.0 はデフォルトでScala 2.11で動作するようにビルドおよび配布されています。(Spark は他のバージョンのScalaでも動作するようにビルドされています) Scalaでアプリケーションを書くには、互換性のあるScalaバージョンを使う必要があるでしょう(例えば、2.11.X)。

Sparkアプリケーションを書くには、MavenのSparkの依存を追加する必要があるでしょう。Sparkは以下のようにMaven Centralを使って利用可能です:

groupId = org.apache.spark
artifactId = spark-core_2.11
version = 2.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 2.0.0 はJava 7以上で動作します。Java8を使っている場合は、Sparkは簡潔な書き込み機能のために lambda 表現 をサポートします。それ以外に、org.apache.spark.api.java.function パッケージの中にあるクラスを使うことができます。

SparkアプリケーションをJavaで書くためにSparkに依存を追加する必要があります。Sparkは以下のようにMaven Centralを使って利用可能です:

groupId = org.apache.spark
artifactId = spark-core_2.11
version = 2.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 2.0.0 はPython 2.6+ あるいは Python 3.4+ で動作します。それは標準のCPythonインタプリタを使うため、NumPyのようなC ライブラリが使われるかもしれません。それは PyPy 2.3+ でも動作します。

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

HDFSデータにアクセスしたい場合には、そのバージョンのHDFSにリンクしているPySparkのビルドを使う必要があります。一般的なHDFSバージョンのためのPrebuilt パッケージ がSparkホームページでも利用可能です。

最後に、プログラムに幾つかのSparkクラスをインポートする必要があります。以下の行を追加します:

from pyspark import SparkContext, SparkConf

PySpark はドライバーおよびワーカーの両方で、Pythonの同じマイナーバージョンを必要とします。PATHにあるデフォルトのpythonバージョンを使います。Pythonのどちらのバージョンを使いたいかをPYSPARK_PYTHONを使って指定することができます。例えば:

$ PYSPARK_PYTHON=python3.4 bin/pyspark
$ PYSPARK_PYTHON=/opt/pypy-2.5/bin/pypy bin/spark-submit examples/src/main/python/pi.py

Sparkプログラムが最初にしなければならないことは、SparkContextオブジェクトを作成する事です。これはSparkにクラスタにアクセスする方法を教えます。SparkContextを作成するために、最初にアプリケーションについての情報を含むSparkConf オブジェクトをビルドする必要があります。

JVMごとに一つのSparkContextだけがアクティブになることができます。新しい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);

Sparkプログラムが最初にしなければならないことは、SparkContextオブジェクトを作成する事です。これはSparkにクラスタにアクセスする方法を教えます。SparkContextを作成するために、最初にアプリケーションについての情報を含むSparkConf オブジェクトをビルドする必要があります。

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

Sparkシェルの中では、特別なインタプリターを認識しているSparkContext、scと呼ばれる変数が既に作成されています。独自のSparkContextを作っても動作しないでしょう。--master 引数を使ってどっちのマスターをコンテキストが接続するかを指定することができます。そして--jars 引数にカンマ区切りのリストで渡すことでクラスパスにJARを追加することができます。--packages 引数にカンマ区切りのmaven coordinateのリストを渡すことで、シェルセッションに依存性(例えばSparkパッケージ)を追加することもできます。依存性が存在するかも知れないどのような追加のリポジトリ(例えば、SonaType)も--repositories 引数に渡すことができます。例えば、確実に4つのコア上でbin/spark-shellを実行するには、以下を使います:

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

あるいは、クラスパスにcode.jarも追加するには、以下を使います:

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

maven coordinatesを使って依存性を追加するには:

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

オプションの完全なリストについては、spark-shell --helpを実行します。舞台裏で spark-shellはもっと多くの一般的な spark-submit スクリプトを呼び出します。

PySparkシェルの中では、特別なインタプリターを認識しているSparkContext、scと呼ばれる変数が既に作成されています。独自のSparkContextを作っても動作しないでしょう。--master引数を使ってどっちのマスターコンテキストをコンテキストが接続するかを指定することができます。 そして--py-files引数にカンマ区切りのリストを渡すことでランタイムにPythonの.zip, .egg あるいは .py ファイルを追加することができます。--packages 引数にカンマ区切りのmaven coordinateのリストを渡すことで、シェルセッションに依存性(例えばSparkパッケージ)を追加することもできます。依存性が存在するかも知れないどのような追加のリポジトリ(例えば、SonaType)も--repositories 引数に渡すことができます。Sparkパッケージが持つどのようなpython依存性(パッケージのrequirements.txtにリスト化されています)も、必要な時にpipを使って手動でインストールされなければなりません。例えば、確実に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内でPySparkシェル、拡張Pythonインタプリタ、を起動することもできます。PySparkはIPython 1.0.0 以降で動作します。IPythonを使用するには、bin/pysparkを実行する時にPYSPARK_DRIVER_PYTHON 変数 ipython に設定します:

$ PYSPARK_DRIVER_PYTHON=ipython ./bin/pyspark

Jupyter notebookを使うには (以前はthe IPython notebookとして知られていました)。

$ PYSPARK_DRIVER_PYTHON=jupyter ./bin/pyspark

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

Jupyter Notebookサーバが起動した後で、"Files"タブから新しい"Python 2" を作成することができます。notebookの中で、Jupyter notebookからSparkを開始使用とする前にnotebookの一部として %pylab inlineコマンドを入力することができます。

並列化されたコレクションはドライバプログラム内(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) を呼ぶかも知れません。後で分散されたデータセット上の操作を説明します。

並列化されたコレクションはドライバプログラム内の既存のコレクション上で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) を呼ぶかも知れません。後で分散されたデータセット上の操作を説明します。

注意: このガイドでは、Java関数を指定するために簡潔なJava8ラムダ構文をしばしば使用しますが、古いバージョンのJavaでorg.apache.spark.api.java.function パッケージの中のインタフェースを実装することができます。詳細はSparkへ関数を渡す で説明します。

並列化されたコレクションはドライバプログラム内の既存の繰り返しまたはコレクション上でSparkContextの parallelize メソッドを呼ぶことで生成されます。コレクションの要素は並列して操作可能な分散データセットからコピーされます。例えば、これは数字の1から5を持つ並列化されたコレクションの生成の仕方です:

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

一度生成すると、分散データセット(distData) は並列して操作されることが可能です。例えば、リストの要素を追加するために、distData.reduce(lambda a, b: a + b) を呼ぶことができます。後で分散されたデータセット上の操作を説明します。

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

テキストファイルのRDDはSparkContextのtextFileメソッドを使って生成することができます。このメソッドはファイルのURI(マシーン上のローカルファイルまたは hdfs://, s3n://, など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はファイルの各ブロック(HDFSではブロックはデフォルトでは64MB)に1つのパーティションを生成しますが、大きな値を渡すことでより大きな数のパーティションを要求することもできます。ブロック数より少ないパーティションを持つことができないことに注意してください。

テキストファイル以外に、SparkのScala APIは幾つかの他のデータフォーマットもサポートします:

• SparkContext.wholeTextFiles は複数の小さなテキストファイルを含むディレクトリを読み込み、それらのペア(ファイル名,内容)を返します。これはtextFileと異なり、各ファイルごとに行あたり1つのレコードを返します。

• SequenceFilesについては、SparkContextの sequenceFile[K, V] メソッドを使用します。この時、K および V はファイル内のキーの種類と値です。これらはIntWritable および TextのようにHadoopのWritableインタフェースのサブクラスでなければなりません。更に、2,3の共通のWritableのためにSparkによってネイティブなタイプを指定することができます。例えば、sequenceFile[Int, String] は自動的にIntWritableおよびTextを読み込むでしょう。

• 他のHadoop 入力フォームに関しては、SparkContext.hadoopRDD メソッドを使うことができます。これは任意のJobConf および、入力フォーマットクラス、キークラスと値クラスを取ります。同じようにしてHadoopジョブのためにこれらに入力ソースを設定するかもしれません。"新しい" MapReduce API (org.apache.hadoop.mapreduce)に基づいたInputFormatのためのSparkContext.newAPIHadoopRDDも使うことができます。

• RDD.saveAsObjectFile およびSparkContext.objectFileはシリアライズ化されたJavaオブジェクトからなる単純な形式へのRDDの保存をサポートします。これはAvroのように特殊化されたフォーマットのように効率的ではありませんが、どのようなRDDを保存する簡単な方法を提供します。

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

テキストファイルのRDDはSparkContextのtextFileメソッドを使って生成することができます。このメソッドはファイルのURI(マシーン上のローカルファイルまたは hdfs://, s3n://, など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ではブロックはデフォルトでは64MB)に1つのパーティションを生成しますが、大きな値を渡すことでより大きな数のパーティションを要求することもできます。ブロック数より少ないパーティションを持つことができないことに注意してください。

テキストファイル以外に、SparkのJava APIは幾つか他のデータフォーマットもサポートします:

• JavaSparkContext.wholeTextFiles は複数の小さなテキストファイルを含むディレクトリを読み込み、それらのペア(ファイル名,内容)を返します。これはtextFileと異なり、各ファイルごとに行あたり1つのレコードを返します。

• SequenceFilesについては、SparkContextの sequenceFile[K, V] メソッドを使用します。この時、K および V はファイル内のキーの種類と値です。これらはIntWritable および TextのようにHadoopのWritableインタフェースのサブクラスでなければなりません。

• 他のHadoop 入力フォームに関しては、JavaSparkContext.hadoopRDD メソッドを使うことができます。これは任意のJobConf および、入力フォーマットクラス、キークラスと値クラスを取ります。同じようにしてHadoopジョブのためにこれらに入力ソースを設定するかもしれません。"新しい" MapReduce API (org.apache.hadoop.mapreduce)に基づいたInputFormatのためのJavaSparkContext.newAPIHadoopRDDも使うことができます。

• JavaRDD.saveAsObjectFile およびJavaSparkContext.objectFileはシリアライズ化されたJavaオブジェクトからなる単純な形式へのRDDの保存をサポートします。これはAvroのように特殊化されたフォーマットのように効率的ではありませんが、どのようなRDDを保存する簡単な方法を提供します。

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

テキストファイルのRDDはSparkContextのtextFileメソッドを使って生成することができます。このメソッドはファイルのURI(マシーン上のローカルファイルまたは hdfs://, s3n://, など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ではブロックはデフォルトでは64MB)に1つのパーティションを生成しますが、大きな値を渡すことでより大きな数のパーティションを要求することもできます。ブロック数より少ないパーティションを持つことができないことに注意してください。

テキストファイル以外に、SparkのPython APIは幾つか他のデータフォーマットもサポートします:

• SparkContext.wholeTextFiles は複数の小さなテキストファイルを含むディレクトリを読み込み、それらのペア(ファイル名,内容)を返します。これはtextFileと異なり、各ファイルごとに行あたり1つのレコードを返します。

• RDD.saveAsObjectFile およびSparkContext.objectFileはpickle化されたPythonオブジェクトからなる単純な形式へのRDDの保存をサポートします。バッチはデフォルトのバッチサイズ 10を使って、pickleシリアライズ化に使われます。

• SequenceFile と Hadoop Input/Output フォーマット

注意 この機能は現在のところ 実験的なもの とマークされていて、上級ユーザのためのものであることに注意してください。将来Spark SQLに基づいた read/writeサポートに置き換えられるかも知れませんが、いずれにしてもSpark SQLは好ましい方法です。

Writableサポート

PySpark SequenceFile はJava内のキー-値のRDDのロードをサポートし、Writableを基本のJavaのタイプに変換し、結果のJavaオブジェクトを Pyroliteを使ってピクル化します。キー-値のペアのRDDをSequenceFileに保存する場合、PySparkは逆のことをします。それはPythonオブジェクトをJavaオブジェクトにアンピクル化し、それらをWritableに変換します。以下のWritableは自動的に変換されます:

配列は追加設定無しで処理することができません。読み書きする場合には、ユーザは独自のArrayWritableサブタイプを指定する必要があります。書き込む場合には、ユーザは配列を独自の ArrayWritable サブタイプに変換する独自のコンバータも指定する必要があります。読み込む場合には、デフォルトのコンバータが独自の ArrayWritable サブタイプを Java のObject[]に変換するでしょう。これはPythonのタプルにピクル化されます。基本的なタイプの配列のためのPython array.arrayを取得ｓるうには、ユーザは独自のコンバータを指定する必要があります。

SequenceFileの保存とロード

テキストファイルと似て、SequenceFiles はパスを指定することによって保存およびロードすることができます。キーと値のクラスは指定することができますが、標準の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 フォーマットの保存およびロード

PySparkは、'new'および'old' Hadoop MapReduce APIについて、任意のHadoop InputFormatの読み込み、あるいは任意のHadoopのOutputFormatの書き込みができます。必要であれば、Hadoop設定をPython dictとして渡すことができます。これはElasticsearch ESInputFormatを使った例です:

$ SPARK_CLASSPATH=/path/to/elasticsearch-hadoop.jar ./bin/pyspark
>>> 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からロードしたデータなど)独自のシリアライズ化バイナリデータがある場合は、まずそのデータをPyroliteのピックラーが処理できるものにScala/Java側で変換しなければならないでしょう。このために Converter trait が提供されます。単にこのtraitを拡張し、変換コードをconvertメソッドの中に実装します。InputFormatにアクセスを必要とする全ての依存物と一緒にこのクラスはSparkジョブjarにパッケージされ、PySparkクラスパッスに含まれるようにすることを忘れないでください。

Cassandra / HBase の独自のコンバータを使ったInputFormat および OutputFormatの例は、Python examples および Converter examples を見てください。

RDDの基本を説明するために、以下の単純なプログラムを考えます:

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

最初の行は外部ファイルからの基本のRDDを定義します。このデータセットはメモリにロードされないか、あるいは以下のように振る舞います: lines は単にファイルへのポインターです。二つ目の行はmap変換の結果としての lineLengths を定義します。ここでも、lazinessのために lineLengthsはすぐには計算されません。最後に、reduceを実行します。これはアクションです。この時点でSparkは個々のマシーン上で実行するために計算をタスクに分割し、各マシーンはmapとローカルのreductionを実行し、ドライバープログラムに回答のみを返します。

後で再び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 は単にファイルへのポインターです。二つ目の行はmap変換の結果としての lineLengths を定義します。ここでも、lazinessのために lineLengthsはすぐには計算されません。最後に、reduceを実行します。これはアクションです。この時点でSparkは個々のマシーン上で実行するために計算をタスクに分割し、各マシーンはmapとローカルのreductionを実行し、ドライバープログラムに回答のみを返します。

後で再びlineLengthsを使用したい場合は、以下を追加するかも知れません:

lineLengths.persist(StorageLevel.MEMORY_ONLY());

reduceの前に、最初にそれが計算された後でメモリ内に lineLengths が保存されるでしょう。

RDDの基本を説明するために、以下の単純なプログラムを考えます:

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

最初の行は外部ファイルからの基本のRDDを定義します。このデータセットはメモリにロードされないか、あるいは以下のように振る舞います: lines は単にファイルへのポインターです。二つ目の行はmap変換の結果としての lineLengths を定義します。ここでも、lazinessのために lineLengthsはすぐには計算されません。最後に、reduceを実行します。これはアクションです。この時点でSparkは個々のマシーン上で実行するために計算をタスクに分割し、各マシーンはmapとローカルのreductionを実行し、ドライバープログラムに回答のみを返します。

後で再びlineLengthsを使用したい場合は、以下を追加するかも知れません:

lineLengths.persist()

reduceの前に、最初にそれが計算された後でメモリ内に lineLengths が保存されるでしょう。

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へそのインスタンスを渡します。.
• Java8では、実装を簡潔に定義するためにlambda 表記を使います。

このガイドのほとんどは簡潔化のためにlambda構文を使用しますが、長い形式の全ての同じ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のAPIはクラスター上で実行するためにドライバープログラム内で渡す関数に強く依存します。これをするには3つのお勧めの方法があります:

• Lambda 表記、簡単な関数のために、表現として書くことができます。(Lambdaは、値を返す複数行の関数あるいは命令文をサポートしません)
• 長いコードのための、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操作はあらゆる種類のオブジェクトを含むRDD上で動作しますが、2,3の特別な操作はキー-値ペアのRDD上でのみ利用可能です。もっとも一般的なものは、キーを使って要素をグループあるいは集約するような分散型"shuffle"操作です。

Scalaでは、これらの操作は、(言語内の組み込みの、単純に(a, b)と書くことで生成される)、Tuple2 オブジェクトを含むRDD上で自動的に利用可能です。キー-値ペアの操作はPairRDDFunctionsクラスの中で利用可能です。これはtupleのRDDの周りを自動的に取り囲みます。

例えば、以下のコードはファイル内のテキストの各行が何回現れるかを数えるためにキー-値ペア上のreduceByKey操作を使用します。

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

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

注意: 独自のオブジェクトをキー-値ペアの操作として使う場合、独自の equals() メソッドがhashCode() メソッドを使って実施されるようにしなければなりません。完全な詳細は、Object.hashCode() ドキュメントの中で説明される規約を見てください。

ほとんどのSpark操作はあらゆる種類のオブジェクトを含むRDD上で動作しますが、2,3の特別な操作はキー-値ペアのRDD上でのみ利用可能です。もっとも一般的なものは、キーを使って要素をグループあるいは集約するような分散型"shuffle"操作です。

Javaでは、キー-値ペアはScalaの標準ライブラリのscala.Tuple2クラスを使って表現されます。tupleを生成するためには単純に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.collect()するために、counts.sortByKey()を使うこともできます。

注意: 独自のオブジェクトをキー-値ペアの操作として使う場合、独自の equals() メソッドがhashCode() メソッドを使って実施されるようにしなければなりません。完全な詳細は、Object.hashCode() ドキュメントの中で説明される規約を見てください。

ほとんどのSpark操作はあらゆる種類のオブジェクトを含むRDD上で動作しますが、2,3の特別な操作はキー-値ペアのRDD上でのみ利用可能です。もっとも一般的なものは、キーを使って要素をグループあるいは集約するような分散型"shuffle"操作です。

Pythonでは、これらの操作は(1, 2)のような組み込みのPythonのtupleを含むRDD上で動作します。単純にそのようなtupleを生成し、期待する操作を呼び出します。

例えば、以下のコードはファイル内のテキストの各行が何回現れるかを数えるためにキー-値ペア上のreduceByKey操作を使用します。

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

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

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のaccumulatorのための組み込みのサポートに使われますが、プログラマは AccumulatorV2のサブクラス化をすることで独自のタイプを生成することもできます。AccumulatorV2 abstract クラスは上書きされるべきいくつかのメソッドを持ちます: accumulatorを0にリセットするためのreset、accumulatorに他の値を追加するためのadd、他の同じタイプのaccumulatorを1つにマージするためのmergeです。上書きする必要がある他のメソッドはscalaのAPIドキュメントで参照することができます。例えば、数学的なベクターを表す MyVectorクラスを持つとした場合、以下のように書けます:

object VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {
  val vec_ : MyVector = MyVector.createZeroVector
  def reset(): MyVector = {
    vec_.reset()
  }
  def add(v1: MyVector, v2: MyVector): MyVector = {
    vec_.add(v2)
  }
  ...
}

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

プログラマが独自のタイプのAccumulatorV2を定義する場合、結果のタイプは同じになるか、あるいは要素の追加で違うものになるかも知れないことに注意してください。

LongAccumulator accum = sc.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

このコードはタイプIntegerのaccumulatorのための組み込みのサポートに使われますが、プログラマは AccumulatorParamのサブクラス化をすることで独自のタイプを生成することもできます。AccumulatorParam インタフェースは2つのメソッドを持ちます: データタイプのための"zero値"の提供のためのzero、および2つの値を一緒に追加するためのaddInPlace。例えば、数学的なベクターを表すVectorクラスを持つとした場合、以下のように書けます:

class VectorAccumulatorParam implements AccumulatorParam<Vector> {
  public Vector zero(Vector initialValue) {
    return Vector.zeros(initialValue.size());
  }
  public Vector addInPlace(Vector v1, Vector v2) {
    v1.addInPlace(v2); return v1;
  }
}

// Then, create an Accumulator of this type:
Accumulator<Vector> vecAccum = sc.accumulator(new Vector(...), new VectorAccumulatorParam());

Javaにおいて、Sparkは、結果のタイプが追加された要素と同じでは無い場所でデータを集めるためのもっと一般的なAccumulable インタフェース(つまり、要素を集めることでリストを構築する)も提供します。

>>> accum = sc.accumulator(0)
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

scala> accum.value
10

このコードはタイプIntのaccumulatorのための組み込みのサポートに使われますが、プログラマは AccumulatorParamのサブクラス化をすることで独自のタイプを生成することもできます。AccumulatorParam インタフェースは2つのメソッドを持ちます: データタイプのための"zero値"の提供のための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())

Spark 1.0 はSparkコアのAPIを 1.X系列のためにフリーズします。ですので、"実験的"あるいは"開発者API"と印が付けられていない今利用可能な全てのAPIは将来のバージョンでサポートされるでしょう。Scalaユーザのための唯一の変更はグループ化の操作です。例えば groupByKey, cogroup および joinは (Key, Seq[Value]) ペアを返すことから (Key, Iterable[Value])を返すように変わりました。

Spark 1.0 はSparkコアのAPIを 1.X系列のためにフリーズします。ですので、"実験的"あるいは"開発者API"と印が付けられていない今利用可能な全てのAPIは将来のバージョンでサポートされるでしょう。幾つかの変換がJava APIに行われました:

• org.apache.spark.api.java.function内のFunctionクラスは1.0でインタフェースになりました。これはFunctionを継承する古いコードは代わりに Functionを継承 しなければなりません。
• mapToPair とmapToDoubleのように、map変換の新しい変数が特別なデータタイプのRDDを生成するために追加されました。
• groupByKey, cogroup および join のようなグルーピング操作は、(Key, List<Value>)ペアを返すことから (Key, Iterable<Value>)を返すように変わりました。

Spark 1.0 はSparkコアのAPIを 1.X系列のためにフリーズします。ですので、"実験的"あるいは"開発者API"と印が付けられていない今利用可能な全てのAPIは将来のバージョンでサポートされるでしょう。Pythonユーザのための唯一の変更はグループ化の操作です。例えば groupByKey, cogroup および joinは、(key, list of values) ペアを返すことから (key, iterable of values) を返すように変わりました。