全部产品

Flume插件

更新时间:2020-03-23 21:06:35

Flume插件

产品介绍

Flume-DataHub插件是基于Flume开发的DataHub数据订阅/发布插件,可以将采集到的数据写入DataHub,也可以从DataHub读取数据写入其他系统。该插件遵守Flume插件开发规范,安装方便,可以很方便的向DataHub发布/订阅数据。

产品安装

安装限制

  1. JDK版本 >= 1.7
  2. Apache Maven 3.x
  3. Flume-NG 1.x

    下载flume(如已下载请跳过)

    首先下载flume,下载链接
  1. $ tar zxvf apache-flume-1.9.0-bin.tar.gz

方便起见,以下介绍以${FLUME_HOME}表示Flume主目录位置。

安装flume-datahub

当前Flume仅支持Linux系统。

直接安装

  1. 首先下载flume-datahub插件,下载链接 。(历史版本可在文末进行下载)
  2. 解压flume插件并放在${FLUME_HOME}/plugins.d目录下

    1. $ tar aliyun-flume-datahub-sink-x.x.x.tar.gz
    2. $ cd aliyun-flume-datahub-sink-x.x.x
    3. $ mkdir ${FLUME_HOME}/plugins.d
    4. $ mv aliyun-flume-datahub-sink ${FLUME_HOME}/plugins.d
  3. 源码安装

  4. 首先下载源码,github地址
  5. 编译并安装
    1. $ cd aliyun-maxcompute-data-collectors
    2. $ mvn clean package -DskipTests=true -Dmaven.javadoc.skip=true
    3. $ cd flume-plugin/target
    4. $ tar zxvf aliyun-flume-datahub-sink-x.x.x.tar.gz
    5. $ mv aliyun-flume-datahub-sink ${FLUME_HOME}/plugins.d

参数介绍

sink 参数介绍

名称 默认值                                   是否必须          描述
datahub.endPoint - 必须 阿里云datahub的服务地址
datahub.accessId - 必须 阿里云access id
datahub.accessKey - 必须 阿里云access key
datahub.project - 必须 datahub项目名称
datahub.topic - 必须 datahub topic名称
datahub.shard.ids 所有shard 可选 写入datahub的指定shard列表,以”,”分隔,例如 0,1,2。
每次从shard列表随机选择一个shard写入DataHub。
在发生shard分裂或者合并时,如果没有指定该参数,那么shard分裂或者合并后,flume会自动调整shard列表,否则需要用户手动修改配置文件。
datahub.enablePb true 可选 是否使用pb传输,部分专有云不支持需要手动设置为false
datahub.compressType none 可选 是否压缩传输,目前支持 LZ4, DEFLATE
datahub.batchSize 1000 可选 datahub每次发送的最大数据量
datahub.maxBufferSize 210241024 可选 datahub单次请求写入数据量的最大值(单位:Byte)。
不建议修改该参数,单次写入数据量过大可能写入失败
datahub.batchTimeout 5 可选 如果数据量没有达到batchSize,向datahub同步数据之前等待的时间(单位:秒)
datahub.retryTimes 3 可选 数据同步失败重试次数
datahub.retryInterval 5 可选 数据同步失败重试间隔(单位:秒)
datahub.dirtyDataContinue true 可选 遇到脏数据是否继续处理,为true时,会自动将脏数据以,分隔符写入脏数据文件,不影响后续数据的处理
datahub.dirtyDataFile DataHub-Flume-dirty-file 可选 脏数据文件
serializer - 必须 数据解析方式,目前支持DELIMITED(分隔符)和REGEX(正则表达式)
serializer.delimiter , 可选 数据字段分割符,如果要使用特殊字符需要添加双引号,例如”\t”
serializer.regex (.*) 可选 数据解析的的正则表达式,每个字段的数据被解析成一个group
serializer.fieldnames - 必须 输入数据字段到datahub字段的映射,以输入的顺序标示字段,如果要跳过某个字段, 不指定列名即可,例如 c1,c2,,c3,表示将输入数据的第一、二、四字段和datahub的c1,c2,c3字段进行匹配。
serializer.charset UTF-8 可选 数据解析编码格式

Source 参数

名称 默认值 是否必须 描述
datahub.endPoint - 必须 阿里云datahub的服务地址
datahub.accessId - 必须 阿里云access id
datahub.accessKey - 必须 阿里云access key
datahub.project - 必须 datahub项目名称
datahub.topic - 必须 datahub topic名称
datahub.subId - 必须 datahub 订阅 id
datahub.startTime - 可选 datahub 指定时间点进行读数据,格式为yyyy-MM-dd HH:mm:ss,使用该参数会首先重置订阅,然后根据订阅读取数据。
datahub.shard.ids - 可选 读取datahub的指定shard列表,以”,”分隔,例如 0,1,2。每次读数据时会从shard列表随机选择一个shard进行消费。
如不指定,则采用协同消费进行数据读取。
不建议使用该参数,如果配置了多个source的情况下,不指定该参数时,协同消费会自动分配shard,尽可能保证每个source负载均衡。
datahub.enablePb true 可选 是否使用pb传输,部分专有云不支持需要手动设置为false
datahub.compressType none 可选 是否压缩传输,目前支持 LZ4, DEFLATE
datahub.batchSize 1000 可选 DataHub每次读取的最大数据量
datahub.batchTimeout 5 可选 如果数据量没有达到batchSize,向datahub同步数据之前等待的时间(单位:秒)
datahub.retryTimes 3 可选 数据读取失败重试次数,重试间隔默认为1S,不可调整
datahub.autoCommit true 可选 读取数据是否自动提交点位,修改为false可能发生数据重复消费问题
datahub.offsetCommitTimeout 30 可选 自动提交点位时间间隔(单位:秒)
datahub.sessionTimeout 60 可选 source功能采取协同消费实现,协同消费超时没有发送心跳,则session会自动关闭
serializer - 必须 数据解析方式,目前支持DELIMITED(分隔符),数据的每个字段将会以datahub schema顺序写成一行,并以delimiter进行分隔
serializer.delimiter , 可选 数据字段分割符,如果要使用特殊字符需要添加双引号,例如”\t”
serializer.charset UTF-8 可选 数据解析编码格式

Sink 使用案例

案例一: DELIMITED serializer

DELIMITED解析数据时将每一行作为一条Record,并以给定的分隔符对数据进行解析。下面以csv文件为例,说明如何使用flume将批量csv文件准时时上传到DataHub。

数据文件

将以下内容保存在本地文件/temp/test.csv中。

  1. 0,YxCOHXcst1NlL5ebJM9YmvQ1f8oy8neb3obdeoS0,true,1254275.1144629316,1573206062763,1254275.1144637289
  2. 0,YxCOHXcst1NlL5ebJM9YmvQ1f8oy8neb3obdeoS0,true,1254275.1144629316,1573206062763,1254275.1144637289
  3. 1,hHVNjKW5DsRmVXjguwyVDjzjn60wUcOKos9Qym0V,false,1254275.1144637289,1573206062763,1254275.1144637289
  4. 2,vnXOEuKF4Xdn5WnDCPbzPwTwDj3k1m3rlqc1vN2l,true,1254275.1144637289,1573206062763,1254275.1144637289
  5. 3,t0AGT8HShzroBVM3vkP37fIahg2yDqZ5xWfwDFJs,false,1254275.1144637289,1573206062763,1254275.1144637289
  6. 4,MKwZ1nczmCBp6whg1lQeFLZ6E628lXvFncUVcYWI,true,1254275.1144637289,1573206062763,1254275.1144637289
  7. 5,bDPQJ656xvPGw1PPjhhTUZyLJGILkNnpqNLaELWV,false,1254275.1144637289,1573206062763,1254275.1144637289
  8. 6,wWF7i4X8SXNhm4EfClQjQF4CUcYQgy3XnOSz0StX,true,1254275.1144637289,1573206062763,1254275.1144637289
  9. 7,whUxTNREujMP6ZrAJlSVhCEKH1KH9XYJmOFXKbh8,false,1254275.1144637289,1573206062763,1254275.1144637289
  10. 8,OYcS1WkGcbZFbPLKaqU5odlBf7rHDObkQJdBDrYZ,true,1254275.1144637289,1573206062763,1254275.1144637289

DataHub Topic Schema

以上数据对应的DataHub schme为:

字段名称 字段类型
id BIGINT
name STRING
gender BOOLEAN
salary DOUBLE
my_time TIMESTAMP
decimal DECIMAL

flume 配置文件

在目录 ${FLUME_HOME}/conf 下创建文件名为datahub_basic.conf的文件,然后将以下内容写入文件。本实例采用Exec Source作为数据源,更多Source可以参考Flume官方文档
注意: ExecSource源可能发生数据丢失,因为EeecSource无法保证将事件放入Channel,在这种情况下,数据将丢失。例如,tail命令获取数据时,此时flume channel已满,而这部分数据将会丢失。建议使用Spooling Directory Source或者Taildir Source,下方有Taildir Source案例介绍。
这里将静态文件/temp/test.csv作为数据源,如果文件为动态写入的日志文件,可使用命令tail -F logFile进行实时采集。

  1. # A single-node Flume configuration for Datahub
  2. # Name the components on this agent
  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1
  6. # Describe/configure the source
  7. a1.sources.r1.type = exec
  8. a1.sources.r1.command = cat /temp/test.csv
  9. # Describe the sink
  10. a1.sinks.k1.type = com.aliyun.datahub.flume.sink.DatahubSink
  11. a1.sinks.k1.datahub.accessId = {YOUR_ALIYUN_DATAHUB_ACCESS_ID}
  12. a1.sinks.k1.datahub.accessKey = {YOUR_ALIYUN_DATAHUB_ACCESS_KEY}
  13. a1.sinks.k1.datahub.endPoint = {YOUR_ALIYUN_DATAHUB_ENDPOINT}
  14. a1.sinks.k1.datahub.project = datahub_project_test
  15. a1.sinks.k1.datahub.topic = test_topic
  16. a1.sinks.k1.serializer = DELIMITED
  17. a1.sinks.k1.serializer.delimiter = ,
  18. a1.sinks.k1.serializer.fieldnames = id,name,gender,salary,my_time,decimal
  19. a1.sinks.k1.serializer.charset = UTF-8
  20. a1.sinks.k1.datahub.retryTimes = 5
  21. a1.sinks.k1.datahub.retryInterval = 5
  22. a1.sinks.k1.datahub.batchSize = 100
  23. a1.sinks.k1.datahub.batchTimeout = 5
  24. a1.sinks.k1.datahub.enablePb = true
  25. a1.sinks.k1.datahub.compressType = DEFLATE
  26. # Use a channel which buffers events in memory
  27. a1.channels.c1.type = memory
  28. a1.channels.c1.capacity = 10000
  29. a1.channels.c1.transactionCapacity = 10000
  30. # Bind the source and sink to the channel
  31. a1.sources.r1.channels = c1
  32. a1.sinks.k1.channel = c1

启动Flume

Dflume.root.logger=INFO,console选项可以将日志实时输出到控制台,如需更多信息可采用DEBUG模式。使用如下命令启动Flume,即可完成CSV文件数据采集进入DataHub:

  1. $ cd ${FLUME_HOME}
  2. $ bin/flume-ng agent -n a1 -c conf -f conf/datahub_basic.conf -Dflume.root.logger=INFO,console

案例二: REGEX serializer

REGEX解析数据时将每一行作为一条Record,并以给定的正则表达式对数据进行解析,一条Record的信息的多个内容以分组表示。下面以日志文件为例,说明flume如何利用正则表达式准时时上传到DataHub。

数据文件

将以下内容保存在本地文件/temp/test.csv中。其中需要同步的数据内容为日期后面的详细数据。

  1. 1. [2019-11-12 15:20:08] 0,j4M6PhzL1DXVTQawdfk306N2KnCDxtR0KK1pke5O,true,1254409.5059812006,1573543208698,1254409.5059819978
  2. 2. [2019-11-12 15:22:35] 0,mYLF8UzIYCCFUm1jYs9wzd2Hl6IMr2N7GPYXZSZy,true,1254409.5645912462,1573543355740,1254409.5645920434
  3. 3. [2019-11-12 15:23:14] 0,MOemUZur37n4SGtdUQyMohgmM6cxZRBXjJ34HzqX,true,1254409.5799291395,1573543394219,1254409.579929538
  4. 4. [2019-11-12 15:23:30] 0,EAFc1VTOvC9rYzPl9zJYa6cc8uJ089EaFd79B25i,true,1254409.5862723626,1573543410134,1254409.5862731598
  5. 5. [2019-11-12 15:23:53] 0,zndVraA4GP7FP8p4CkQFsKJkxwtYK3zXjDdkhmRk,true,1254409.5956010541,1573543433538,1254409.5956018514
  6. 6. [2019-11-12 15:24:00] 0,9YrjjoALEfyZm07J7OuNvDVNyspIzrbOOAGnZtHx,true,1254409.598201082,1573543440061,1254409.5982018793
  7. 7. [2019-11-12 15:24:23] 0,mWsFgFlUnXKQQR6RpbAYDF9OhGYgU8mljvGCtZ26,true,1254409.6073950487,1573543463126,1254409.607395447
  8. 8. [2019-11-12 15:26:51] 0,5pZRRzkW3WDLdYLOklNgTLFX0Q0uywZ8jhw7RYfI,true,1254409.666525653,1573543611475,1254409.6665264503
  9. 9. [2019-11-12 15:29:11] 0,hVgGQrXpBtTJm6sovVK4YGjfNMdQ3z9pQHxD5Iqd,true,1254409.7222845491,1573543751364,1254409.7222853464
  10. 10. [2019-11-12 15:29:52] 0,7wQOQmxoaEl6Cxl1OSo6cr8MAc1AdJWJQaTPT5xs,true,1254409.7387664048,1573543792714,1254409.738767202
  11. 11. [2019-11-12 15:30:30] 0,a3Th5Q6a8Vy2h1zfWLEP7MdPhbKyTY3a4AfcOJs2,true,1254409.7538966285,1573543830673,1254409.7538974257
  12. 12. [2019-11-12 15:34:54] 0,d0yQAugqJ8M8OtmVQYMTYR8hi3uuX5WsH9VQRBpP,true,1254409.8589555968,1573544094247,1254409.8589563938

DataHub Topic Schema

以上数据对应的DataHub schme为:

字段名称 字段类型
id BIGINT
name STRING
gender BOOLEAN
salary DOUBLE
my_time TIMESTAMP
decimal DECIMAL

flume 配置文件

在目录 ${FLUME_HOME}/conf 下创建文件名为datahub_basic.conf的文件,然后将以下内容写入文件。本实例采用Exec Source作为数据源,更多Source可以参考Flume官方文档
注意: ExecSource源可能发生数据丢失,因为EeecSource无法保证将事件放入Channel,在这种情况下,数据将丢失。例如,tail命令获取数据时,此时flume channel已满,而这部分数据将会丢失。建议使用Spooling Directory Source或者Taildir Source,下方有Taildir Source案例介绍。
这里将静态文件/temp/test.csv作为数据源,如果文件为动态写入的日志文件,可使用命令tail -F logFile进行实时采集。

  1. # A single-node Flume configuration for Datahub
  2. # Name the components on this agent
  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1
  6. # Describe/configure the source
  7. a1.sources.r1.type = exec
  8. a1.sources.r1.command = cat /temp/test.csv
  9. # Describe the sink
  10. a1.sinks.k1.type = com.aliyun.datahub.flume.sink.DatahubSink
  11. a1.sinks.k1.datahub.accessId = {YOUR_ALIYUN_DATAHUB_ACCESS_ID}
  12. a1.sinks.k1.datahub.accessKey = {YOUR_ALIYUN_DATAHUB_ACCESS_KEY}
  13. a1.sinks.k1.datahub.endPoint = {YOUR_ALIYUN_DATAHUB_ACCESS_KEY}
  14. a1.sinks.k1.datahub.project = datahub_project_test
  15. a1.sinks.k1.datahub.topic = test_topic
  16. a1.sinks.k1.serializer = REGEX
  17. a1.sinks.k1.serializer.regex = \\[\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\] (\\d+),(\\S+),([a-z]+),([-+]?[0-9]*\\.?[0-9]*),(\\d+),([-+]?[0-9]*\\.?[0-9]*)
  18. a1.sinks.k1.serializer.fieldnames = id,name,gender,salary,my_time,decimal
  19. a1.sinks.k1.serializer.charset = UTF-8
  20. a1.sinks.k1.datahub.retryTimes = 5
  21. a1.sinks.k1.datahub.retryInterval = 5
  22. a1.sinks.k1.datahub.batchSize = 100
  23. a1.sinks.k1.datahub.batchTimeout = 5
  24. # Use a channel which buffers events in memory
  25. a1.channels.c1.type = memory
  26. a1.channels.c1.capacity = 10000
  27. a1.channels.c1.transactionCapacity = 10000
  28. # Bind the source and sink to the channel
  29. a1.sources.r1.channels = c1
  30. a1.sinks.k1.channel = c1

启动Flume

Dflume.root.logger=INFO,console选项可以将日志实时输出到控制台,如需更多信息可采用DEBUG模式。使用如下命令启动Flume,即可完成CSV文件数据采集进入DataHub:

  1. $ cd ${FLUME_HOME}
  2. $ bin/flume-ng agent -n a1 -c conf -f conf/datahub_basic.conf -Dflume.root.logger=INFO,console

案例三: Flume Taildir Source

在上面的介绍中提到,Flume使用exec source时,可能会导致数据丢失,所以在实际生产环境中并不建议使用。如果想要采集本地日志,可以使用Taildir Source或者Spooling Directory Source。下面以Taildir为例,介绍日志文件的采集。Taildir将会可以指定文件组,然后观察指定的文件,并在检测到新行添加到每个文件后,几乎实时的进行读取。如果正在写入新行,则此源将重试读取它们,以等待写入完成。 Taildir Source会把每个文件的已经读到的位置信息以json格式储存在positionFile文件中,source event 放入channel失败,已读位置不会更新,所以Taildir Source是可靠的。

数据文件

所有的日志将以如下格式追加到文件末尾。日志文件命名格式为 *.log

  1. 0,YxCOHXcst1NlL5ebJM9YmvQ1f8oy8neb3obdeoS0,true,1254275.1144629316,1573206062763,1254275.1144637289

DataHub Topic Schema

以上数据对应的DataHub schme为:

字段名称 字段类型
id BIGINT
name STRING
gender BOOLEAN
salary DOUBLE
my_time TIMESTAMP
decimal DECIMAL

flume 配置文件

在目录 ${FLUME_HOME}/conf 下创建文件名为datahub_basic.conf的文件,然后将以下内容写入文件。

  1. # A single-node Flume configuration for Datahub
  2. # Name the components on this agent
  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1
  6. # Describe/configure the source
  7. a1.sources.r1.type = TAILDIR
  8. a1.sources.r1.positionFile = /temp/taildir_position.json
  9. a1.sources.r1.filegroups = f1
  10. a1.sources.r1.filegroups.f1 = /temp/.*log
  11. # Describe the sink
  12. a1.sinks.k1.type = com.aliyun.datahub.flume.sink.DatahubSink
  13. a1.sinks.k1.datahub.accessId = {YOUR_ALIYUN_DATAHUB_ACCESS_ID}
  14. a1.sinks.k1.datahub.accessKey = {YOUR_ALIYUN_DATAHUB_ACCESS_KEY}
  15. a1.sinks.k1.datahub.endPoint = {YOUR_ALIYUN_DATAHUB_ACCESS_KEY}
  16. a1.sinks.k1.datahub.project = datahub_project_test
  17. a1.sinks.k1.datahub.topic = test_topic
  18. a1.sinks.k1.serializer = DELIMITED
  19. a1.sinks.k1.serializer.delimiter = ,
  20. a1.sinks.k1.serializer.fieldnames = id,name,gender,salary,my_time,decimal
  21. a1.sinks.k1.serializer.charset = UTF-8
  22. a1.sinks.k1.datahub.retryTimes = 5
  23. a1.sinks.k1.datahub.retryInterval = 5
  24. a1.sinks.k1.datahub.batchSize = 100
  25. a1.sinks.k1.datahub.batchTimeout = 5
  26. a1.sinks.k1.datahub.enablePb = true
  27. a1.sinks.k1.datahub.compressType = DEFLATE
  28. # Use a channel which buffers events in memory
  29. a1.channels.c1.type = memory
  30. a1.channels.c1.capacity = 10000
  31. a1.channels.c1.transactionCapacity = 10000
  32. # Bind the source and sink to the channel
  33. a1.sources.r1.channels = c1
  34. a1.sinks.k1.channel = c1

启动Flume

Dflume.root.logger=INFO,console选项可以将日志实时输出到控制台,如需更多信息可采用DEBUG模式。使用如下命令启动Flume,即可完成CSV文件数据采集进入DataHub:

  1. 1. $ cd ${FLUME_HOME}
  2. 2. $ bin/flume-ng agent -n a1 -c conf -f conf/datahub_basic.conf -Dflume.root.logger=INFO,console

Source 使用案例

案例一

DataHub-Flume Source可以将Datahub中的数据读取出来,并且可靠的移动到另外的系统中,本文以logger(直接输出到控制台)为例,介绍DataHub-Flume Source的使用方法。

DataHub Topic Schema

以上数据对应的DataHub schme为:

字段名称 字段类型
id BIGINT
name STRING
gender BOOLEAN
salary DOUBLE
my_time TIMESTAMP
decimal DECIMAL

flume 配置文件

在目录 ${FLUME_HOME}/conf 下创建文件名为datahub_source.conf的文件,然后将以下内容写入文件。

  1. 1. # A single-node Flume configuration for Datahub
  2. 2. # Name the components on this agent
  3. 3. a1.sources = r1
  4. 4. a1.sinks = k1
  5. 5. a1.channels = c1
  6. 6.
  7. 7. # Describe/configure the source
  8. 8. a1.sources.r1.type = com.aliyun.datahub.flume.sink.DatahubSource
  9. 9. a1.sources.r1.datahub.endPoint = {YOUR_ALIYUN_DATAHUB_ACCESS_KEY}
  10. 10. a1.sources.r1.datahub.accessId = {YOUR_ALIYUN_DATAHUB_ACCESS_ID}
  11. 11. a1.sources.r1.datahub.accessKey = {YOUR_ALIYUN_DATAHUB_ACCESS_KEY}
  12. 12. a1.sources.r1.datahub.project = datahub_test
  13. 13. a1.sources.r1.datahub.topic = test_flume
  14. 14. a1.sources.r1.datahub.subId = {YOUR_ALIYUN_DATAHUB_SUB_ID}
  15. 15. a1.sources.r1.serializer = DELIMITED
  16. 16. a1.sources.r1.serializer.delimiter = ,
  17. 17. a1.sources.r1.serializer.charset = UTF-8
  18. 18. a1.sources.r1.datahub.retryTimes = 3
  19. 19. a1.sources.r1.datahub.batchSize = 1000
  20. 20. a1.sources.r1.datahub.batchTimeout = 5
  21. 21. a1.sources.r1.datahub.enablePb = false
  22. 22.
  23. 23. # Describe the sink
  24. 24. a1.sinks.k1.type = logger
  25. 25.
  26. 26. # Use a channel which buffers events in memory
  27. 27. a1.channels.c1.type = memory
  28. 28. a1.channels.c1.capacity = 10000
  29. 29. a1.channels.c1.transactionCapacity = 10000
  30. 30.
  31. 31. # Bind the source and sink to the channel
  32. 32. a1.sources.r1.channels = c1
  33. 33. a1.sinks.k1.channel = c1

启动Flume

  1. $ cd ${FLUME_HOME}
  2. $ bin/flume-ng agent -n a1 -c conf -f conf/datahub_source.conf -Dflume.root.logger=INFO,console

Flume metric

DataHub-Flume 支持Flume的内置计数监控器,用户可以利用监控器来监控自己的Flume插件的运行情况。DataHub-Flume插件的Sink和Source都支持metric信息显示,具体参数含义可查看下表(只含DataHub相关的参数,更多参数含义参考Flume官方文档
1.)DatahubSink

名称 描述
BatchEmptyCount batch timeout时没有数据需要写入DataHub发生的次数
BatchCompleteCount Batch处理成功次数,仅包含全部写入成功的情况
EventDrainAttemptCount 尝试写入DataHub的数据数量(解析成功数量)
BatchUnderflowCount 成功写入DataHub的数据数量小于需要写入的数据量发生的次数。数据解析完成,但写入DataHub时部分失败或全部失败。
EventDrainSuccessCount 成功写入DataHub的数据量

2.)DatahubSource

名称 描述
EventReceivedCount Source接收到的DataHub的数据数量
EventAcceptedCount Source将DataHub数据成功写入channel的数量

使用方法

Flume提供了多种监控方法,本文以HTTP监控为例,介绍Flume监控工具的使用,更多的监控方法可以参考Flume官方文档
使用http方式监控,只需要在Flume插件启动时增加两个参数即可,-Dflume.monitoring.type=http -Dflume.monitoring.port=1234,其中type将监控方式指定为http,port为指定的端口号。使用示例如下:

  1. bin/flume-ng agent -n a1 -c conf -f conf/datahub_basic.conf -Dflume.root.logger=INFO,console -Dflume.monitoring.type=http -Dflume.monitoring.port=1234

插件成功启动之后,便可以登录web界面进行查看。地址为 http://ip:1234/metrics

FAQ

  • Q: flume启动报错org.apache.flume.ChannelFullException: Space for commit to queue couldn’t be acquired. Sinks are likely not keeping up with sources, or the buffer size is too tight
  • A: flume默认堆内存20MB,配置的batchSize过大时,flume使用的堆内存会超出20M。
    1. 解决方案1:调小batchSize;
    2. 解决方案2:调大flume最大堆内存
      1. $ vim bin/flume-ng
      2. JAV**A_OPTS**="-Xmx20m" ==> JAV**A_OPTS**="-Xmx1024m"
  • Q: DataHub-Flume插件是否支持json格式
  • A: 目前不支持,不过用户可以通过自定义正则表达式进行数据解析,或者修改DataHub-Flume插件代码,添加JSONEvent进行支持。
  • Q: DataHub-Flume插件支持Blob Topic吗?
  • A: 目前DataHub-Flume插件仅支持Tuple Topic,暂不支持blob。
  • Q:flume 报错 org.apache.flume.ChannelException: Put queue for MemoryTransaction of capacity 1 full, consider committing more frequently, increasing capacity or increasing thread count
  • A: channel已满,source数据写入channel失败。可以在配置文件中修改channel capacity解决,并且可以适当降低datahub source的batchSize。
  • Q: 使用旧版本flume时报错,可能会因为jar包冲突导致无法正常启动。

例如:使用flume1.6时,启动时报错:java.lang.NoSuchMethodError:com.fasterxml.jackson.databind.ObjectMapper.readerFor(Lcom/fasterxml/jackson/databind/JavaType;)Lcom/fasterxml/jackson/databind/ObjectReader;。因为新版本的插件依赖的jar包和flume本身依赖的jar包版本不一致,使用了flume依赖的旧版本jar包导致新版本的method找不到。

  • A: 删除${FLUME_HOME}/lib目录下的三个jar包即可。
    • jackson-annotations-2.3.0.jar
    • jackson-databind-2.3.1.jar
    • jackson-annotations-2.3.0.jar
  • Q: 使用flume采集数据时,空字符串自动转为null
  • A: 在flume插件2.0.2中对于非空字符串会做trim,空字符串直接转为null。flume插件2.0.3中已经优化掉,非空字符串写入DataHub依旧为空字符串。

历史版本下载

插件版本 flume版本(推荐)
2.0.3 1.9
2.0.4 1.9

2.0.3

  • 支持shardid传输列表;
  • 支持DataHub的DECIMAL、TIMESTAMP等类型。
  • 修复了一些一直日志信息,排查问题更容易;
  • 增加DataHub source,可以使用插件读取DataHub数据;
  • 支持压缩传输和二进制传输等功能;
  • 修复了空字符串会转为null的BUG;

2.0.4

  • 修复了fieldnames传输空列时的一个BUG