30億行から国別の売上へ
毎週月曜、小売チェーンは国別の売上をひとつ知りたいと考えます。注文の表は30億行あり、20台の機械に分かれて置かれています。
20本のプログラムを書きたくはありません。欲しい結果を一度だけ書き、仕事の分け方はシステムに任せたいのです。
このレッスンのゴール
メモ
データの処理を、Sparkのパイプラインとして表せます。
キーワード
| ことば | いみ | れい |
|---|---|---|
| DataFrame | 名前のついた列を持ち、Sparkが扱う表 | |
| 変換 | 今ある表から、新しい表を作る手順 | filter |
| アクション | 計画を実行して、答えを返す手順 | count |
| 遅延評価 | 答えを求められるまで、仕事を始めないこと | |
| パーティション | 表の一部分で、1台のワーカーが持つもの | |
| ジョブ | アクション1回のために、Sparkが行う仕事 |
学生が入力するセルは、このプレビューでは空白です。
DataFrameは問い合わせられる表
MapReduceでは、mapとreduceを自分で書きます。Sparkでは、DataFrameという表に対して、欲しい結果を書きます。列には名前があります。
filter や groupBy のようなコードでも、SQLでも問い合わせられ、どちらも同じ計画になります。ラボには、24件の注文が orders という名前で読み込まれています。
ラボの答え方
ラボのターミナルにSparkはありません。curlで、小さなサービスに質問を送ります。サービスは、いつも文字の表で答えます。
列の名前はファイルから来ているので英語です。country、product、quantity などです。このサービスは、変換だけの式でも結果を表示します。その際にSparkが仕事をします。
先頭の4行を見る
curl -s -d 'SELECT * FROM orders LIMIT 4' http://spark:8000/sqlここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
異なる国の数を数える
curl -s -d 'orders.select("country").distinct().count()' http://spark:8000/dfここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
変換は待ち、アクションが動かす
filter や groupBy のような変換は、新しいDataFrameを作りますが、まだ何も読みません。Sparkは手順を計画に足すだけです。これを遅延評価と呼びます。
count や collect のようなアクションは、答えを求めます。そこでSparkは、計画の全部を実行します。これがジョブです。
例: 手順とその種類
| 手順 | 種類 | データを読むか |
|---|---|---|
| filter(orders.quantity > 3) | 変換 | いいえ、計画に足される |
| select("country") | 変換 | いいえ |
| groupBy("country").sum("quantity") | 変換 | いいえ |
| count() | アクション | はい、計画を実行する |
| collect() | アクション | はい、全部の行を持ち帰る |
学生が入力するセルは、このプレビューでは空白です。
求められたときに動く計画
# 計画は、手順のリストです。手順を足しても、データは読まれません。
rows = [["France", 2], ["Japan", 1], ["France", 5]]
plan = []
plan.append(["keep_over", 1])
print("計画:", plan)
def run_plan(rows, plan):
# アクションは、計画の全部をデータに対して実行します。
for step in plan:
if step[0] == "keep_over":
kept = []
for row in rows:
if row[1] > step[1]:
kept.append(row)
rows = kept
return rows
print("結果:", run_plan(rows, plan))
# 計画の中の 1 を 2 に変えて、もう一度実行してみましょう。
計画を実行する: 大口の注文
curl -s -d 'orders.filter(orders.quantity >= 5).count()' http://spark:8000/dfここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
よくある間違い
注意
filter の行が、すぐにデータを読むと考えがちです。実際は、計画に手順をひとつ足すだけです。列があるかどうかは、Sparkがその場で確かめます。データの中の悪い値は、アクションが計画を実行したときに初めて現れます。
パーティションが仕事を分ける
Sparkは、DataFrameをパーティションに分けます。各ワーカーは、自分のパーティションを足します。そのあとSparkが、部分的な合計を足し合わせます。
これはmapとreduceと同じ考え方ですが、自分で書く必要はありません。Spark UIには、各ジョブとその計画が出ます。何が動いたかを確かめられます。
部分的な合計から全体へ
# 数量の列を3つに分けたものです。それぞれ別のワーカーが持ちます。
partitions = [[3, 1, 2], [4, 2], [6, 1]]
partials = []
for part in partitions:
total = 0
for quantity in part:
total = total + quantity
partials.append(total)
print("部分的な合計:", partials)
grand = 0
for value in partials:
grand = grand + value
print("全体の合計:", grand)
# 6 を最初の分割に移して、もう一度実行してみましょう。
ordersのパーティション数を尋ねる
curl -s -d 'orders.rdd.getNumPartitions()' http://spark:8000/dfここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
3つに分けて、それぞれの行を数える
curl -s -d 'orders.repartition(3).rdd.glom().map(len).collect()' http://spark:8000/dfここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
国ごとの注文の数
curl -s -d 'SELECT country, count(*) AS orders_count FROM orders GROUP BY country ORDER BY orders_count DESC, country' http://spark:8000/sqlここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
Spark UIはジョブを見せる
ラボのバーからSpark UIを開いてください。Jobsタブには、アクションごとの仕事が並びます。SQL / DataFrameタブには、問い合わせとその計画が並びます。
上のコマンドをもう一度実行して、新しい行が増えるのを見ましょう。パイプラインが遅い理由を、技術者はこうして調べます。
さらに先へ: 商品ごとの平均
/sql に送る問い合わせを、自分でひとつ書いてください。商品ごとに、注文の数と、1件あたりの数量の平均を出します。平均の大きい順に並べます。
実行する前に、どの商品が先頭に来るか予想してください。そのあとSpark UIを開いて、自分の問い合わせが作ったジョブを探しましょう。