Sparkアプリケーションの基本と、はじめに押さえておきたい重要な概念

ゼロからはじめるSparkアプリケーション入門（1）

SparkでCSVファイルの解析をやってみよう

　ここではSparkでのプログラミングに慣れるため、インタラクティブシェルを用いてCSVファイルの解析を行います。本来Sparkは大量データの高速処理が魅力ですが、データの量が増えても処理は変わらないため、小さなサンプルデータでSparkの処理を見ていきます。

　まずはサンプルのCSVファイルを用意します。

id,transaction_id,name,quantity,price
1,t1,商品1,5,100
2,t1,商品2,3,150
3,t2,商品3,10,210
4,t2,商品4,2,1050
5,t2,商品5,40,80
6,t3,商品6,150,30
7,t3,商品1,10,100
8,t4,商品4,5,1050

　ヘッダ1行とデータ8行のCSVデータを用意します。このCSVファイルは商品売買のトランザクションごと（transaction_id）の購入商品（name）と数量（quantity）、商品の購入時価格（price）にIDを付与したものです。サンプルデータであるため、フィールドも値も適当ですが、このデータから、トランザクションごとの購入価格のサマリーを出します。

　このデータを「sample_transaction.csv」というファイル名でローカルに保存し、保存したファイルと同じディレクトリでspark-shellまたはpysparkを起動しておきます。

　以下に今回の処理の流れをまとめます。

1. CSVファイルの読み込み
2. データの整形と中間データ作成
3. 集計処理
4. コンソールに表示して確認
5. ファイルへ出力

CSVファイルの読み込み

　まずはCSVファイルの読み込みを行います。spark-shell、pysparkは起動時にsc（SparkContext）というSparkの動作設定などを扱うオブジェクトを自動で生成します。今回、SparkContextの説明は省略し別の回で触れようと思います。

　また、spark-shell、pysparkは起動したディレクトリを起点にします。先ほど作成したファイル「sample_transaction.csv」を、scのtextFile関数で読み込みRDDを作成します。RDDについては後半で説明します。

ScalaでのCSVファイルの読み込み

val raw = sc.textFile("./sample_transaction.csv")    //scを使って、sample_transaction.csvからRDDを作成

PythonでのCSVファイルの読み込み

raw = sc.textFile("./sample_transaction.csv")

データの整形と中間データ作成

　ここでは読み込んだCSVファイルのデータの整形を行い、後の集計処理につなげるための中間データを作成しています。まず、今回のCSVファイルにはヘッダ行が付いていますので、first関数で1行目を取得し、filter関数でheader行以外のデータを抽出して、map処理で各行をカンマで分割し、transaction_idをkeyに数量と価格のタプルを作成します。作成した各行のタプルをtemp変数に格納します。

Scalaの場合

val head = raw.first()	//ヘッダ行（1行目）の取得
val temp = raw.filter(_ != head).map(row => { //filterでheader行以外を処理
        val fields = row.split(",") //csvを","の区切りでに配列に変換
        (fields(1),fields(3).toInt * fields(4).toInt)
        })

Pythonの場合

#map用の関数の定義
def map_row(row):
    fields = row.split(",")
    return (fields[1],int(fields[3]) * int(fields[4]))
    
head = raw.first()
temp = raw.filter(lambda x: x != head).map(map_row)
)

補足

　インタラクティブシェルで実行する場合、中間データの確認を手軽に行うことができます。次のような関数を利用して、コンソール上に値を表示させることが可能です。

temp.take(5) //5件のデータを取得する
temp.top(10) //値が大きい順に10件
temp.takeOrdered(10) //値が小さい順に10件
temp.takeSample(false,3) //ランダムに3件

集計処理

　作成した中間データ（transaction_id, 金額）をreduceByKey関数を用いてtransaction_idごとに集計します。

Scalaの場合

val summary = temp.reduceByKey((x,y)=>x+y)

Pythonの場合

summary = temp.reduceByKey(lambda x,y:x+y)

コンソールに表示

　集計後のsummaryデータをコンソールに全件出力させます。

Scalaの場合

summary.foreach(println)

Pythonの場合

def p(x): 
    print x

summary.foreach(p)

補足

　今回利用しているデータは全部で4件と少量ですが、本格的なデータセットを扱う場合は、件数が大きくなりがちです。コンソールに全件出力する際は注意ください。

ファイルへの出力

　最後に集計したデータをファイルに出力します。以下のサンプルはカレントディレクトリにsummaryを作成し、ファイルを出力します。

Scalaの場合

summary.saveAsTextFile("./summary")

Pythonの場合

summary.saveAsTextFile("./summary")

　下記がファイル出力の結果です。

補足

　Sparkでのファイル出力はディレクトリを指定します。指定したディレクトリ内には、0byteの出力成否のファイル（上記図の_SUCCESS）とpart-XXXXXの命名規則に従ったファイルが出力されます。作成されるpart-XXXXXはRDDのpartition数に依存します。partitionについての詳細は以後の連載で触れていきます。

【PR】