当前位置： 首页 > news >正文

网站分类目录查询,电脑网,广州重点场所,e福州app官方下载文章作者邮箱#xff1a;yugongshiyesina.cn 地址#xff1a;广东惠州 ▲ 本章节目的 ⚪ 了解Interceptor的概念和配置参数#xff1b; ⚪ 掌握Interceptor的使用方法#xff1b; ⚪ 掌握Interceptor的Host Interceptor#xff1b; ⚪ 掌握Interceptor的…文章作者邮箱yugongshiyesina.cn              地址广东惠州 ▲ 本章节目的 ⚪ 了解Interceptor的概念和配置参数 ⚪ 掌握Interceptor的使用方法 ⚪ 掌握Interceptor的Host Interceptor ⚪ 掌握Interceptor的Static Interceptor ⚪ 掌握Interceptor的UUID Interceptor ⚪ 掌握Interceptor的Search And Replace Interceptor ⚪ 掌握Interceptor的Regex Filtering Interceptor ⚪ 掌握Interceptor的Custom Interceptor 一、Timestamp Interceptor 1. 概述

1. Timestamp Interceptor是在headers中来添加一个timestamp字段来标记数据被收集的时间。

2. Timestamp Interceptor结合HDFS Sink可以实现数据按天存储。 2. 配置属性 属性 解释 type timestamp
   3. 案例

3. 编写格式文件添加如下内容 a1.sources s1 a1.channels c1 a1.sinks k1 a1.sources.s1.type netcat a1.sources.s1.bind 0.0.0.0 a1.sources.s1.port 8090

   给Interceptor起名

   a1.sources.s1.interceptors i1

   指定Timestamp Interceptor

   a1.sources.s1.interceptors.i1.type timestamp a1.channels.c1.type memory a1.sinks.k1.type logger a1.sources.s1.channels c1 a1.sinks.k1.channel c1 2. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f in.conf - Dflume.root.loggerINFO,console 4. 数据按天存放

4. 编写格式文件添加如下内容 a1.sources s1 a1.channels c1 a1.sinks k1 a1.sources.s1.type netcat a1.sources.s1.bind hadoop01 a1.sources.s1.port 8090 a1.sources.s1.interceptors i1 a1.sources.s1.interceptors.i1.type timestamp a1.channels.c1.type memory a1.sinks.k1.type hdfs a1.sinks.k1.hdfs.path hdfs://hadoop01:9000/flumedata/date%Y-%m-%d a1.sinks.k1.hdfs.fileType DataStream a1.sinks.k1.hdfs.rollInterval 3600 a1.sources.s1.channels c1 a1.sinks.k1.channel c1 2. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f hdfsin.conf - Dflume.root.loggerINFO,console 二、Host Interceptor 1. 概述

5. Host Interceptor是在headers中添加一个字段host。

6. Host Interceptor可以用于标记数据来源于哪一台主机。 2. 配置属性 属性 解释 type 必须是host
   3. 案例

7. 编写格式文件添加如下内容 a1.sources s1 a1.channels c1 a1.sinks k1 a1.sources.s1.type netcat a1.sources.s1.bind 0.0.0.0 a1.sources.s1.port 8090

   给Interceptor起名

   a1.sources.s1.interceptors i1 i2

   指定Timestamp Interceptor

   a1.sources.s1.interceptors.i1.type timestamp

   指定Host Interceptor

   a1.sources.s1.interceptors.i2.type host a1.channels.c1.type memory a1.sinks.k1.type logger a1.sources.s1.channels c1 a1.sinks.k1.channel c1 2. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f in.conf - Dflume.root.loggerINFO,console 三、Static Interceptor 1. 概述

8. Static Interceptor是在headers中添加指定字段。

9. 可以利用这个Interceptor来标记数据的类型。 2. 配置属性 属性 解释 type 必须是static key 指定在headers中的字段名 value 指定在headers中的字段值
   3. 案例

10. 编写格式文件添加如下内容 a1.sources s1 a1.channels c1 a1.sinks k1 a1.sources.s1.type netcat a1.sources.s1.bind 0.0.0.0 a1.sources.s1.port 8090

    给Interceptor起名

    a1.sources.s1.interceptors i1 i2 i3

    指定Timestamp Interceptor

    a1.sources.s1.interceptors.i1.type timestamp

    指定Host Interceptor

    a1.sources.s1.interceptors.i2.type host

    指定Static Interceptor

    a1.sources.s1.interceptors.i3.type static a1.sources.s1.interceptors.i3.key kind a1.sources.s1.interceptors.i3.value log a1.channels.c1.type memory a1.sinks.k1.type logger a1.sources.s1.channels c1 a1.sinks.k1.channel c1 2. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f in.conf - Dflume.root.loggerINFO,console 四、UUID Interceptor 1. 概述

11. UUID Interceptor是在headers中添加一个id字段。

12. 可以用于标记数据的唯一性。 2. 配置属性 属性 解释 type 必须是org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
    3. 案例

13. 编写格式文件添加如下内容 a1.sources s1 a1.channels c1 a1.sinks k1 a1.sources.s1.type netcat a1.sources.s1.bind 0.0.0.0 a1.sources.s1.port 8090

    给Interceptor起名

    a1.sources.s1.interceptors i1 i2 i3 i4

    指定Timestamp Interceptor

    a1.sources.s1.interceptors.i1.type timestamp

    指定Host Interceptor

    a1.sources.s1.interceptors.i2.type host

    指定Static Interceptor

    a1.sources.s1.interceptors.i3.type static a1.sources.s1.interceptors.i3.key kind a1.sources.s1.interceptors.i3.value log

    指定UUID Interceptor

    a1.sources.s1.interceptors.i4.type org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder a1.channels.c1.type memory a1.sinks.k1.type logger a1.sources.s1.channels c1 a1.sinks.k1.channel c1 2. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f in.conf - Dflume.root.loggerINFO,console 五、Search And Replace Interceptor 1. 概述

14. Search And Replace Interceptor在使用的时候需要指定正则表达式会根据正则表达式的规则将符合正则表达式的数据替换为指定形式的数据。

15. 在替换的时候不会替换headers中的数据而是会替换body中的数据。 2. 配置属性 属性 解释 type 必须是search_replace searchPattern 指定要匹配的正则形式 replaceString 指定要替换的字符串
    3. 案例

16. 编写格式文件添加如下内容 a1.sources s1 a1.channels c1 a1.sinks k1 a1.sources.s1.type http a1.sources.s1.port 8090

    给拦截器起名

    a1.sources.s1.interceptors i1

    指定类型

    a1.sources.s1.interceptors.i1.type search_replace a1.sources.s1.interceptors.i1.searchPattern [0-9] a1.sources.s1.interceptors.i1.replaceString * a1.channels.c1.type memory a1.sinks.k1.type logger a1.sources.s1.channels c1 a1.sinks.k1.channel c1 2. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f searchin.conf - Dflume.root.loggerINFO,console 六、Regex Filtering Interceptor 1. 概述

17. Regex Filtering Interceptor在使用的时候需要指定正则表达式。

18. 属性excludeEvents的值如果不指定默认是false。

19. 如果没有配置excludeEvents的值或者配置excludeEvents的值配置为false则只有符合正则表达式的数据会留下来其他不符合正则表达式的数据会被过滤掉如果excludeEvents的值那么符合正则表达式的数据会被过滤掉其他的数据则会被留下来。 2. 配置属性 属性 解释 type 必须是regex_filter regex 指定正则表达式 excludeEvents true或者false
    3. 案例

20. 编写格式文件添加如下内容

    定义 数据源(输入端) 缓冲区 输出源(输出端)

    a1.sources r1
    a1.channels c1
    a1.sinks k1

    输入端

    a1.sources.r1.type spooldir
    a1.sources.r1.spoolDir /opt/upload
    a1.sources.r1.fileSuffix .done

    拦截器

    a1.sources.r1.interceptors i1
    a1.sources.r1.interceptors.i1.type regex_filter
    #全部都是符合条件的数据
    a1.sources.r1.interceptors.i1.regex ^.INFO.$
    #排除符合正则表达式的数据

    a1.sources.r1.interceptors.i1.excludeEvents true

    输出端

    a1.sinks.k1.type hdfs
    a1.sinks.k1.hdfs.path hdfs://flume45:9000/interceptors/%Y%m%d/%H
    #是否使用本地时间戳
    a1.sinks.k1.hdfs.useLocalTimeStamp true

    序列化

    a1.sinks.k1.hdfs.fileType DataStream
    a1.sinks.k1.hdfs.rollInterval 0

    使用一个在内存中缓冲事件的通道

    a1.channels.c1.type memory

    连接通道

    a1.sources.r1.channels c1
    a1.sinks.k1.channel c1 2. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f regexin.conf - Dflume.root.loggerINFO,console 七、Custom Interceptor 1. 概述

21. 在Flume中也允许自定义拦截器。但是不同于其他组件自定义Interceptor的时候需要再额外覆盖其中的内部接口。

22. 步骤 a. 构建Maven工程导入对应的依赖。 b. 自定义一个类实现Interceptor接口覆盖其中initializeintercept和close方法。 c. 定义静态内部类实现Interceptor.Builder内部接口。 d. 打成jar包方法Flume安装目录的lib目录下。 e. 编写格式文件添加如下内容 a1.sources s1 a1.channels c1 a1.sinks k1 a1.sources.s1.type netcat a1.sources.s1.bind 0.0.0.0 a1.sources.s1.port 8090

    指定拦截器

    a1.sources.s1.interceptors i1 a1.sources.s1.interceptors.i1.type cn.tedu.flume.interceptor.AuthInterceptor$Builder a1.channels.c1.type memory a1.sinks.k1.type logger a1.sources.s1.channels c1 a1.sinks.k1.channel c1 f. 启动Flume ../bin/flume-ng agent -n a1 -c ../conf -f authin.conf - Dflume.root.loggerINFO,console