flume攔截器


攔截器作用:攔截器是簡單的插件式組件,設置在source和channel之間。source接收到的事件,在寫入channel之前,攔截器都可以進行轉換或者刪除這些事件。每個攔截器只處理同一個source接收到的事件。可以自定義攔截器。

flume修改時間戳的插件見 https://github.com/haebin/flume-timestamp-interceptor

 

有一個缺陷是,DateUtils.parseDate(timestamp, dateFormat)里面的dateFormat不支持unix時間戳,只能自己手動添加了

原來是:

  1. String timestamp = get(index, data);
  2. now = DateUtils.parseDate(timestamp, dateFormat).getTime();
  3. headers.put(TIMESTAMP, Long.toString(now));

修改后

  1. String timestamp = get(index, data);
  2. if (dateFormat[0].equals("tsecond")){
  3. now = Long.parseLong(timestamp)*1000;
  4. }
  5. else if(dateFormat[0].equals("tmillisecond")){
  6. now = Long.parseLong(timestamp);
  7. }
  8. else if(dateFormat[0].equals("tnanosecond")){
  9. now = Long.parseLong(timestamp)/1000000;
  10. }
  11. else {
  12. now = DateUtils.parseDate(timestamp, dateFormat).getTime();
  13. }
  14. headers.put(TIMESTAMP, Long.toString(now));

flume配置:

  1. kafka_sn_hive.sources.s1.interceptors = timestamp
  2. kafka_sn_hive.sources.s1.interceptors.timestamp.type = org.apache.flume.interceptor.EventTimestampInterceptor$Builder
  3. kafka_sn_hive.sources.s1.interceptors.timestamp.preserveExisting = false
  4. kafka_sn_hive.sources.s1.interceptors.timestamp.delimiter = ,
  5. kafka_sn_hive.sources.s1.interceptors.timestamp.dateIndex = 4
  6. kafka_sn_hive.sources.s1.interceptors.timestamp.dateFormat = tsecond

表示按逗號做分隔符的第四個(從0開始)字段是一個秒單位的時間戳。

在flume里面,時間戳是毫秒級別的,所以要判斷這個字段是秒還是毫秒納秒

 

見http://lisux.me/lishuai/?p=867


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM