共用方式為

pandas 函式 API

  • 發行項

pandas 函式 API 可讓您直接套用 Python 原生函式，以將 pandas 實例採用並輸出至 PySpark DataFrame。 與 pandas 使用者定義函式類似，函式 API 也會使用 Apache Arrow 來傳輸資料和 pandas 來處理資料;不過，Pandas 函式 API 中的 Python 類型提示是選擇性的。

Pandas 函式 API 有三種類型：

  • 群組地圖
  • 地圖
  • 群組對應

pandas 函式 API 會利用 pandas UDF 執行所使用的相同內部邏輯。 它們會共用 PyArrow、支援的 SQL 類型和組態等特性。

如需詳細資訊，請參閱 即將推出的 Apache Spark 3.0 版本中的部落格文章 New Pandas UDF 和 Python 類型提示。

群組地圖

您可以使用 來轉換群組資料 groupBy().applyInPandas() ，以實作「split-apply-combine」模式。 Split-apply-combine 包含三個步驟：

  • 使用 DataFrame.groupBy 將資料分割成群組。
  • 在每個群組上套用函式。 函式的輸入和輸出都是 pandas.DataFrame 。 輸入資料包含每個群組的所有資料列和資料行。
  • 將結果合併成新的 DataFrame 。

若要使用 groupBy().applyInPandas() ，您必須定義下列專案：

  • 定義每個群組計算的 Python 函式
  • 物件 StructType 或定義輸出架構的字串 DataFrame

傳回 pandas.DataFrame 之 的資料行標籤必須符合所定義輸出架構中的功能變數名稱，如果指定為字串，則比對欄位資料類型，例如整數索引。 請參閱 pandas。DataFrame 說明如何在建構 時標記資料行 pandas.DataFrame 。

套用函式之前，群組的所有資料都會載入記憶體中。 這可能會導致記憶體不足的例外狀況，特別是在群組大小扭曲時。 maxRecordsPerBatch的組態不會套用在群組上，因此您必須確定已分組的資料符合可用的記憶體。

下列範例示範如何使用 groupby().apply() 來減去群組中每個值的平均值。

df = spark.createDataFrame(
    [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
    ("id", "v"))

def subtract_mean(pdf):
    # pdf is a pandas.DataFrame
    v = pdf.v
    return pdf.assign(v=v - v.mean())

df.groupby("id").applyInPandas(subtract_mean, schema="id long, v double").show()
# +---+----+
# | id|   v|
# +---+----+
# |  1|-0.5|
# |  1| 0.5|
# |  2|-3.0|
# |  2|-1.0|
# |  2| 4.0|
# +---+----+

如需詳細使用量，請參閱 pyspark.sql.GroupedData.applyInPandas。

地圖

您可以使用 pandas 實例執行對應作業， DataFrame.mapInPandas() 以便將 的 pandas.DataFrame 反覆運算器轉換成另一個反覆運算器 pandas.DataFrame ，代表目前的 PySpark DataFrame，並以 PySpark DataFrame 傳回結果。

基礎函式會接受 並輸出 的 pandas.DataFrame 反覆運算器。 它可以傳回任意長度的輸出，與某些 Pandas UDF 相較之下，例如數列到數列。

下列範例示範如何使用 mapInPandas() ：

df = spark.createDataFrame([(1, 21), (2, 30)], ("id", "age"))

def filter_func(iterator):
    for pdf in iterator:
        yield pdf[pdf.id == 1]

df.mapInPandas(filter_func, schema=df.schema).show()
# +---+---+
# | id|age|
# +---+---+
# |  1| 21|
# +---+---+

如需詳細的使用方式，請參閱 pyspark.sql.DataFrame.mapInPandas。

群組對應

針對具有 pandas 實例的共同群組對應作業，請使用 DataFrame.groupby().cogroup().applyInPandas() 將兩個 PySpark DataFrame 共同群組在一起，然後將 Python 函式套用至每個共同群組，如下所示：

  • 隨機顯示資料，讓共用金鑰的每個 DataFrame 群組會共同群組在一起。
  • 將函式套用至每個共同群組。 函式的輸入是兩 pandas.DataFrame 個 (，選擇性 Tuple 代表索引鍵) 。 函式的輸出是 pandas.DataFrame 。
  • pandas.DataFrame將所有群組中的 合併為新的 PySpark DataFrame 。

若要使用 groupBy().cogroup().applyInPandas() ，您必須定義下列專案：

  • Python 函式，定義每個共同群組的計算。
  • StructType物件或字串，定義輸出 PySpark DataFrame 的架構。

傳回 pandas.DataFrame 之 的資料行標籤必須符合所定義輸出架構中的功能變數名稱，如果指定為字串，則比對欄位資料類型，例如整數索引。 請參閱 pandas。DataFrame 說明如何在建構 時標記資料行 pandas.DataFrame 。

套用函式之前，會將共同群組的所有資料載入記憶體中。 這可能會導致記憶體不足的例外狀況，特別是在群組大小扭曲時。 未套用 maxRecordsPerBatch 的組態，而且您必須確定已群組的資料符合可用的記憶體。

下列範例示範如何使用 groupby().cogroup().applyInPandas() 在兩個 asof join 資料集之間執行 。

import pandas as pd

df1 = spark.createDataFrame(
    [(20000101, 1, 1.0), (20000101, 2, 2.0), (20000102, 1, 3.0), (20000102, 2, 4.0)],
    ("time", "id", "v1"))

df2 = spark.createDataFrame(
    [(20000101, 1, "x"), (20000101, 2, "y")],
    ("time", "id", "v2"))

def asof_join(l, r):
    return pd.merge_asof(l, r, on="time", by="id")

df1.groupby("id").cogroup(df2.groupby("id")).applyInPandas(
    asof_join, schema="time int, id int, v1 double, v2 string").show()
# +--------+---+---+---+
# |    time| id| v1| v2|
# +--------+---+---+---+
# |20000101|  1|1.0|  x|
# |20000102|  1|3.0|  x|
# |20000101|  2|2.0|  y|
# |20000102|  2|4.0|  y|
# +--------+---+---+---+

如需詳細使用量，請參閱 pyspark.sql.PandasCogroupedOps.applyInPandas。