如何设计 Spark 程序来处理 300 个最新文件？

Question

情况

新的小文件会定期出现。我需要对最近的 300 个文件进行计算。所以基本上有一个前进的窗口。窗口大小为 300，需要对窗口进行计算。

但是要知道的非常重要的一点是，这不是火花流计算。因为在火花流中，窗口的单位/范围是时间。这里的单位/范围是文件数。

解决方案1

我会维护一个dict，dict的大小是300。每个新文件进来，我把它变成spark数据框，放入dict。然后我确保 dict 中最旧的文件被弹出 如果 dict 的长度超过 300。 在此之后，我会将 dict 中的所有数据帧合并为一个更大的数据帧并进行计算。

上述过程将循环运行。每次新文件进来时，我们都会循环。

解决方案 1 的伪代码

for file in file_list:
    data_frame = get_data_frame(file)
    my_dict[ timestamp ] = data_frame

    for timestamp in my_dict.keys():
        if timestamp older than 24 hours:
            # not only unpersist, but also delete to make sure the memory is released 
            my_dict[timestamp].unpersist
            del my_dict[ timestamp ] 

    # pop one data frame from the dict
    big_data_frame = my_dict.popitem()

    for timestamp in my_dict.keys():
        df = my_dict.get( timestamp )
        big_data_frame = big_data_frame.unionAll(df)

    # Then we run SQL on the big_data_frame to get report

解决方案 1 的问题

总是达到内存不足或gc开销限制

问题

您是否发现解决方案 1 中有任何不当之处？

有没有更好的解决方案？

这是使用 spark 的正确情况吗？

Answer 1

一个观察，你可能不想使用popitem，Python 字典的键没有排序，所以你不能保证你弹出最早的项目。相反，我每次都会使用排序的时间戳列表重新创建字典。假设您的文件名只是时间戳：

my_dict = {file:get_dataframe(file) for file in sorted(file_list)[-300:]}

不确定这是否能解决您的问题，您能否将错误的完整堆栈跟踪粘贴到问题中？您的问题可能发生在 Spark 合并/连接中（未包含在您的问题中）。

Answer 2

我对此的建议是流式传输，但不是关于时间，我的意思是你仍然会设置一些窗口和滑动间隔，但说它是 60 秒。

因此，您每 60 秒就会在“x”分区中获得文件内容的 DStream。这些“x”分区代表您拖放到 HDFS 或文件系统上的文件。 因此，通过这种方式，您可以跟踪已读取的文件/分区数量，如果它们少于 300，则等到它们变为 300。当计数达到 300 后，您就可以开始处理了。

Answer 3

如果可以跟踪最新的文件，或者可以偶尔发现它们，那么我建议做类似的事情

sc.textFile(','.join(files));

或者如果可以识别特定模式来获取这 300 个文件，那么

sc.textFile("*pattern*");

甚至可以有逗号分隔的模式，但可能会发生一些匹配多个模式的文件，而不是一次读取。