如何在spark structured streaming with java中检查从kafka获取的数据?

ttisahbt  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(283)

我正试图从Kafka那里获取数据,以激发结构化流媒体,但我无法检查自己是否做得很好。我想在控制台上打印Kafka的数据,但控制台上什么也没有。这可能是因为Kafka的大量数据,但我不知道。
我使用的是windows10。我检查了kafka的端口是由“netstat-an | findstr target | ip”建立的。目标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();
    }
}
8i9zcol2

8i9zcol21#

我测试你的代码,它可以打印。
首先,你应该检查你的Kafka主题,确保有消息在里面。
然后检查你的spark应用程序,确保它可以连接你的Kafka经纪人。

相关问题