Spark シェルと RDD の基本操作 - E-MapReduce - Alibaba Cloud ドキュメントセンター

このトピックでは、SparkシェルとResilient Distributed Datasets（RDD）の基本操作について説明します。

Sparkシェルの起動

Sparkシェルは、インタラクティブなデータ分析のための強力なツールであり、APIを学習するための簡単な方法を提供します。 Sparkシェルは、ScalaとPythonの両方をサポートしています。

Sparkシェルを起動するには、次の手順を実行します。

SSH モードでクラスターのマスターノードにログインします。詳細については、「クラスターへのログイン」をご参照ください。
次のコマンドを実行して Spark シェルを起動します。
```
spark-shell
```
Spark シェルでは、sc という名前の変数に特別な SparkContext がすでに作成されています。自己作成の SparkContext は有効になりません。--master パラメーターを使用して SparkContext が接続されるマスターノードを指定し、--jars パラメーターを使用してクラスパスに追加される JAR パッケージを指定できます。複数の JAR パッケージはコンマ (,) で区切ります。パラメーターの詳細については、spark-shell --help コマンドを実行してください。

RDDの基本操作

SparkはRDDを提供します。 RDDは、並列に操作できる要素のコレクションです。並列化されたコレクション、または共有ファイルシステム、Hadoop Distributed File System（HDFS）、HBase、Hadoop InputFormatを提供する外部データセットなどの外部データセットを使用して、RDDを作成できます。

サンプルコード:

並列化されたコレクションを使用して RDD を作成する
```
val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)
```
外部データセットを使用して RDD を作成する
```
val distFile = sc.textFile("data.txt")
```

RDD の操作は、変換 (transformation) とアクション (action) の 2 種類に分類できます。変換は、アクションとは異なり、すぐには実行されません。

変換 (Transformations)

操作	説明
`map()`	RDD の各要素に関数を適用し、結果を新しい RDD として返します。
`flatMap()`	RDD の各要素に関数を適用し、RDD をフラット化してから、結果を新しい RDD として返します。
`filter()`	関数が true を返す RDD の要素を選択し、結果を新しい RDD として返します。
`distinct()`	関数を使用せずに RDD の一意の要素を返します。
`union()`	2 つの RDD の要素を結合し、結果を新しい RDD として返します。
`intersection()`	両方の RDD に出現する要素を返します。
`subtract()`	両方の RDD に出現する要素を削除します。
`cartesian()`	2 つの RDD のデカルト積を生成します。

アクション (Actions)

操作	説明
`collect()`	RDD のすべての要素を返します。
`count()`	RDD の要素数を返します。
`countByValue()`	RDD 内で各要素が出現する回数を返します。
`reduce()`	RDD 内のすべての要素を集約します (合計の計算など)。
`fold(0)(func)`	`reduce()` に似ていますが、初期値としてゼロ値を取る点が異なります。
`aggregate(0)(seqOp,combop)`	`reduce()` に似ていますが、返される RDD のデータの型が元の RDD のデータの型と異なる点が異なります。
`foreach(func)`	RDD の各要素に特定の関数を適用します。

Spark シェルの使用開始例

まず、外部ストレージシステムからテキストファイルを読み取って RDD を作成します。map オペレーターを使用して、テキストファイル内の各行の長さを計算します。reduce オペレーターを使用して、長さを合計します。これにより、テキストファイル内のすべての行の合計長を取得できます。

SSH モードでクラスターのマスターノードにログインします。詳細については、「クラスターへのログイン」をご参照ください。
data.txt ファイルを作成し、HDFS にアップロードします。
この例の data.txt ファイルの内容は以下のとおりです。
```
Hello Spark
This is a test file
1234567890
```
次のコマンドを使用して data.txt を HDFS にアップロードできます。
```
hadoop fs -put data.txt /user/root/
```
Spark シェルを起動します。
```
spark-shell
```

次のコマンドを実行して、data.txt ファイル内のすべての行の合計文字長を計算します。

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