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。
