Flume 常用配置项

2021-07-06 08:47:03

注：以下配置项均为常见配置项，查询详细配置项可以访问 flume 官网

Source 常见配置项

Avro Source

配置项名称 默认值 描述
Channel –
type – 组件类型名称，必须为avro
bind – 要监听的主机名或IP地址
port – 要绑定的端口号
threads – 产生的最大工作线程数
ipFilterRules – 使用此配置定义N个netty ipFilter模式规则。

Thrift Source

配置项名称 默认值 描述
Channel –
type – 组件类型名称，必须为Thrift
bind – 要监听的主机名或IP地址
port – 要绑定的端口号
threads – 产生的最大工作线程数
ipFilterRules – 使用此配置定义N个netty ipFilter模式规则。

Exec Source

配置项名称 默认值 描述
Channel –
type – 组件名称，必须为 exec
command – 需要执行的指令
restartThrottle 10000 重启之前需要等待的时间(10秒)
restart false cmd挂掉是否应该自动重启
batchSize 20 每次可以读取的行数
batchTimeout 3000 这批数据必须处理完的时间

Kafka Source

配置项名称 默认值 描述
channels –
type – 组件名称，必须为org.apache.flume.source.kafka,KafkaSource
zookeeperConnect – Zookeeper的地址
groupId flume consumer group的唯一标识，在多个source或多个agent中设置相同的ID表示它们属于同一个consumer group
topic – Kafka的Topic，只能是一个
batchSize 1000 一个批次中最大能取到的消息数

Netcat Source

配置项名称 默认值 描述
channels –
type – 组件名称，必须为 netcat
bind – 需要绑定的主机
port – 需要绑定的端口号
max-line-length 512 每条消息的最大字节数
ack-every-event true 为每条成功接收的消息回复‘OK‘

Sequence Generator Source

配置项名称 默认值 描述
channels –
type – 组件名称，必须为 seq

Http Source

配置项名称 默认值 描述
type 组件名称，必须为http
port – 需要绑定的端口
bind 0.0.0.0 需要监听的地址
handler org.apache.flume.source.http.JSONHandler handler类的完整名称
handler.* – handler的参数

Custom Source

配置项名称 默认 描述
channels –
type – 组件名称，为用户自定义的完整类名称

Channels 常用配置项

Memory Channel

配置项名称 默认值 描述
type – 组件名称，必须为 memory
capacity 100 channel最大能存储的event数量
transactionCapacity 100 每次事务中能从source中拉取或下沉到sink的最大event数量
keep-alive 3 添加
byteCapacityBufferPercentage 20 定义byteCapacityBuffer 的占比，它与 byteCapacity之和为100%，byteCapacity见下栏
byteCapacity 在这个channel中允许存放的最大字节数 这个字节数只计算event body,这也是需要提供 byteCapacityBufferPercentage 配置参数的原因。 默认为JVM可用最大内存的80% (即通过命令行传递的-Xmx值的80%)。 注意，如果一个agent里面包含了多个channel，channel里面又包含了多个相同的events，那么这些events会被重复计算。 将它设为 0 将会导致这个值强制设定为200GB的容量

JDBC Channel

配置项名称 默认值 描述
type – 组件名称，必须为 jdbc
db.type DERBY 数据库类型，必须为 DERBY.
driver.class org.apache.derby.jdbc.EmbeddedDriver JDBC 驱动类
driver.url (constructed from other properties) JDBC 连接路径 URL
db.username “sa” 数据库用户
db.password – 密码
connection.properties.file – JDBC 连接的配置文件
create.schema true 如果为真，则如果库不存在，就会自动创建
create.index true 创建索引以加快查找
create.foreignkey true 是否允许创建外键
transaction.isolation “READ_COMMITTED” 隔离级别 READ_UNCOMMITTED, READ_COMMITTED, SERIALIZABLE, REPEATABLE_READ
maximum.connections 10 运行连接数据库的最大连接数
maximum.capacity 0 (不限制) channel中允许的最大event数
sysprop.* 数据库特殊配置
sysprop.user.home 嵌入式Derby数据库的存储路径

Kafka Channel

配置项名称 默认值 描述
type – 组件名称，必须为 org.apache.flume.channel.kafka.KafkaChannel
brokerList – Kafka Broker列表，建议至少有两台用于实现高可用，格式为hostname:port，broker之间用‘，’隔开
zookeeperConnect – Kafka集群使用的ZooKeeper的URI 格式是‘，’分隔的主机名hostname:port。如果使用chroot，则在末尾添加一次。 例如: zookeeper-1:2181,zookeeper-2:2182,zookeeper-3:2181/kafka
topic flume-channel Kafka topic
groupId flume channel用于在kafka注册的GroupID，多个channel必须使用相同的topic和group，以确保当一个agent失败时，另一个agent可以获得数据 注意:相同GroupID但是没有channel的consummer将导致丢失
readSmallestOffset false 当设置为true时，通道将从最老的事件开始读取主题中的所有数据，当为false时，它将只读取通道启动后写入的事件
Other Kafka Properties – 这些属性用于配置通道使用的Kafka生产者和消费者。Kafka支持的任何属性都可以使用。惟一的要求是在属性名前面加上前缀 kafka.. 举个例子: kafka.producer.type

File Channel

配置项名称 默认值 描述
type – 组件名称，必须为 file.
checkpointDir ~/.flume/file-channel/checkpoint checkpoint文件的存储目录
useDualCheckpoints false 备份checkpoint文件，如果它为 true, backupCheckpointDir 必须 被设置
backupCheckpointDir – 用于备份的checkpoint目录，它 必须不能 与原checkpoint目录一样
dataDirs ~/.flume/file-channel/data ‘，’分隔的目录列表，用于存储日志文件。在不同的磁盘上使用多个目录可以提高filechannel的性能
transactionCapacity 10000 channel支持的最大事务数
checkpointInterval 30000 checkpoint之间的最大时间间隔
maxFileSize 2146435071 单个日志文件最大的字节数
minimumRequiredSpace 524288000 最少需要的空闲空间(以字节为单位)。为了避免数据损坏，当空闲空间低于此值时，filechannel将停止接受take/put请求
capacity 1000000 单个channel存储的最大容量
keep-alive 3 一个put操作执行的最大时间
checkpointOnClose true 在关闭时是否创建检查点，如果设置为true，可以加速filechannel的后续启动。

Sinks 常用配置

HDFS Sink

Name 默认值 描述
channel –
type – 组件名称，必须为 hdfs
hdfs.path – HDFS 目录路径 (例如 hdfs://namenode/flume/webdata/)
hdfs.filePrefix FlumeData Flume创建的文件名的前缀
hdfs.fileSuffix – Flume创建的文件名的后缀 (例如 .avro - 注意：当前不会自动添加)
hdfs.inUsePrefix – 当前正在写入的文件名的前缀
hdfs.inUseSuffix .tmp 当前正在写入的文件名的后缀
hdfs.rollInterval 30 触发文件滚动要等待的秒数 (0 = 无需根据时间进行滚动)
hdfs.rollSize 1024 触发文件滚动的字节数大小 (0: 无需根据文件大小滚动)
hdfs.rollCount 10 触发文件滚动的event数量 (0 = 无需根据events进行滚动)
hdfs.idleTimeout 0 自动关闭文件的时间间隔(如果没有数据写入) (0 = 不自动关闭)
hdfs.batchSize 100 每次写入 HDFS前读取的events数量
hdfs.codeC – 压缩成以下格式之一: gzip, bzip2, lzo, lzop, snappy
hdfs.fileType SequenceFile 当前序列化格式，有以下三种： SequenceFile, DataStream or CompressedStream (1)DataStream 不能设置 codeC (2)CompressedStream 需要设置 hdfs.codeC
hdfs.maxOpenFiles 5000 允许同时打开文件的最大数量。如果超过这个数字，则关闭最旧的文件。
hdfs.minBlockReplicas – 指定每个HDFS块的最小副本数。如果没有指定，它来自类路径中的默认Hadoop配置.
hdfs.writeFormat – writeFormat格式. 有以下两种： “Text” ， “Writable” (默认).
hdfs.callTimeout 10000 HDFS打开、写入、刷新、关闭操作允许的最大毫秒数。 如果发生许多HDFS超时操作，则应增加此数值。
hdfs.threadsPoolSize 10 HDFS操作的线程池数量
hdfs.round false 时间戳是否应该四舍五入 (如果为true, 时间戳会向下取时间)
hdfs.roundValue 1 四舍五入到它的最高倍数hdfs.roundUnit), 比当前时间更小
hdfs.roundUnit second 四舍五入的数值 - second, minute or hour.

Hive sink

Name 默认值 描述
channel –
type – 组件名称，必须为 hive
hive.metastore – Hive metastore 的路径 (例如 thrift://a.b.com:9083 )
hive.database – Hive database
hive.table – Hive 表名
hive.partition – ‘，’分隔分区值列表，标识要写入的分区。 例如: 如果表的分区为 (continent: string, country :string, time : string) 那么 ‘Asia,India,2014-02-26-01-21’ 将会被解释为 continent=Asia,country=India,time=2014-02-26-01-21
batchSize 15000 单个Hive事务中能写入的最大event数量

Logger sink

配置项名称 默认值 描述
channel –
type – 组件名称，必须为 logger

Avro Sink

配置项名称 默认值 描述
channel –
type – 组件名称，必须为 avro.
hostname – 需要绑定的主机或IP地址
port – 监听的端口号
batch-size 100 一次性发送的最大events数量
connect-timeout 20000 允许第一次(连接)请求的时间(ms).

Thrift Sink

配置项名称 默认值 描述
channel –
type – 组件名称，必须为 thrift.
hostname – 需要绑定的主机或IP地址
port – 监听的端口号
batch-size 100 一次性发送的最大events数量
connect-timeout 20000 允许第一次(连接)请求的时间(ms).

HBase Sink

配置项名称 默认值 描述
channel –
type – 组件名称，必须为 hbase
table – 需要写入到Hbase的表名
columnFamily – Hbase的列族
zookeeperQuorum – quorum 配置. 这个配置的值为 hbase.zookeeper.quorum它在 hbase-site.xml文件中

Flume 常用配置项

上一篇：Chrome中访问https的网址，无法访问，提示“您的连接不是私密连接”的解决办法


下一篇：.Net5使用EFCore CodeFirst模式 数据迁移并实现DBContext依赖注入