これに、クラスタは要るのでしょうか
同じ集計を、Spark、SQL、ふつうの Python のどれで書くかは、データの大きさで決まります。選ぶ前に、3通りで同じ仕事をしてみます。
あるチームが、国ごとの売上を求めたいと考えています。1人は Spark を使おうと言い、もう1人は、10行の Python のループで足りると言います。
どちらも正しいことがあります。決め手は、データの大きさです。今日は同じ仕事を3通りで行い、そのあと、理由をつけて選びます。
このレッスンのゴール
メモ
同じ仕事を、Spark の DataFrame、SQL、ふつうの Python で行い、それぞれがいつ向くかを言える。
キーワード
| ことば | いみ | れい |
|---|---|---|
| DataFrame | 行と名前つきの列でできた、Spark が扱う表 | |
| Spark SQL | DataFrame に SQL で問い合わせること | |
| シャッフル | 同じキーが集まるよう、行がワーカーの間を移ること | Exchange |
| 物理計画 | Spark が実際に動かす手順の一覧 | |
| メモリに収まる | 1台のマシンのメモリに入る大きさ |
学生が入力するセルは、このプレビューでは空白です。
1つの仕事を、3通りで
仕事は、国ごとの売上です。数量かける価格を、国ごとに足します。ラボには、24件の注文が orders として読み込まれています。
まず Spark のDataFrameの式で、次に Spark のSQLで、最後に同じファイルをふつうの Python のループで求めます。3つが合わなければ、どれかが間違っています。
DataFrame のやり方
curl -s -d 'orders.groupBy("country").agg(F.round(F.sum(orders.quantity * orders.unit_price), 2).alias("revenue")).orderBy("country")' http://spark:8000/dfここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
SQL のやり方
curl -s -d 'SELECT country, round(sum(quantity * unit_price), 2) AS revenue FROM orders GROUP BY country ORDER BY country' http://spark:8000/sqlここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
ふつうの Python のやり方
curl -s -d 'import csv, collections; c = collections.Counter(); [c.update({r["country"]: float(r["quantity"]) * float(r["unit_price"])}) for r in csv.DictReader(open("/tmp/lab/orders.csv"))]; sorted(c.items())' http://spark:8000/pyここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
変えられる、同じ仕事のループ
# 同じ仕事を、ふつうの Python のループで。国ごとの売上です。
orders = [["UK", 2, 25.0], ["Japan", 1, 150.0], ["UK", 1, 150.0], ["Japan", 4, 12.5]]
revenue = {}
for order in orders:
country = order[0]
money = order[1] * order[2]
if country in revenue:
revenue[country] = revenue[country] + money
else:
revenue[country] = money
print(revenue)
# orders に ["France", 3, 12.5] を足して、もう一度実行してください。
DataFrame は、リストではない
DataFrame は、ループで回せる表に見えますが、計画です。行は、あなたのプログラムの中にありません。だから、リストの技の多くが使えません。
大きさは、アクションの count() で聞きます。len() で長さを聞くと、Python は断ります。
リストのように長さを聞く
curl -s -d 'len(orders)' http://spark:8000/dfここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
Spark のように大きさを聞く
curl -s -d 'orders.count()' http://spark:8000/dfここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
計画は、データの動き先を示す
Spark は、これから動かす物理計画を表示できます。探す言葉は Exchange です。これはシャッフルで、同じキーが集まるように、行がワーカーの間を移ります。
シャッフルは、大きな仕事で高くつく部分です。計画を読めば、それがどこにあるか分かります。
Spark が表示する、物理計画
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#19], functions=[count(1)])
+- Exchange hashpartitioning(country#19, 4), ENSURE_REQUIREMENTS, [plan_id=244]
+- HashAggregate(keys=[country#19], functions=[partial_count(1)])
+- InMemoryTableScan [country#19]
計画の主な演算子を並べる
curl -s -d 'orders.groupBy("country").count()._jdf.queryExecution().simpleString()' http://spark:8000/df | grep -oE "Exchange|HashAggregate|FileScan"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
よくある間違い
注意
Spark は大きいから、いつもよい道具だ、と思いがちです。24行では、起動と計画が増えるだけです。Spark が役に立つのは、データが1台に収まらないとき、または仕事を分けるときです。
例: 道具の選び方
| データ | 妥当な選び方 | わけ |
|---|---|---|
| ファイルに24行 | ふつうの Python のループ | 分けるものがない |
| メモリに収まる200万行 | pandas、またはデータベースの SQL | 1台で数秒 |
| 何台にもまたがる数十億行 | Spark | 仕事を分ける必要がある |
| 毎晩動かす仕事 | 読みやすく書かれ、確かめのあるもの | 毎回、同じ答え |
学生が入力するセルは、このプレビューでは空白です。
さらに: 別の問い
製品ごとの販売数量を、DataFrame のやり方、SQL のやり方、ふつうの Python のやり方で求めます。3つは合いますか。
ヒント: product でまとめ、quantity を足します。そのあと DataFrame の式の計画を表示し、その Exchange を探します。