Spark が遅いのはなぜですか??-Python チュートリアル-php.cn

Spark が遅いのはなぜですか??

Mary-Kate Olsen

リリース： 2024-12-11 19:43:11

オリジナル

873 人が閲覧しました

Spark が遅いのはなぜですか??

「なぜ Spark は遅いのですか??」という目を引くタイトルから始まりますが、Spark を「遅い」と呼ぶことはさまざまな意味を持つことに注意することが重要です。集計では遅いですか?データ読み込み?さまざまなケースが存在します。また、「Spark」は広義の用語であり、そのパフォーマンスはプログラミング言語や使用状況などの要素に依存します。それでは、本題に入る前に、より正確になるようにタイトルを調整しましょう。

私は主に Databricks 上の Python で Spark を使用するため、範囲をさらに絞り込みます。

洗練されたタイトルは次のようになります:

「Spark の第一印象: 「速いと聞いていたのに、なぜ遅く感じるの?」初心者の視点」

執筆動機（何気ない思い）

パンダ、NumPy、機械学習ライブラリを幅広く使っている私は、並列分散処理でビッグデータを処理できる Spark の機能の魅力に感心していました。ようやく仕事で Spark を使用できるようになったとき、パンダよりも遅いように見えるシナリオに困惑しました。何が問題だったのかよくわかりませんでしたが、いくつかの洞察を発見したので、それらを共有したいと思います。

火花が遅くなるのはいつですか?

本題に入る前に

Spark の基本アーキテクチャについて簡単に説明しましょう。

Why Is Spark Slow??

(クラスターモードの概要)

Spark クラスターは、実際の処理を実行する ワーカーノード と、実行を調整および計画する ドライバーノード で構成されます。このアーキテクチャは、以下で説明するすべてに影響を与えるため、覚えておいてください。

それでは、本題に入ります。

1. データセットの大きさが十分ではありません

Spark は大規模なデータ処理用に最適化されていますが、小規模なデータセットも処理できます。ただし、このベンチマークを見てください:

Why Is Spark Slow??

(単一ノードマシンでの Apache Spark のベンチマーク)

結果は、15 GB 未満のデータセットでは、集計タスクにおいて pandas が Spark よりも優れていることを示しています。なぜ？一言で言えば、Spark の最適化によるオーバーヘッドが、小規模なデータセットの利点を上回ります。

リンクには、Spark が遅くないケースが示されていますが、これらは多くの場合 ローカルクラスターモード です。スタンドアロン設定の場合、ノード間のネットワーク通信のオーバーヘッドにより、データセットが小さいと不利になる可能性があります。

pandas: ネットワークやストレージ I/O を使用せず、単一マシン上のメモリ内のすべてを処理します。
Spark: RDD (Resilient Distributed Datasets) を使用し、ワーカー間のネットワーク通信 (分散されている場合) を伴い、並列処理のためにデータを整理する際にオーバーヘッドが発生します。

2. 遅延評価について理解する

Spark は 遅延評価 を採用しています。これは、変換がすぐには実行されず、アクション (収集、カウント、表示など) が計算をトリガーするまで延期されることを意味します。

例 (パンダ):

df = spark.read.table("tpch.lineitem").limit(1000).toPandas()
df["l_tax_percentage"] = df["l_tax"] * 100
for l_orderkey, group_df in df.groupby("l_orderkey"):
    print(l_orderkey, group_df["l_tax_percentage"].mean())

ログイン後にコピー

実行時間: 3.04 秒

Spark での同等物:

from pyspark.sql import functions as F
sdf = spark.read.table("tpch.lineitem").limit(1000)
sdf = sdf.withColumn("l_tax_percentage", F.col("l_tax") * 100)

for row in sdf.select("l_orderkey").distinct().collect():
    grouped_sdf = sdf.filter(F.col("l_orderkey") == row.l_orderkey).groupBy("l_orderkey").agg(
        F.mean("l_tax_percentage").alias("avg_l_tax_percentage")
    )
    print(grouped_sdf.show())

ログイン後にコピー

実行時間: 3 分経ってもまだ実行中です。

なぜ?

遅延評価: すべての変換はキューに入れられ、show などのアクション中にのみ実行されます。
ドライバーからドライバーへの通信: 収集や表示などの操作には、作業者からドライバーへのデータ転送が含まれるため、遅延が発生します。

Spark コードは、これをパンダで効果的に実行します。

for l_orderkey, group_df in df.groupby("l_orderkey"):
    df["l_tax_percentage"] = df["l_tax"] * 100
    print(l_orderkey, group_df["l_tax_percentage"].mean())

ログイン後にコピー

Spark の キャッシュ を使用するか、ロジックを再構築して計算の繰り返しを最小限に抑えることで、このようなパターンを回避します。