apache flume sources

2025-02-09 技术教程

Avro Source

简介：监听Avro端口并从外部的Avro客户端接收事件流，多个flume agent可以通过Avro形成一个组织。

Property NameDefaultDescription

channels–

type–需要配置为Avro

bind–需要监听的主机名或ip地址

port–监听的端口

threads–工作的最大线程数

selector.type

selector.*

interceptors–空格分割的拦截器列表

interceptors.*

compression-typenone这块可以是“none”或者“deflate”，压缩类型必须和AvroSource匹配

sslfalse设置为true开启ssl. 同时必须明确 “keystore”和“keystore-password”.

keystore–这是ssl需要的keystore file的地址

keystore-password–这是ssl需要的keystore的密码

keystore-typeJKSkeystore的类型. 可以是“JKS”或者“PKCS12”.

exclude-protocolsSSLv3空格分割的需要排序的SSL/TLS协议. SSLv3默认被排除

ipFilterfalse设置为true开启netty的ip过滤

ipFilter.rules–通过逗号分隔的ip过滤规则

ipFilter.rules例子：ipFilter.rules=allow:ip:127.*,allow:name:localhost,deny:ip:*

2．Thrift Source

简介:监控Thrift客户端并从外部的Thrift客户端接收事件流.

Property NameDefaultDescription

channels–

type–需要配置为Thrift

bind–需要监听的主机名或ip地址

port–需要监听的端口

threads–工作的最大线程数

selector.type

selector.*

interceptors–空格分割的拦截器列表

interceptors.*

sslfalse设置为true开启ssl. 同时必须明确 “keystore”和“keystore-password”.

keystore–This is the path to a Java keystore file. Required for SSL.

keystore-password–The password for the Java keystore. Required for SSL.

keystore-typeJKSThe type of the Java keystore. This can be “JKS” or “PKCS12”.

exclude-protocolsSSLv3Space-separated list of SSL/TLS protocols to exclude. SSLv3 will always be excluded in addition to the protocols specified.

kerberosfalseSet to true to enable kerberos authentication. In kerberos mode, agent-principal and agent-keytab are required for successful authentication. The Thrift source in secure mode, will accept connections only from Thrift clients that have kerberos enabled and are successfully authenticated to the kerberos KDC.

agent-principal–The kerberos principal used by the Thrift Source to authenticate to the kerberos KDC.

agent-keytab—-The keytab location used by the Thrift Source in combination with the agent-principal to authenticate to the kerberos KDC.

3.exec source

简介:运行一个给定的unix命令,并且该命令会不断产生标准数据(stderr直接丢弃除非属性 logStdErr设置为true),如果进程因为任何原因失败了,source仍然会继续运行但是没有数据.

Property NameDefaultDescription

channels–

type–需要设置为exec

command–执行的unix命令

shell–shell将会调用的命令,如:/bin/sh -c.

restartThrottle10000 多少毫秒重新执行

restartfalse如果命令失败了是否重启

logStdErrfalse是否stderr被记录

batchSize20一次发送到channel的记录条数

batchTimeout3000如果缓存还没有慢,数据在缓存多少毫秒后被发送

selector.typereplicatingreplicating或者multiplexing

selector.* 取决于selector.type的值

interceptors–空格分割的拦截器列表

interceptors.*

警告:ececsource和其他的异步都是无法保证的source,数据可能会丢失,比如当你在tailf一个文件的时候,进程出现了问题,当恢复正常的时候execsource无法知道上次读取到了什么位置,只会从当前位置开始tail,解决方法是可以使用spool source.

提示:当使用tail时,使用-F参数而不是-f,-F支持文件旋转

4.JMS source

简介:JMS source从JMS目标地址比如queue或者是topic获取信息,我们只有测试过ActiveMQ

Property NameDefaultDescription

channels–

type–需要设置为jms

initialContextFactory–e.g: org.apache.activemq.jndi.ActiveMQInitialContextFactory

connectionFactory–The JNDI name the connection factory shoulld appear as

providerURL–The JMS provider URL

destinationName–Destination name

destinationType–Destination type (queue or topic)

messageSelector–Message selector to use when creating the consumer

userName–Username for the destination/provider

passwordFile–File containing the password for the destination/provider

batchSize100Number of messages to consume in one batch

converter.typeDEFAULTClass to use to convert messages to flume events. See below.

converter.*–Converter properties.

converter.charsetUTF-8Default converter only. Charset to use when converting JMS TextMessages to byte arrays.

converter:BytesMessage,TextMessage,ObjectMessage

5.Spooling Directory Source

简介:监听给定目录下文件,然后将文件传输,该source是可靠的不会丢失数据.放到该目录下的文件必须是不可变的,唯一的.

如果有以下条件产生,任务将会报错:

1.当目录中文件被打开写入时,flume将会报错,任务结束

2.当文件被再次使用的时候,flume将会报错,任务结束

为了解决这些问题,使用唯一标识,如timestamp将会有效解决

Property NameDefaultDescription

channels–

type–需要设置为spooldir.

spoolDir–读取文件的目录

fileSuffix.COMPLETED当文件读取完毕后加的后缀名

deletePolicynever是否删除文件,never或者immediate

fileHeaderfalse文件是否存储到一个绝对路径

fileHeaderKeyfile绝对路径的值

basenameHeaderfalseWhether to add a header storing the basename of the file.

basenameHeaderKeybasenameHeader Key to use when appending basename of file to event header.

ignorePattern^$ 使用正则表达式表示哪些文件被跳过

trackerDir.flumespool存储进程matadata文件的路径,如果不是绝对路径的话将被解释为spooldir的相对路径

consumeOrderoldest/youngest/random,文件被处理的先后,使用文件的最后修改时间来比较,如果时间一致,文件小的先被处理

maxBackoff4000当缓冲池满了后多少毫秒后重新尝试发送

batchSize100每次传送到channel的记录数

inputCharsetUTF-8文件被当做文本的编码.

decodeErrorPolicyFAIL当文件无法解码时怎么做, FAIL: Throw an exception and fail to parse the file. REPLACE: Replace the unparseable character with the “replacement character” char, typically Unicode U+FFFD. IGNORE: Drop the unparseable character sequence.

deserializerLINE指定文件被怎么样指定为事件, 默认将每一行当做一个事件.自己实现的类必须implement EventDeserializer.Builder.

deserializer.* Varies per event deserializer.

bufferMaxLines–(Obselete) 这个配置目前被忽略

bufferMaxLineLength5000(Deprecated) 多少字节的行可以被提交,不赞成使用, Use deserializer.maxLineLength instead.

selector.typereplicatingreplicating or multiplexing

selector.* Depends on the selector.type value

interceptors–Space-separated list of interceptors

interceptors.*