如何检查我使用 Java 在 Spark-structured-streaming 中从 Kafka 获取数据？

Question

我正在尝试从 kafka 获取数据到 spark-structured-streaming，但我无法检查我是否做得很好。我想在控制台上打印来自 kafka 的数据，但控制台上什么也没有。这可能是因为来自 Kafka 的数据量很大，但我不知道。

我正在使用 Windows 10. 我检查了 kafka 的端口是由 "netstat -an | findstr TARGET_IP"。 TARGET_IP表示kafka生产者的IP。根据以上结果的 PID，我检查了 "tasklist /FI "PID eq 5406""。 5406是java.exe的PID，PID 5406占用的内存在不断增加

public static void main( String[] args ) {
    SparkSession spark = SparkSession.builder()
            .master("local")
            .appName("App").getOrCreate();
    Dataset<Row> df = spark
            .readStream()
            .format("kafka")
            .option("kafka.bootstrap.servers", "TARGET_IP:TARGET_PORT")
            .option("subscribe", "TARGET_TOPIC")
            .option("startingOffsets", "earliest")
            .load();
    df.printSchema();
    StreamingQuery queryone = df.writeStream().trigger(Trigger.ProcessingTime(1000)).format("console").start();
    try {
        queryone.awaitTermination();
    } catch (StreamingQueryException e) {
        e.printStackTrace();
    }
}

Answer 1

我测试了你的代码，可以打印。

首先你应该检查你的 kafka topic，确保里面有消息。

然后检查你的spark app，确保它可以连接你的kafka broker。

如何检查我使用 Java 在 Spark-structured-streaming 中从 Kafka 获取数据？

How can I check I get data from Kafka in Spark-structured-streaming with Java?

java

apache-kafka

apache-spark

spark-structured-streaming