Apache Spark Streaming：如何比较 2 个 dstream 中的 2 个数据帧

Question

我是 Apache Spark 的初学者。我正在尝试运行一个接收一些数据的流作业，将其转换为数据帧并运行一些处理，如加入和删除重复项等。现在我必须缓存这个处理过的数据，以便我可以将它附加到下一个 dstream（使用一些联合/连接）并再次进行处理。

我尝试使用 dataframe.cache() 来缓存并在下一批流中重新使用它。

例如，如果 df 是从 dstream 形成的 rdd。

   foreachrdd{
new =df.unionAll(processed)
new.registerTempTable("TableScheme")
sql.( //perform inner join and some other processing)
processed=new
processed.cache();

}

当我们执行 Dataframe.cache 或 Dataframe.persist() 时，我们是在缓存实际数据还是应用的 DAG / 转换？当第二个流到来时，我的程序以

退出

Caused by: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
org.apache.spark.SparkException: Attempted to use BlockRDD[1] at socketTextStream after its blocks have been removed!

Answer 1

cache() 功能执行 DAG 中定义的处理直到 cache() 调用，并适当地存储数据，以便在执行多个操作时不会重复计算。

您想要的是跨流批次的持久性。

有几种方法可以做到这一点：

• streamingContext.remember(Minutes(5)) 保留先前批次的数据 Minutes(5) 的时间量
• windowing在数据中移动一个固定的时间窗口，让你可以对多于一批的数据进行操作
• updateStateByKey & mapWithState 提供跨批次维护和转换状态的机制

您选择的方法很大程度上取决于您的用例。