pandas - 将 Spark 数据帧转换为 Pandas/R 数据帧的要求

Question

我在 Hadoop 的 YARN 上运行 Spark。这种转换是如何进行的？ collect() 是否在转换之前发生？

我还需要在每个从属节点上安装 Python 和 R 才能进行转换吗？我正在努力寻找这方面的文档。

Answer 1

toPandas (PySpark)/as.data.frame (SparkR)

必须在创建本地数据框之前收集数据。例如toPandas方法如下所示:

def toPandas(self):
    import pandas as pd
    return pd.DataFrame.from_records(self.collect(), columns=self.columns)

您需要在每个节点上安装 Python，并以最佳方式安装所有依赖项。

SparkR 对应物(as.data.frame)只是collect 的别名。

总而言之，在这两种情况下，数据都被收集到驱动程序节点并转换为本地数据结构(pandas.DataFrame 和base::data.frame 分别在 Python 和 R 中)。

向量化的用户定义函数

自 Spark 2.3.0 PySpark 还提供了一组 pandas_udf (SCALAR, GROUPED_MAP, GROUPED_AGG) 对由

定义的数据 block 并行操作

• SCALAR 变体情况下的分区
• GROUPED_MAP 和 GROUPED_AGG 情况下的分组表达式。

每个 block 由

表示

• 在 SCALAR 和 GROUPED_AGG 变体的情况下，一个或多个 pandas.core.series.Series。
• 在 GROUPED_MAP 变体的情况下，单个 pandas.core.frame.DataFrame。

同样，从 Spark 2.0.0 开始，SparkR 提供了 dapply和 gapply在 data.frames 上运行的函数分别由分区和分组表达式定义。

上述功能:

• 不要向司机收钱。除非数据仅包含单个分区(即使用 coalesce(1))或分组表达式很简单(即 groupBy(lit(1)))，否则不存在单节点瓶颈.
• 将相应的 block 加载到相应执行程序的内存中。因此，它受到每个执行程序上可用的单个 block /内存大小的限制。