Spark Streaming：打印 JavaInputDStream答案

【问题标题】：Spark Streaming: Print JavaInputDStreamSpark Streaming：打印 JavaInputDStream
【发布时间】：2020-09-10 20:14:19
【问题描述】：

目前我正在使用 Spark Streaming 以及从 Kafka 读取消息的可能性。通过 Kafka Producer，我向一个主题发送消息，并希望在 Spark Streaming 的帮助下阅读该主题。

我使用以下 Java 代码来查询消息：

package apache_spark_streaming;

import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public final class Spark_Kafka_Example {

private static final String BOOTSTRAP_SERVERS_CONNECTION = "XXXXX";
private static final String SPARK_CONNECTION = "spark://XXXXX:7077";
private static final String TOPIC_NAME = "KafkaTesting1";
private static final Set<String> TOPIC_1 = new HashSet<>(Arrays.asList(TOPIC_NAME.split(",")));

public static Map<String, Object> getProperties() {

    try {
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", BOOTSTRAP_SERVERS_CONNECTION);
        kafkaParams.put("key.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class.getName());
        kafkaParams.put("value.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class.getName());
        kafkaParams.put("group.id", "Stream Testing");
        kafkaParams.put("auto.offset.reset", "earliest");
        kafkaParams.put("enable.auto.commit", false);
        return kafkaParams;
    } 
    catch (Exception e) {
        e.printStackTrace();
    }
    return null;
}
public static void main(String[] args) throws Exception {

    // Create context with a 2 seconds batch interval
    SparkConf sparkConf = new SparkConf().setAppName("Kafka Example").setMaster(SPARK_CONNECTION);   
    JavaStreamingContext sc = new JavaStreamingContext(sparkConf, Durations.seconds(2));

    JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(
           sc
           , LocationStrategies.PreferConsistent()
           , ConsumerStrategies.Subscribe(TOPIC_1, getProperties())
    );    

    stream.print();
    sc.start(); 
    sc.awaitTermination();
  }
}

我的问题是我不知道如何在命令行上输出消息。也许我只有一个理解问题如何正确使用 JavaInputDStreams。

目前我仅使用 print() 函数将其作为输出：

17/07/10 16:59:20 INFO JobScheduler：添加的作业时间为 1499698760000 毫秒

希望你能帮我解决这个“问题”。

更新我试过了

stream.foreachRDD(consumerRecordJavaRDD -> {
   consumerRecordJavaRDD.foreach(stringStringConsumerRecord -> {
   //.to get topic name: stringStringConsumerRecord.topic()
   //To get value : stringStringConsumerRecord.value()
  } }

【问题讨论】：

你想输出什么消息？您需要配置记录器。
感谢您的回复。我从 Oracle DB 读取数据并将以下字符串发送到 Kafka（例如）：{214;Bambi Duesseldorf;Duesseldorf;Nordrhein-Westfalen;Perlmuttknopf, Der;2016-09-29 08:41:58.538;10} 怎么能我配置了 Logger？
@hbenner 更新您的答案，而不是添加为答案

标签： java apache-kafka spark-streaming

【解决方案1】：

创建流后，您将提取内容并使用过滤器或映射来处理它。

查看 Spark/java 的流式示例目录：https://github.com/apache/spark/tree/master/examples/src/main/java/org/apache/spark/examples/streaming

JavaQueueStream 举个例子。

【讨论】：

【解决方案2】：

您直接绑定到打印输入流，我无法打印流的实际值。

使用以下代码打印您的输入流，该流由您的 Kafka 生产者生成。

  JavaDStream<String> data = stream.map(v -> {
        return v.value();    // mapping to convert into spark D-Stream 
  });
  
  data.print();

【讨论】：