1

我想从 Kafka 接收到 Flink 程序的最新数据，但是 Flink 正在读取历史数据。我已经设置auto.offset.reset为latest如下所示，但它没有工作

properties.setProperty("auto.offset.reset", "latest");

Flink Programm 正在使用以下代码从 Kafka 接收数据

//getting stream from Kafka and giving it assignTimestampsAndWatermarks

        DataStream<JoinedStreamEvent> raw_stream = envrionment.addSource(new FlinkKafkaConsumer09<JoinedStreamEvent>("test",
                new JoinSchema(), properties)).assignTimestampsAndWatermarks(new IngestionTimeExtractor<>());

我正在关注 https://issues.apache.org/jira/browse/FLINK-4280上的讨论，这建议以下面提到的方式添加源

Properties props = new Properties();
...

FlinkKafkaConsumer kafka = new FlinkKafkaConsumer("topic", schema, props);
kafka.setStartFromEarliest();
kafka.setStartFromLatest();
kafka.setEnableCommitOffsets(boolean); // if true, commits on checkpoint if checkpointing is enabled, otherwise, periodically.
kafka.setForwardMetrics(boolean);
...

env.addSource(kafka)

我做了同样的事情，但是我无法访问setStartFromLatest()

 FlinkKafkaConsumer09 kafka = new FlinkKafkaConsumer09<JoinedStreamEvent>( "test", new JoinSchema(),properties);

我应该怎么做才能接收发送到 Kafka 的最新值而不是从历史中接收值？

4

2 回答 2

0

通过为发送者和消费者创建新group id名称test1并保持主题名称与test.

现在我想知道，这是解决此问题的最佳方法吗？因为每次我需要给一个新的组 id

有什么方法可以读取发送到 Kafka 的数据吗？

于 2017-08-23T18:43:06.290 回答
0

我相信这对你有用。它对我有用。修改属性和您的 kafka 主题。

public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "ip:port");
        properties.setProperty("zookeeper.connect", "ip:port");
        properties.setProperty("group.id", "your-group-id");


        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer09<>("your-topic", new SimpleStringSchema(), properties));



        stream.writeAsText("your-path", FileSystem.WriteMode.OVERWRITE)
            .setParallelism(1);

        env.execute();

    }
于 2017-08-28T07:38:41.367 回答