Python Spark Streaming 仅运行一次

Question

我一直在使用 Python 开发一些 Spark Streaming，特别是 textFileStream，并且我注意到一个有点奇怪的行为。我想知道是否有人可以帮我解释一下。

我目前的代码设置如下:

def fileName(data):
    debug = data.toDebugString()
    pattern = re.compile("file:/.*\.txt")
    files = pattern.findall(debug)
    return files

if __name__ == "__main__":
    sc = SparkContext(appName="PythonStreamingFileNamePrinter")
    ssc = StreamingContext(sc, 1)

    lines = ssc.textFileStream("file:///test/input/")

    files = lines.foreachRDD(fileName)
    print(files)

    ssc.start()
    ssc.awaitTermination()

fileName 函数简单地从调试流中获取正在处理的文件的名称 ( Spark Streaming: How to get the filename of a processed file in Python )。但是，此代码仅运行一次，仅打印一次文件。当我修改函数如下:

def fileName(data):
    debug = data.toDebugString()
    pattern = re.compile("file:/.*\.txt")
    files = pattern.findall(debug)
    print(files);

正如预期的那样，它每秒检查一次目录。似乎唯一“循环”的代码位于 foreachRDD 中。

我的这个假设是否正确，所有处理(包括循环、条件等)都必须发生在映射函数等内部？

谢谢，中号

Answer 1

DStream 由许多随着时间的推移而构建的 rdd 组成。lines 是一个 DStream。

当您对行执行 foreachRDD 时，流中的每个 rdd 都会转换为字符串。因此，当您打印它时，您将获得代表流中所有 rdd 的字符串列表。这意味着，这发生在“流的末尾”。

当您在 fileName 函数中打印字符串时，您正在为正在处理的流中的每个 rdd 执行此操作。因此，您可以在流运行时获取它。

此外，正如我在上一个问题中向您提到的，这里不需要 foreachRDD。它不是满足这种特定需求的“ Spark 流方式”，也许这就是它让您感到困惑的原因。

这里更直接的方法是在 DStream 本身上使用映射(这将影响其中的所有 rdd)，然后使用 pprint。

请记住，与常规 rdd 不同，您不能仅在流中收集(或任何类似的)rdd 并在流运行时返回结果。您需要对这些数据执行一些操作，将其保存到某个外部源(如果需要)或将其作为整个流状态的一部分进行处理。