关于flume中涉及到时间戳的错误解决,Expected timestamp in the Flume even

在搭建flume集群收集日志写入hdfs时发生了下面的错误：

java.lang.NullPointerException: Expected timestamp in the Flume event headers, but it was null

        at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:204)

        at org.apache.flume.formatter.output.BucketPath.replaceShorthand(BucketPath.java:200)

        at org.apache.flume.formatter.output.BucketPath.escapeString(BucketPath.java:396)

        at org.apache.flume.sink.hdfs.HDFSEventSink.process(HDFSEventSink.java:388)

        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)

        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)

        at java.lang.Thread.run(Thread.java:744)

2015-06-18 15:31:41,836 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:160)] Unable to deliver event. Exception follows.

org.apache.flume.EventDeliveryException: java.lang.NullPointerException: Expected timestamp in the Flume event headers, but it was null

        at org.apache.flume.sink.hdfs.HDFSEventSink.process(HDFSEventSink.java:471)

        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)

        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)

        at java.lang.Thread.run(Thread.java:744)

Caused by: java.lang.NullPointerException: Expected timestamp in the Flume event headers, but it was null

        at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:204)

        at org.apache.flume.formatter.output.BucketPath.replaceShorthand(BucketPath.java:200)

        at org.apache.flume.formatter.output.BucketPath.escapeString(BucketPath.java:396)

        at org.apache.flume.sink.hdfs.HDFSEventSink.process(HDFSEventSink.java:388)

        ... 3 more

 

原因是因为写入到hfds时使用到了时间戳来区分目录结构，flume的消息组件event在接受到之后在header中没有发现时间戳参数，导致该错误发生，有三种方法可以解决这个错误；

1、agent1.sources.source1.interceptors = t1

     agent1.sources.source1.interceptors.t1.type = timestamp  为source添加拦截，每条event头中加入时间戳；（效率会慢一些）

2、agent1.sinks.sink1.hdfs.useLocalTimeStamp = true 为sink指定该参数为true （如果客户端和flume集群时间不一致数据时间会不准确）

3、在向source发送event时，将时间戳参数添加到event的header中即可，header是一个map，添加时mapkey为timestamp(推荐使用)