在 Spark 中加入 JavaPairRDD 和 JavaPairDStream

Question

我有 JavaPairDStream>String, Long> 流 和 JavaPairRDD >String, Long> 批处理。 现在我想加入这两个。基本上我想使用 Spark 将实时数据与批处理数据连接起来。但是直接在参数内使用 JavaPairDStream 来加入会出错。我必须做出哪些改变才能加入这两者。这也是正确的方法还是有另一种方法可以在 spark 中结合批处理视图和实时视图。

例如：stream.leftOuterJoin(batch); 这显示错误。我知道它们是两种不同的数据类型，但由于 JavaPairDStream 是实时 RDD 的抽象，它应该可以工作。

任何关于如何加入批处理视图和实时视图的建议都会有所帮助。谢谢。

对于 JavaPairDStream 和 JavaPairRDD 中的错误 > 深表歉意。我无法找到正确的转义字符来正确编写它。

Answer 1

好的，我已经得到了这里提到的解决方案Transform Operation

变换操作

transform 操作（以及它的变体，如 transformWith）允许在 DStream 上应用任意 RDD-to-RDD 函数。它可用于应用未在 DStream API 中公开的任何 RDD 操作。例如，将数据流中的每个批次与另一个数据集连接的功能并未直接在 DStream API 中公开。但是，您可以轻松地使用转换来执行此操作。这实现了非常强大的可能性。例如，如果您想通过将输入数据流与预先计算的垃圾邮件信息（可能也使用 Spark 生成）连接起来，然后基于它进行过滤来进行实时数据清理。

import org.apache.spark.streaming.api.java.*;
// RDD containing spam information
final JavaPairRDD<String, Double> spamInfoRDD = jssc.sparkContext().newAPIHadoopRDD(...);
JavaPairDStream<String, Integer> cleanedDStream = wordCounts.transform( new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
          @Override public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> rdd) throws Exception {

      rdd.join(spamInfoRDD).filter(...); 
      // join data stream with spam information to do data cleaning
  ...
    }
});