Spark 中的 pandas API

此頁面說明 Spark 中的 pandas API（「Spark 中的 pandas」）的優點，以及何時應該使用它來取代 pandas（或與 pandas 搭配使用）。

Spark 中的 pandas 速度可能比 pandas 快得多，而且提供讓 pandas 使用者熟悉的語法。它提供 Spark 的功能，同時具備 pandas 的熟悉度。

以下是 Spark 中的 pandas 的主要優點

  • 在單機工作負載上執行查詢的速度較快（因為 Spark 中的 pandas 使用所有可用的核心，平行處理查詢，並最佳化查詢）
  • Spark 中的 pandas 可擴充至叢集中的多台機器，並可處理大數據
  • Spark 中的 pandas 允許在比記憶體資料集更大的資料集上執行查詢
  • Spark 中的 pandas 為 pandas 使用者提供熟悉的語法

pandas 有許多限制

  • pandas 必須在執行查詢前將所有資料載入記憶體，這可能會很慢
  • pandas 無法處理比單機上可用記憶體更大的資料集（因為所有資料都必須載入記憶體）
  • pandas 計算在單一核心上執行，而且不會利用單機的所有可用核心
  • pandas 計算無法擴充至多台機器
  • pandas 沒有查詢最佳化器，因此使用者必須手動編寫最佳化程式碼，否則會導致程式碼執行速度變慢

讓我們來看一些簡單的範例，以更深入瞭解 Spark 中的 pandas 如何克服 pandas 的限制。我們也會探討 Spark 中的 pandas 的限制。

在此頁面的最後，您將看到如何同時使用 pandas 和 Spark 中的 pandas。這不是非此即彼的決定 - 在許多情況下，同時使用兩者是一個很好的選擇。

Spark 中的 pandas 範例

此部分說明 Spark 中的 pandas 如何在 localhost 上的單一檔案中執行查詢的速度比 pandas 快。Spark 中的 pandas 並非對所有查詢都一定比較快，但此範例顯示它何時提供顯著的加速。

假設您有一個 Parquet 檔案，其中有 9 個欄位和 10 億列資料。以下是檔案的前三列

+-------+-------+--------------+-------+-------+--------+------+------+---------+
| id1   | id2   | id3          |   id4 |   id5 |    id6 |   v1 |   v2 |      v3 |
|-------+-------+--------------+-------+-------+--------+------+------+---------|
| id016 | id046 | id0000109363 |    88 |    13 | 146094 |    4 |    6 | 18.8377 |
| id039 | id087 | id0000466766 |    14 |    30 | 111330 |    4 |   14 | 46.7973 |
| id047 | id098 | id0000307804 |    85 |    23 | 187639 |    3 |    5 | 47.5773 |
+-------+-------+--------------+-------+-------+--------+------+------+---------+

以下是如何使用 Spark 中的 pandas 讀取檔案並執行群組聚合。

import pyspark.pandas as ps

df = ps.read_parquet("G1_1e9_1e2_0_0.parquet")[
    ["id1", "id2", "v3"]
]
df.query("id1 > 'id098'").groupby("id2").sum().head(3)

此查詢在配備 64 GB RAM 的 2020 M1 Macbook 上執行 Spark 3.5.0 時，執行時間為 62 秒。

讓我們將此與未最佳化的 pandas 程式碼進行比較。

import pandas as pd

df = pd.read_parquet("G1_1e9_1e2_0_0.parquet")[
    ["id1", "id2", "v3"]
]
df.query("id1 > 'id098'").groupby("id2").sum().head(3)

此查詢會傳回錯誤，因為配備 64GB RAM 的機器沒有足夠的空間來儲存 10 億列資料於記憶體中。

讓我們手動新增一些 pandas 最佳化，以執行查詢

df = pd.read_parquet(
    "G1_1e9_1e2_0_0.parquet",
    columns=["id1", "id2", "v3"],
    filters=[("id1", ">", "id098")],
    engine="pyarrow",
)
df.query("id1 > 'id098'").groupby("id2").sum().head(3)

此查詢在 pandas 2.2.0 中執行時間為 275 秒。

手動使用 pandas 編碼這些最佳化可能會產生錯誤的結果。以下是群組查詢的範例，其正確無誤，但具有錯誤的列群組篩選謂詞

df = pd.read_parquet(
    "G1_1e9_1e2_0_0.parquet",
    columns=["id1", "id2", "v3"],
    filters=[("id1", "==", "id001")],
    engine="pyarrow",
)
df.query("id1 > 'id098'").groupby("id2").sum().head(3)

這會傳回錯誤的結果，即使群組聚合邏輯是正確的！

使用 pandas 時，您需要在讀取 Parquet 檔案時手動套用欄位篩選和列群組篩選。使用 Spark 中的 pandas 時，Spark 最佳化器會自動套用這些查詢增強功能，因此您不需要手動輸入。

讓我們更詳細地探討 Spark 中 pandas 的優點。

Spark 中 pandas 的優點

讓我們回顧 Spark 中 pandas 的優點

更快的查詢執行

Spark 中的 pandas 可以比 pandas 更快地執行查詢，因為它使用所有可用的核心來平行化運算，並在執行查詢之前最佳化查詢以允許有效率的執行。

pandas 運算只會在單一核心上執行。

可擴充至大於記憶體的資料集

pandas 在執行查詢之前會將資料載入記憶體，因此它只能查詢符合記憶體大小的資料集。

Spark 可以透過串流資料和遞增執行運算來查詢大於記憶體的資料集。

當資料集大小增加時，pandas 以傳回錯誤而臭名昭著，而 Spark 沒有這個限制。

可執行於多部機器叢集

Spark 可執行於單一機器，或分佈於叢集中的多部機器。

當 Spark 執行於單一機器時，運算會執行於所有可用的核心。這通常比僅於單一核心執行運算的 pandas 更快。

當您想對大型資料集執行運算，或僅僅存取更多 RAM/核心以使查詢執行得更快時，於多部機器上擴充運算非常棒。

pandas 使用者熟悉的語法

Spark 上的 pandas 被設計成提供 pandas 使用者熟悉的語法。

熟悉的語法是重點所在 - Spark 上的 pandas 提供 Spark 的效能，並使用 pandas 使用者習慣的相同語法。

提供存取 Spark 經過實戰考驗的查詢最佳化器

Spark 上的 pandas 運算會在執行前，由 Spark 的 Catalyst 最佳化器最佳化。

這些最佳化會簡化查詢並加入最佳化。

在本文稍早，我們看到 Spark 如何在讀取 Parquet 檔案時自動加入欄位修剪/列群組篩選最佳化。pandas 沒有查詢最佳化器，因此您需要自行加入這些最佳化。手動加入最佳化很繁瑣且容易出錯。如果您沒有手動套用正確的最佳化，您的查詢將傳回錯誤的結果。

Spark 上 pandas 的限制

Spark 上的 pandas 並不支援 pandas 支援的所有 API，原因有兩個

  • 有些功能尚未加入 Spark 上的 pandas

  • 有些 pandas 功能與 Spark 的分散式平行執行模型不符

Spark 會將資料框分割成多個區塊，以便它們可以平行處理，因此某些 pandas 作業無法順利轉換到 Spark 的執行模型。

將 Spark 上的 pandas 與一般的 pandas 結合使用

經常會同時使用 Spark 上的 pandas 和 pandas，以取得兩全其美的效果。

假設您有一個大型資料集，您會將其清理並彙總成一個較小的資料集，然後傳遞到 scikit-learn 機器學習模型。

您可以使用 Spark 上的 pandas 清理和彙總資料集，以利用快速查詢時間和平行執行。處理資料集後，您可以使用 to_pandas() 將其轉換為 pandas 資料框，然後使用 scikit-learn 執行機器學習模型。如果資料集可以縮小到足以放入 pandas 資料框中，則此方法效果很好。

Spark 上的 pandas 查詢執行模型不同

Spark 上的 pandas 執行查詢的方式與 pandas 完全不同。

Spark 上的 pandas 使用延遲評估。它將查詢轉換為未解析的邏輯計畫，使用 Spark 對其進行最佳化，並且僅在請求結果時才執行運算。

pandas 使用立即評估。它將所有資料載入記憶體，並在呼叫時立即執行作業。pandas 不套用查詢最佳化，並且在執行查詢之前必須將所有資料載入記憶體。

在比較 Spark 上的 pandas 和 pandas 時，您必須小心考量將資料載入記憶體需要多少時間，以及執行查詢需要多少時間。許多資料集需要很長的時間才能載入 pandas。

您也可以使用 Spark 上的 pandas 將資料載入記憶體，但這通常被視為反模式。如果資料載入記憶體中，則儲存中的資料變更時（透過附加、合併或刪除）不會更新資料集。在某些情況下，保留 Spark 資料框是明智的，以加快查詢時間，但必須小心使用，因為它可能會導致查詢結果不正確。

Spark 上的 pandas 和 PySpark 的差異

Spark 上的 pandas 和 PySpark 都會接收查詢，將它們轉換為未解析的邏輯計畫，然後使用 Spark 執行它們。

PySpark 和 Spark 上的 pandas 都具有類似的查詢執行模型。將查詢轉換為未解析的邏輯計畫相對快速。最佳化查詢並執行它需要更多時間。因此，PySpark 和 Spark 上的 pandas 應具有類似的效能。

Spark 上的 pandas 和 PySpark 之間的主要差異僅在於語法。

結論

Spark 上的 pandas 是希望更快執行查詢並希望利用 Spark 的最佳化程式，而不是撰寫自己的最佳化的 pandas 使用者的絕佳替代方案。

pandas on Spark 使用了 pandas 使用者熟悉的語法，因此很容易學習。

pandas on Spark 也是與 pandas 搭配使用的絕佳技術。您可以在 pandas on Spark 中使用大數據和高效能處理功能，在將資料集轉換成與其他技術相容的 pandas 資料框之前處理資料集。

查看 文件 以進一步瞭解如何使用 pandas on Spark。

最新消息

封存