如何将 Spark 数据帧转换为 Pandas 并返回 Kedro？

Posted 2023-04-15

【中文标题】如何将 Spark 数据帧转换为 Pandas 并返回 Kedro？

【英文标题】：How to convert Spark data frame to Pandas and back in Kedro?

【发布时间】：2019-11-11 19:33:01

【问题描述】：

我试图了解在 Kedro 中将来自一个节点的 Spark 数据帧转换为作为另一个节点输入所需的 Pandas 而不创建冗余转换步骤的最佳方法是什么。

【参考方案1】：

Kedro 目前为此支持 2 种策略：

使用Transcoding 功能

这需要在catalog.yml 中为同一数据集定义两个DataCatalog 条目，以通用格式（Parquet、JSON、CSV 等）处理同一文件：

my_dataframe@spark:
  type: kedro.contrib.io.pyspark.SparkDataSet
  filepath: data/02_intermediate/data.parquet

my_dataframe@pandas:
  type: ParquetLocalDataSet
  filepath: data/02_intermediate/data.parquet

然后像这样在管道中使用它们：

Pipeline([
    node(my_func1, "spark_input", "my_dataframe@spark"),
    node(my_func2, "my_dataframe@pandas", "output"),
])

在这种情况下，kedro 理解my_dataframe 在两种情况下都是同一个数据集，并正确解析节点执行顺序。同时kedro会使用SparkDataSet实现保存，ParquetLocalDataSet实现加载，所以第一个节点应该输出pyspark.sql.DataFrame，而第二个节点会收到pandas.Dataframe。

使用Pandas to Spark 和Spark to Pandas 节点装饰器

注意：Spark <-> Pandas 内存转换是 notorious 的内存需求，因此只有在已知数据帧很小的情况下这是一个可行的选择。

可以按照文档装饰节点：

from spark import get_spark
from kedro.contrib.decorators import pandas_to_spark

@pandas_to_spark(spark_session)
def my_func3(data):
    data.show() # data is pyspark.sql.DataFrame

甚至是整个管道：

Pipeline([
    node(my_func4, "pandas_input", "some_output"),
    ...
]).decorate(pandas_to_spark)