Spark Streaming 可以做除字数以外的任何事情吗？答案

【问题标题】：Can Spark Streaming do Anything Other Than Word Count?Spark Streaming 可以做除字数以外的任何事情吗？
【发布时间】：2023-03-07 02:22:01
【问题描述】：

我正在尝试掌握 Spark Streaming，但遇到了困难。尽管阅读了文档并分析了示例，但我希望做的不仅仅是对文本文件/流/Kafka 队列进行字数统计，这是我们唯一可以从文档中理解的内容。

我希望监听传入的 Kafka 消息流，按键分组消息，然后处理它们。下面的代码是该过程的简化版本；从 Kafka 获取消息流，通过 key reduce 对消息进行分组，然后进行处理。

JavaPairDStream<String, byte[]> groupByKeyList = kafkaStream.reduceByKey((bytes, bytes2) -> bytes);

    groupByKeyList.foreachRDD(rdd -> {
        List<MyThing> myThingsList = new ArrayList<>();
        MyCalculationCode myCalc = new MyCalculationCode();

        rdd.foreachPartition(partition -> {
            while (partition.hasNext()) {
                Tuple2<String, byte[]> keyAndMessage = partition.next();
                MyThing aSingleMyThing = MyThing.parseFrom(keyAndMessage._2);   //parse from protobuffer format
                myThingsList.add(aSingleMyThing);
            }
        });

        List<MyResult> results = myCalc.doTheStuff(myThingsList);

        //other code here to write results to file

    });

在调试时，我看到(partition.hasNext()) myThingsList 与外部forEachRDD 中声明的List<MyThing> myThingsList 具有不同的内存地址。

调用List<MyResult> results = myCalc.doTheStuff(myThingsList); 时没有结果，因为myThingsList 是List 的不同实例。

我想要这个问题的解决方案，但更希望参考文档来帮助我理解为什么这不起作用（如预期的那样）以及我如何为自己解决它（我不是指指向Spark 文档的单页，还有部分/段落，或者最好仍然是指向“JavaDoc”的链接，该链接不提供带有非功能注释代码的 Scala 示例。

【问题讨论】：

标签： java apache-kafka spark-streaming

【解决方案1】：

您看到不同列表地址的原因是因为 Spark 没有在驱动程序上本地执行 foreachPartition，它必须序列化函数并将其发送到处理分区处理的 Executor 上。您必须记住，虽然使用代码感觉就像所有东西都在一个位置运行，但计算实际上是分布式的。

我在您的代码中看到的第一个问题与您的reduceByKey 有关，它接受两个字节数组并返回第一个，这真的是您想要做的吗？这意味着您正在有效地删除部分数据，也许您正在寻找combineByKey，这将允许您返回JavaPairDStream<String, List<byte[]>。

关于你的 protobuf 的解析，在我看来你不想要 foreachRDD，你需要一个额外的 map 来解析数据：

kafkaStream
 .combineByKey(/* implement logic */)
 .flatMap(x -> x._2)
 .map(proto -> MyThing.parseFrom(proto))
 .map(myThing -> myCalc.doStuff(myThing))
 .foreachRDD(/* After all the processing, do stuff with result */)

【讨论】：