使用comitAsync将结果发送到另一主题的spark流

Question

我正在使用spark streaming文档中提供的策略来致力于Kafka本身。我的流程是这样的：主题A-->火花流[foreachRdd process->send to Topic b]向主题A提交偏移量

    JavaInputDStream<ConsumerRecord<String, Request>> kafkaStream = KafkaUtils.createDirectStream(
            streamingContext,
            LocationStrategies.PreferConsistent(),
            ConsumerStrategies.<String, Request>Subscribe(inputTopics, kafkaParams)
    );

    kafkaStream.foreachRDD(rdd -> {
                OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd).offsetRanges();
                rdd.foreachPartition(
                        consumerRecords -> {
                            OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
                            System.out.println(String.format("$s %d %d $d", o.topic(), o.partition(), o.fromOffset(), o.untilOffset()));
                            consumerRecords.forEachRemaining(record -> doProcess(record));
                        });

                ((CanCommitOffsets) kafkaStream.inputDStream()).commitAsync(offsetRanges);
            }
    );

Answer 1

这就是我想出来的，它接收输入数据，然后使用输出主题发送请求。生产者必须在foreach循环中创建，否则spark将尝试序列化并将其发送给所有工作人员。注意，响应是异步发送的。这意味着我在这个系统中至少使用了一种语义。

kafkaStream.foreachRDD(rdd -> {
        OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
        rdd.foreachPartition(
                partition -> {
                    OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
                    System.out.println(String.format("%s %d %d %d", o.topic(), o.partition(), o.fromOffset(), o.untilOffset()));

                    // Print statements in this section are shown in the executor's stdout logs
                    KafkaProducer<String, MLMIOutput> producer = new KafkaProducer(producerConfig(o.partition()));
                    partition.forEachRemaining(record -> {

                        System.out.println("request: "+record.value());

                        Response data = new  Response …
                        // As as debugging technique, users can write to DBFS to verify that records are being written out
                        // dbutils.fs.put("/tmp/test_kafka_output",data,true)
                        ProducerRecord<String, Response> message = new ProducerRecord(outputTopic, null, data);
                        Future<RecordMetadata> result = producer.send(message);
                        try {
                            RecordMetadata metadata = result.get();
                            System.out.println(String.format("offset='$d' partition='%d' topic='%s'timestamp='$d",
                            metadata.offset(),metadata.partition(),metadata.topic(),metadata.timestamp()));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (ExecutionException e) {
                            e.printStackTrace();
                        }
                    });
                    producer.close();
                });

        ((CanCommitOffsets) kafkaStream.inputDStream()).commitAsync(offsetRanges);
    }

）；