1、概述
flink提供了一個特有的kafka connector去讀寫kafka topic的數(shù)據(jù)。flink消費kafka數(shù)據(jù)，并不是完全通過跟蹤kafka消費組的offset來實現(xiàn)去保證exactly-once的語義，而是flink內(nèi)部去跟蹤offset和做checkpoint去實現(xiàn)exactly-once的語義
flink與kafka整合，相應版本對于的maven依賴如下表
maven依賴舉例
<flink.version>1.7.0</flink.version>
<scala.binary.version>2.11</scala.binary.version>
<scala.version>2.11.12</scala.version>
<dependency>
? <groupId>org.apache.flink</groupId>
? <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
? <version>${flink.version}</version>
? <scope>provided</scope>
</dependency>
2、代碼整合
2.1、添加source:Kafka Consumer
flink利用FlinkKafkaConsumer來讀取訪問kafka, 根據(jù)kafka版本不同F(xiàn)linkKafkaConsumer的類名也會變化，會變?yōu)镕linkKafkaConsumer
[08,09,10...]后面的數(shù)字就是對于的kafka的大版本號 。
初始化FlinkKafkaConsumer 需要如下參數(shù)
1、topic名字，用來指定消費一個或者多個topic的數(shù)據(jù)
2、kafka的配置信息，如zk地址端口，kafka地址端口等
3、反序列化器(schema)，對消費數(shù)據(jù)選擇一個反序列化器進行反序列化。
flink kafka的消費端需要知道怎么把kafka中消息數(shù)據(jù)反序列化成java或者scala中的對象。用戶通過使用DeserializationSchema，每一條kafka的消息都會作用于DeserializationSchema的eserialize(byte[] message)方法。來將kafka的消息轉(zhuǎn)換成用戶想要的結(jié)構(gòu)。
用戶通過自定義schema將接入數(shù)據(jù)轉(zhuǎn)換成自定義的數(shù)據(jù)結(jié)構(gòu)，主要通過實現(xiàn)KeyedDeserializationSchema或者DeserializationSchema接口來完成，可以自定義。flink內(nèi)置的 對DeserializationSchema 的實現(xiàn)有
public class SimpleStringSchema implements DeserializationSchema<String>
public class TypeInformationSerializationSchema<T> implements DeserializationSchema<T>
對 KeyedDeserializationSchema的實現(xiàn)有
public class TypeInformationKeyValueSerializationSchema<K, V> implements KeyedDeserializationSchema<Tuple2<K, V>>
public class JSONKeyValueDeserializationSchema implements KeyedDeserializationSchema<ObjectNode>
例如：
val myConsumer = new FlinkKafkaConsumer010[String]("topic",new SimpleStringSchema,p)
2.2、自定義schema舉例
public class MySchema implements KeyedDeserializationSchema<KafkaMsgDTO> {
? ? @Override
? ? public KafkaMsgDTO deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
? ? ? ? String msg = new String(message, StandardCharsets.UTF_8);
? ? ? ? String key = null;
? ? ? ? if(messageKey != null){
? ? ? ? ? ? key = new String(messageKey, StandardCharsets.UTF_8);
? ? ? ? }
? ? ? ? return new KafkaMsgDTO(msg,key,topic,partition,offset);
? ? }
? ? @Override
? ? public boolean isEndOfStream(KafkaMsgDTO nextElement) {
? ? ? ? return false;
? ? }
? ? @Override
? ? public TypeInformation<KafkaMsgDTO> getProducedType() {
? ? ? ? return getForClass(KafkaMsgDTO.class);
? ? }
}
<dependency>
? <groupId>org.apache.flink</groupId>
? <artifactId>flink-connector-kafka-base_2.11</artifactId>
? <version>1.7.0</version>
</dependency>
public class KafkaMsgDTO {
? ? private String topic;
? ? private int partition;
? ? private long offset;
? ? private String mesg;
? ? @Override
? ? public String toString() {
? ? ? ? return "KafkaMsgDTO{" +
? ? ? ? ? ? ? ? "topic='" + topic + '\'' +
? ? ? ? ? ? ? ? ", partition=" + partition +
? ? ? ? ? ? ? ? ", offset=" + offset +
? ? ? ? ? ? ? ? ", mesg='" + mesg + '\'' +
? ? ? ? ? ? ? ? ", key='" + key + '\'' +
? ? ? ? ? ? ? ? '}';
? ? }
? ? private String key;
? ? public KafkaMsgDTO(){
? ? }
? ? public KafkaMsgDTO(String mesg,String key,String topic,int partition,long offset){
? ? ? ? this.mesg = mesg;
? ? ? ? this.key = key;
? ? ? ? this.topic = topic;
? ? ? ? this.partition = partition;
? ? ? ? this.offset = offset;
? ? }
? ? public String getKey() {
? ? ? ? return key;
? ? }
? ? public void setKey(String key) {
? ? ? ? this.key = key;
? ? }
? ? public String getTopic() {
? ? ? ? return topic;
? ? }
? ? public void setTopic(String topic) {
? ? ? ? this.topic = topic;
? ? }
? ? public int getPartition() {
? ? ? ? return partition;
? ? }
? ? public void setPartition(int partition) {
? ? ? ? this.partition = partition;
? ? }
? ? public long getOffset() {
? ? ? ? return offset;
? ? }
? ? public void setOffset(long offset) {
? ? ? ? this.offset = offset;
? ? }
? ? public String getMesg() {
? ? ? ? return mesg;
? ? }
? ? public void setMesg(String mesg) {
? ? ? ? this.mesg = mesg;
? ? }
}
2.3、指定offset位置進行消費
val myConsumer = new FlinkKafkaConsumer010[KafkaMsgDTO]("topic",new MySchema(),p)
//? ? myConsumer.setStartFromEarliest()? ? ?
//從最早開始消費，消費過的數(shù)據(jù)會重復消費，從kafka來看默認不提交offset.
//? ? myConsumer.setStartFromLatest()? ? ? ?
//從最新開始消費，不消費流啟動前未消費的數(shù)據(jù)，從kafka來看默認不提交offset.
? ? ? myConsumer.setStartFromGroupOffsets()
//從消費的offset位置開始消費，kafka有提交offset，這是默認消費方式
//如果沒有做checkpoint 數(shù)據(jù)進入sink就會提交offset,如果sink里面邏輯失敗。offset照樣會提交，程序退出，如果重啟流，消費失敗的數(shù)據(jù)不會被重新消費
//如果做了checkpoint 會保證數(shù)據(jù)的端到端精準一次消費。sink里面邏輯失敗不會提交offset
2.4、checkpointing
env.enableCheckpointing(5000);
val stream = env.addSource(myConsumer)
2.5、sink邏輯
stream.addSink(x=>{
? println(x)
? println(1/(x.getMesg.toInt%2))//消息是偶數(shù)就會報錯，分母為0
? println(x)
})
val stream = env.addSource(myConsumer)
//實驗表明如果sink處理邏輯有一部線程在跑，如果異步線程失敗。offset照樣會提交。
stream.addSink(x=>{
? println(x)
? new Thread(new Runnable {
? ? override def run(): Unit = {
? ? ? println(1/(x.getMesg.toInt%2))//消息是偶數(shù)就會報錯，分母為0
? ? }
? }).start()
? println(x)
})
2.6、指定到某個topic的offset位置進行消費
val specificStartOffsets = new java.util.HashMap[KafkaTopicPartition, java.lang.Long]()
specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L)
specificStartOffsets.put(new KafkaTopicPartition("myTopic", 1), 31L)
specificStartOffsets.put(new KafkaTopicPartition("myTopic", 2), 43L)
myConsumer.setStartFromSpecificOffsets(specificStartOffsets)
