如何在 Spark Streaming 中跨多个批次间隔传输数据流

Question

我正在使用 Apache Spark Streaming 1.6.1 编写一个 Java 应用程序，它连接两个 Key/Value 数据流并将输出写入 HDFS。这两个数据流包含 K/V 个字符串，并使用 textFileStream() 定期从 HDFS 提取到 Spark 中。

两个数据流不同步，这意味着在时间t0 stream1 中的某些键可能会在时间t1 出现在stream2 中，反之亦然。因此，我的目标是连接两个流并计算 "leftover" 键，在下一批间隔中应考虑将其用于连接操作。

为了更好地阐明这一点，请查看以下算法：

variables:
stream1 = <String, String> input stream at time t1
stream2 = <String, String> input stream at time t1
left_keys_s1 = <String, String> records of stream1 that didn't appear in the join at time t0
left_keys_s2 = <String, String> records of stream2 that didn't appear in the join at time t0

operations at time t1:
out_stream = (stream1 + left_keys_s1) join (stream2 + left_keys_s2)
write out_stream to HDFS
left_keys_s1 = left_keys_s1 + records of stream1 not in out_stream (should be used at time t2)
left_keys_s2 = left_keys_s2 + records of stream2 not in out_stream (should be used at time t2)

我尝试使用 Spark Streaming 实现此算法但未成功。最初，我以这种方式为剩余的键创建两个空流（这只是一个流，但生成第二个流的代码类似）：

JavaRDD<String> empty_rdd = sc.emptyRDD(); //sc = Java Spark Context
Queue<JavaRDD<String>> q = new LinkedList<JavaRDD<String>>();
q.add(empty_rdd);
JavaDStream<String> empty_dstream = jssc.queueStream(q);
JavaPairDStream<String, String> k1 = empty_dstream.mapToPair(new PairFunction<String, String, String> () {
                                 @Override
                                 public scala.Tuple2<String, String> call(String s) {
                                   return new scala.Tuple2(s, s);
                                 }
                               });

稍后，这个空流与 stream1 统一（即 union()），最后，在加入之后，我添加了 stream1 的剩余键并调用 window()。 stream2 也是如此。

问题在于生成 left_keys_s1 和 left_keys_s2 的操作是没有操作的转换，这意味着 Spark 不会创建任何 RDD 流图，因此它们永远不会被执行。我现在得到的是一个连接，它只输出键在同一时间间隔内在 stream1 和 stream2 中的记录。

你们有什么关于使用 Spark 正确实现这个的建议吗？

谢谢，马可

Answer 1

通过保留对保存这些值的 RDD 的引用，应该可以将值从一批转移到下一批。

不要尝试使用 queueDStream 合并流，而是声明一个可变的 RDD 引用，它可以在每个流间隔更新。

这是一个例子：

在这个流作业中，我们从一个包含 100 个整数的 RDD 开始。每个间隔，生成 10 个随机数，并减去那些初始的 100 个整数。这个过程一直持续到初始的 100 个元素的 RDD 为空。此示例显示如何将元素从一个间隔转移到下一个间隔。

  import scala.util.Random
  import org.apache.spark.streaming.dstream._

  val ssc = new StreamingContext(sparkContext, Seconds(2))

  var targetInts:RDD[Int] = sc.parallelize(0 until 100)

  var loops = 0

  // we create an rdd of functions that generate random data. 
  // evaluating this RDD at each interval will generate new random data points.
  val randomDataRdd = sc.parallelize(1 to 10).map(_ => () => Random.nextInt(100))

  val dstream = new ConstantInputDStream(ssc, randomDataRdd)

  // create values from the random func rdd

  dataDStream.foreachRDD{rdd => 
                        loops += 1
                        targetInts = targetInts.subtract(rdd)
                        if (targetInts.isEmpty) {println(loops); ssc.stop(false)}
                       }


  ssc.start()

运行这个例子和绘制 loops 对 targetInts.count 给出了以下图表：

我希望这能为您提供足够的指导来实现完整的用例。

如何在 Spark Streaming 中跨多个批次间隔传输数据流

How to carry data streams over multiple batch intervals in Spark Streaming

apache-spark

spark-streaming

dstream