摘要：最近在用做流上的异常检测，期间遇到了很多问题，但是发现网上相关的资料很少，基本只能自己啃文档和瞎尝试。其中如有错漏，欢迎指出。即从一条数据中获得时间戳，然后以的格式返回。丢弃掉中的附加信息使用这一设置时，得到的中的元素是的和组成的键值对。
最近在用Apache beam做流上的异常检测，期间遇到了很多问题，但是发现网上相关的资料很少，基本只能自己啃文档和瞎尝试。
所以想把自己踩过的坑记录下来，希望能对大家有所帮助。
其中如有错漏，欢迎指出。
KafkaIO
顾名思义，是从kafka上读取数据到beam上或者将beam上的数据写入到kafka中。官方文档中没有直接的教程，要在GitHub上的源码中找到相关使用说明。
Github上的Kafka源码
这里仅说明读数据部分。
maven依赖示例
org.apache.beam
beam-sdks-java-io-kafka
...
读数据示例
PCollection> lines = //这里kV后说明kafka中的key和value均为String类型
p.apply(KafkaIO.read()
.withBootstrapServers("hadoop1:9092, hadoop2:9092")//必需，设置kafka的服务器地址和端口
.withTopic("mytopic")//必需，设置要读取的kafka的topic名称
.withKeyDeserializer(StringDeserializer.class)//必需
.withValueDeserializer(StringDeserializer.class)//必需
.withMaxNumRecords(301)
.withTimestampFn(new MyTimestampFunction())
.updateConsumerProperties(ImmutableMap.of("auto.offset.reset", "earliest"))
.withoutMetadata()
)
以下分别后面非必需的一些设置
1.设置最大记录条数
.withMaxNumRecords(301)
通过这个函数，可以设置最大读取的记录条数。
2.设置PCollection中元素对应的时间戳
.withTimestampFn(new MyTimestampFunction())
当不进行这个设置的时候，beam会根据当前的系统时间为每个元素分配一个时间戳。
而有的时候，我们希望用kafka的数据中自身带有的时间戳来作为PCollection中元素的时间戳，从而进行后续的窗口操作。这时就需要通过上面的函数来达到这一目的。
其中MyTimestampFunction()是我们自定义的一个函数，其要实现SerializableFunction, Instant>这个接口。
即从一条kafka数据中获得时间戳，然后以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.设置读kafka数据的顺序
updateConsumerProperties(ImmutableMap.of("auto.offset.reset", "earliest"))
KafkaIO默认的数据读取顺序是从最新的数据开始。当我们开发测试的时候，如果没有一个生产者同步向kafka生产数据，那么这里就拿不到数据。（在这坑了很久，才发现这个原因...）
当我们想实现类似于kafka shell中的--from-beginning的功能的时候，即从最早的数据开始读，就需要进行这一设置。
这里不仅可以改变读取数据的顺序，按照类似的方式，还可以进行其他设置。
4.丢弃掉kafka中的附加信息
.withoutMetadata()
使用这一设置时，得到的PCollection中的元素是kafka的key和value组成的键值对。
当不使用其时，得到的PCollection中的元素是KafkaRecord。会附件很多元数据。
5.其他设置
// custom function for watermark (default is record timestamp)
* .withWatermarkFn(new MyWatermarkFunction())
*
* // restrict reader to committed messages on Kafka (see method documentation).
* .withReadCommitted()
*
在源码的使用说明中还提到另外的两个设置，但因为暂时没用到，这里就暂且省略了。
文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。
转载请注明本文地址：https://www.ucloud.cn/yun/35904.html
