python - 在 python 3.5.2 中，如何优雅地链接对象上未知数量的函数而不是更改类型？

Question

介绍

我不太确定标题是否清楚。我不是以英语为母语的人，所以如果有人对这篇文章的内容有更好的总结，请编辑！

环境

• python 3.5.2
• pyspark 2.3.0

上下文

我有一个火花数据框。这些数据在写入弹性搜索之前会进行转换。

就我而言，我有两个转换。它们是map数据帧 rdd 上的函数。

但是，我不想硬写它们，而是想这样做，以便我可以提供我的函数（处理数据转换）X 函数，这些函数将一个一个地应用于数据帧（对于第一个函数）和/或前一个变换函数的结果。

初步工作

这是以前的状态，不需要，硬写：

df.rdd.map(transfo1) \
        .map(transfo2) \
        .saveAsNewAPIHadoopFile

到目前为止我所拥有的

def write_to_index(self, transformation_functions: list, dataframe):
    // stuff
    for transfo in transformation_functions:
        dataframe = dataframe.rdd.map(transfo)

        dataframe.saveAsNewAPIHadoopFile

但是，这有一个问题：如果第一次转换的返回不是数据帧，它将在循环的第二次迭代中失败，因为生成的对象没有 rdd 属性。

工作解决方案

        object_to_process = dataframe.rdd
        for transfo in transformation_functions:
            object_to_process = object_to_process.map(transfo)

        object_to_process.saveAsNewAPIHadoopFile

上面的解决方案似乎有效（至少会引发任何错误）。但我想知道是否有更优雅的解决方案或任何内置的 python 解决方案。

标签： pythonapache-sparkpysparkrdd

Answer 1

您可以使用此单线：

from functools import reduce

def write_to_index(self, transformation_functions: list, dataframe):
    reduce(lambda x, y: x.map(y), transformation_functions, dataframe.rdd).saveAsNewAPIHadoopFile

如果写得很详细，应该与

dataframe.rdd.map(transformation_functions[0]) \
             .map(transformation_functions[1]) \
             .map(...) \
             .saveAsNewAPIHadoopFile