将自定义函数应用于 spark 数据框组答案

【问题标题】：Apply a custom function to a spark dataframe group将自定义函数应用于 spark 数据框组
【发布时间】：2017-01-28 17:40:24
【问题描述】：

我有一个非常大的包含这些列的时间序列数据表：

LicensePlate/UberRide 数据的每个集合都应考虑整个数据集进行处理。换句话说，我不需要逐行处理数据，而是按 (LicensePlate/UberRide) 分组的所有行一起处理。

我计划将 spark 与 dataframe api 一起使用，但我对如何对 spark 分组数据帧执行自定义计算感到困惑。

我需要做的是：

如何执行第 3 步和第 4 步？关于我应该使用哪个 spark API（dataframe、dataset、rdd、也许是 pandas...）的任何提示？

整个工作流程如下：

【问题讨论】：

Pandas 不是 Spark 的一部分，您可以使用 DataFrame 但您将使用 have to do it in Scala 和 add Python wrapper，RDD 应该可以正常工作。
不能直接使用Spark吗？我正在使用 Spark 1.6.2
如果你指的是 PySpark，那么就像我说的那样 - RDD 应该可以正常工作。
在向新手提问之前，我会进一步调查究竟如何使用 rdds... :-)
@NischalHp : df.rdd.keyBy(lambda x: (x['key1'], x['key2'])) \ .groupByKey() \ .map(lambda groupped_data: my_map_fn(分组数据））

【解决方案1】：

虽然 Spark 提供了一些与 Pandas 集成的方法，但它并没有使 Pandas 成为分布式的。因此，无论您在 Spark 中使用 Pandas 做什么都是简单的本地操作（在转换中使用时对驱动程序或执行程序）。

如果您正在寻找具有类似 Pandas API 的分布式系统，您应该查看dask。
You can define User Defined Aggregate functions or Aggregators 处理分组 Datasets 但 API 的这一部分只能在 Scala 中直接访问。创建一个时write a Python wrapper 并不难。
RDD API 提供了许多函数，可用于分组执行操作，从低级 repartition / repartitionAndSortWithinPartitions 开始，以许多 *byKey 方法结束（combineByKey、groupByKey、@ 987654334@等）。

哪一个适用于您的情况取决于您要应用的函数的属性（它是否具有关联性和可交换性，是否可以在流上工作，是否需要特定的顺序）。

最通用但效率低的方法可以总结如下：
```
h(rdd.keyBy(f).groupByKey().mapValues(g).collect())
```
其中f 从值映射到key，g 对应于每组聚合，h 是最终合并。大多数时候你可以做得比这更好，所以它应该只作为最后的手段。
相对复杂的逻辑可以用DataFrames/Spark SQL和window functions来表达。
另见Applying UDFs on GroupedData in PySpark (with functioning python example)

【讨论】：

【解决方案2】：

您正在寻找的东西自 Spark 2.3 以来就存在：Pandas 矢量化 UDF。它允许对 DataFrame 进行分组并使用 pandas 应用自定义转换，分布在每个组中：

df.groupBy("groupColumn").apply(myCustomPandasTransformation)

但是，我还不知道在 Scala 中进行分组转换的实用方法，因此欢迎提供任何其他建议。

编辑：在 Scala 中，您可以使用 Dataset 的 groupByKey + mapGroups/flatMapGroups 实现与 Spark 早期版本相同的功能。

【讨论】：