摘要：最近在用做流上的異常檢測，期間遇到了很多問題，但是發(fā)現(xiàn)網(wǎng)上相關(guān)的資料很少，基本只能自己啃文檔和瞎嘗試。其中如有錯(cuò)漏，歡迎指出。即從一條數(shù)據(jù)中獲得時(shí)間戳，然后以的格式返回。丟棄掉中的附加信息使用這一設(shè)置時(shí)，得到的中的元素是的和組成的鍵值對。
最近在用Apache beam做流上的異常檢測，期間遇到了很多問題，但是發(fā)現(xiàn)網(wǎng)上相關(guān)的資料很少，基本只能自己啃文檔和瞎嘗試。
所以想把自己踩過的坑記錄下來，希望能對大家有所幫助。
其中如有錯(cuò)漏，歡迎指出。
KafkaIO
顧名思義，是從kafka上讀取數(shù)據(jù)到beam上或者將beam上的數(shù)據(jù)寫入到kafka中。官方文檔中沒有直接的教程，要在GitHub上的源碼中找到相關(guān)使用說明。
Github上的Kafka源碼
這里僅說明讀數(shù)據(jù)部分。
maven依賴示例
org.apache.beam
beam-sdks-java-io-kafka
...
讀數(shù)據(jù)示例
PCollection> lines = //這里kV后說明kafka中的key和value均為String類型
p.apply(KafkaIO.read()
.withBootstrapServers("hadoop1:9092, hadoop2:9092")//必需，設(shè)置kafka的服務(wù)器地址和端口
.withTopic("mytopic")//必需，設(shè)置要讀取的kafka的topic名稱
.withKeyDeserializer(StringDeserializer.class)//必需
.withValueDeserializer(StringDeserializer.class)//必需
.withMaxNumRecords(301)
.withTimestampFn(new MyTimestampFunction())
.updateConsumerProperties(ImmutableMap.of("auto.offset.reset", "earliest"))
.withoutMetadata()
)
以下分別后面非必需的一些設(shè)置
1.設(shè)置最大記錄條數(shù)
.withMaxNumRecords(301)
通過這個(gè)函數(shù)，可以設(shè)置最大讀取的記錄條數(shù)。
2.設(shè)置PCollection中元素對應(yīng)的時(shí)間戳
.withTimestampFn(new MyTimestampFunction())
當(dāng)不進(jìn)行這個(gè)設(shè)置的時(shí)候，beam會(huì)根據(jù)當(dāng)前的系統(tǒng)時(shí)間為每個(gè)元素分配一個(gè)時(shí)間戳。
而有的時(shí)候，我們希望用kafka的數(shù)據(jù)中自身帶有的時(shí)間戳來作為PCollection中元素的時(shí)間戳，從而進(jìn)行后續(xù)的窗口操作。這時(shí)就需要通過上面的函數(shù)來達(dá)到這一目的。
其中MyTimestampFunction()是我們自定義的一個(gè)函數(shù)，其要實(shí)現(xiàn)SerializableFunction, Instant>這個(gè)接口。
即從一條kafka數(shù)據(jù)中獲得時(shí)間戳，然后以Instant（org.joda.time.Instant）的格式返回。
public class MyTimestampFunction implements SerializableFunction, Instant> {
public Instant apply(KV input){
String[] temps = input.getValue().split(",");
DateTime t = new DateTime(Long.valueOf(temps[1]));
return t.toInstant();
}
}
3.設(shè)置讀kafka數(shù)據(jù)的順序
updateConsumerProperties(ImmutableMap.of("auto.offset.reset", "earliest"))
KafkaIO默認(rèn)的數(shù)據(jù)讀取順序是從最新的數(shù)據(jù)開始。當(dāng)我們開發(fā)測試的時(shí)候，如果沒有一個(gè)生產(chǎn)者同步向kafka生產(chǎn)數(shù)據(jù)，那么這里就拿不到數(shù)據(jù)。（在這坑了很久，才發(fā)現(xiàn)這個(gè)原因...）
當(dāng)我們想實(shí)現(xiàn)類似于kafka shell中的--from-beginning的功能的時(shí)候，即從最早的數(shù)據(jù)開始讀，就需要進(jìn)行這一設(shè)置。
這里不僅可以改變讀取數(shù)據(jù)的順序，按照類似的方式，還可以進(jìn)行其他設(shè)置。
4.丟棄掉kafka中的附加信息
.withoutMetadata()
使用這一設(shè)置時(shí)，得到的PCollection中的元素是kafka的key和value組成的鍵值對。
當(dāng)不使用其時(shí)，得到的PCollection中的元素是KafkaRecord。會(huì)附件很多元數(shù)據(jù)。
5.其他設(shè)置
// custom function for watermark (default is record timestamp)
* .withWatermarkFn(new MyWatermarkFunction())
*
* // restrict reader to committed messages on Kafka (see method documentation).
* .withReadCommitted()
*
在源碼的使用說明中還提到另外的兩個(gè)設(shè)置，但因?yàn)闀簳r(shí)沒用到，這里就暫且省略了。
文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/35904.html
