如何打印Kafka主题数据在Flink程序？

Question

我通过以下说明创建了一个主题：

C:\kafka_2.12-0.10.2.1>.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test < C:\User11\Desktop\Data.csv

然后，我测试了这个主题是否有正确的数据。之后，我想在Flink程序中打印这个主题。我的计划是：

 try{
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Properties properties = new Properties();
    DataStream<String> stream = env
            .addSource(new FlinkKafkaConsumer09<String>("test", new SimpleStringSchema(),properties));

           stream.print();
    env.execute();
    } catch (Exception e) {
        e.printStackTrace();
    }

但是我得到了这个信息（因为信息太长了，我不得不写一些）：

[main]INFOorg.apache.flink.streaming.api.environment.LocalStream环境-在本地嵌入式Flink迷你集群上运行作业[main]INFOorg.apache.flink.runtime.minicluster.MiniCluster-启动Flink迷你集群[main]INFOorg.apache.flink.runtime.minicluster.MiniCluster-启动指标注册表[main]INFO<--plhd--3[main]INFOorg.apache.flink.runtime.minicluster.MiniCluster-启动RPC服务[flink-akka.actor.default-调度器-2]INFOakka.event.slf4j。Slf4jLogger-Slf4jLogger启动[主]INFOorg.apache.flink.runtime.minicluster.MiniCluster-启动高可用性服务[主]INFOorg.apache.flink.runtime.blob.BobServer-创建BLOB服务器存储目录C：\用户\用户11\AppData\本地\Temp\bobStore-a02ff126-35cc-4c1b-b300-8689d19ff5d2[主]INFO<--plhd-在0.0.0.0:57907启动BLOB服务器-最大并发请求：50-最大积压：1000

此外，我也看到了这个链接，它并没有解决我的问题：如何从flink访问/读取Kafka主题数据？

你能告诉我这里有什么问题吗？

非常感谢。

Answer 1

问题解决了。首先，我用以下命令填充了Kafka主题：

/home/kafka_2.11-2.0.0/bin/kafka-console-producer.sh --broker-list 10.32.0.2:9092,10.32.0.3:9092,10.32.0.4:9092 --topic flinkTopic < transactions2.csv

然后，使用此代码，我可以打印Kafka主题：

 final StreamExecutionEnvironment env = 
 StreamExecutionEnvironment.getExecutionEnvironment();
 Properties prop = new Properties();
 prop.setProperty("bootstrap.servers", 
 "10.32.0.2:9092,10.32.0.3:9092,10.32.0.4:9092");
 prop.setProperty("group.id", "test");
    FlinkKafkaConsumer<String> myConsumer= new FlinkKafkaConsumer<> 
  ("flinkTopic", new SimpleStringSchema(),prop);
    myConsumer.setStartFromEarliest();
    DataStream<String> stream = env.addSource(myConsumer);
    stream.print();
    env.execute("Flink Streaming Java API Skeleton");

我希望这对其他人有用。