我正在尝试从 kafka 获取数据到 spark-structured-streaming,但我无法检查我是否做得很好。我想在控制台上打印来自 kafka 的数据,但控制台上什么也没有。可能是因为来自卡夫卡的数据量很大,但我不知道。我正在使用 Windows 10。我检查了 kafka 的端口是由“netstat -an | findstr TARGET_IP”建立的。TARGET_IP 表示kafka生产者的IP。根据以上结果的 PID,我检查了“任务列表/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(); }}
添加回答
举报
0/150
提交
取消