默認(rèn)：從topic中指定的group上次消費(fèi)的位置開始消費(fèi)。

所以必須配置group.id參數(shù)從消費(fèi)者組提交的偏移量開始讀取分區(qū)（kafka或zookeeper中）。如果找不到分區(qū)的偏移量，auto.offset.reset將使用屬性中的設(shè)置。如果是默認(rèn)行為(setStartFromGroupOffsets),那么任務(wù)從檢查點(diǎn)重啟，按照重啟前的offset進(jìn)行消費(fèi)，如果直接重啟不從檢查點(diǎn)重啟并且group.id不變，程序會按照上次提交的offset的位置繼續(xù)消費(fèi)。如果group.id改變了，則程序按照auto.offset.reset設(shè)置的屬性進(jìn)行消費(fèi)。但是如果程序帶有狀態(tài)的算子，還是建議使用檢查點(diǎn)重啟。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(5000); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.setParallelism(1);Properties props = new Properties(); props.setProperty("bootstrap.servers",KAFKA_BROKER); props.setProperty("zookeeper.connect", ZK_HOST); props.setProperty("group.id",GROUP_ID); props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); FlinkKafkaConsumer011<String> consumer = new FlinkKafkaConsumer011<>(TOPIC, new SimpleStringSchema(), props);consumer.setStartFromGroupOffsets();

注意：以下五種方式運(yùn)行時(shí)優(yōu)先級都比KafkaProperties中配置的auto.offset.reset優(yōu)先級高。

方式一 : 指定topic, 指定partition的offset位置

Map<KafkaTopicPartition, Long> offsets = new HashedMap(); offsets.put(new KafkaTopicPartition("topic_name", 0), 11L); offsets.put(new KafkaTopicPartition("topic_name", 1), 22L); offsets.put(new KafkaTopicPartition("topic_name", 2), 33L); consumer.setStartFromSpecificOffsets(offsets);

Map<KafkaTopicPartition, Long> Long參數(shù)指定的offset位置

KafkaTopicPartition構(gòu)造函數(shù)有兩個參數(shù)，第一個為topic名字，第二個為分區(qū)數(shù).

• 如果使用者需要讀取在提供的偏移量映射中沒有指定偏移量的分區(qū)，則它將回退到setStartFromGroupOffsets()該特定分區(qū)的默認(rèn)組偏移行為。
• 當(dāng)作業(yè)從故障中自動恢復(fù)或使用保存點(diǎn)手動恢復(fù)時(shí)，這些起始位置配置方法不會影響起始位置。在恢復(fù)時(shí)，每個Kafka分區(qū)的起始位置由存儲在保存點(diǎn)或檢查點(diǎn)中的偏移量確定。

consumer.setStartFromSpecificOffsets(offsets);

方式二: 從topic中最初的數(shù)據(jù)開始消費(fèi)

consumer.setStartFromEarliest();

方式三: 從指定的時(shí)間戳開始

consumer.setStartFromTimestamp(1559801580000l);

對于每個分區(qū)，時(shí)間戳大于或等于指定時(shí)間戳的記錄將用作起始位置。如果分區(qū)的最新記錄早于時(shí)間戳，則只會從最新記錄中讀取分區(qū)。在此模式下，Kafka中的已提交偏移將被忽略，不會用作起始位置。時(shí)間戳指的是kafka中消息自帶的時(shí)間戳。

方式四： 從最新的數(shù)據(jù)開始消費(fèi)

consumer.setStartFromLatest();

方式五（同一默認(rèn)）

參見： https://mp.weixin.qq.com/s?__biz=MzU5Mzk3MDA3Mw==&mid=2247483866&idx=2&sn=6a3b458caf5bebf0171f9fbd834b7517&chksm=fe09172cc97e9e3a590f5ea2978d078b1b46d94f86bd344173fa69c1d63790b09d2fe173bffb&token=1856795336&lang=zh_CN#rd

總結(jié)

以上是生活随笔為你收集整理的指定开始_Flink-Kafka指定offset的五种方式的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。