Spark 流中的容错答案

【问题标题】：Fault tolerance in Spark streamingSpark 流中的容错
【发布时间】：2017-11-01 15:54:20
【问题描述】：

我试图在这里解释容错。假设我在 hdfs 中有 1 到 10 个文件，并且 spark 流已经读取了这个文件。现在我的火花流不幸停止了。我在 hdfs 中有文件说 1 到 20，其中 1 到 10 个文件已经被 spark 流解析，并且新添加了 11 到 20 个文件。现在我开始 Spark Streaming，我可以看到文件 1-30。由于我在 hdfs 中的第 21 个文件时启动了 spark，所以我的 spark styreaming 会丢失文件 11-20。我如何丢失文件。我使用文件流。

【问题讨论】：

我使用 fileStream。下面是代码 sn-p.val lines = ssc.fileStream[LongWritable, Text, TextInputFormat]("/home/File")

标签： scala apache-spark hadoop2

【解决方案1】：

Spark Streaming 中 fileStream 的行为是监视一个文件夹并在那里拾取新文件。因此，它只会在进程开始后拾取新文件。为了处理 11-20 的文件，您可能需要在处理开始后重命名它们。

处理这种情况的更好方法是使用像 Kafka 这样的消息队列，您可以在其中继续处理您喜欢的任何点的流： https://blog.cloudera.com/blog/2017/06/offset-management-for-apache-kafka-with-apache-spark-streaming/

【讨论】：

【解决方案2】：

Spark Streaming 还提供checkpointing 的选项。如果启用，该过程将在每个批次开始之前保存检查点（在指定文件夹中）。然后，如果 Spark Streaming 进程由于某种原因崩溃，它可以从最后一个检查点开始。

def createContext(folderName):
    sc = SparkContext(appName='SparkApplication')
    ssc = StreamingContext(sc, 2)  # 2 second window

    ## Your stream configuration here

    ssc.checkpoint(folderName)
    return ssc


ssc = StreamingContext.getOrCreate('/path/to/checkpoint/directory', 
                                   lambda: createContext('/path/to/dir') )
ssc.start()
ssc.awaitTermination()

【讨论】：