本次遇到的问题描述，日志采集同步时，当单条日志（日志文件中一行日志）超过2M大小，数据无法采集同步到kafka，分析后，共踩到如下几个坑。
1、flume采集时，通过shell+EXEC（tail -F xxx.log 的方式） source来获取日志时，当单条日志过大超过1M时，source端无法从日志中获取到Event。
2、日志超过1M后，flume的kafka sink 作为生产者发送给日志给kafka失败，kafka无法收到消息。
以下针对踩的这两个坑做分析，flume 我使用的是1.9.0版本。 kafka使用的是2.11-2.0.0版本
问题一、flume采集时，通过shell+EXEC（tail -F xxx.log 的方式） source来获取日志时，当单条日志过大超过1M时，source端无法从日志中获取到Event。flume的配置如下：
......
agent.sources = seqGenSrc
......
# For each one of the sources, the type is defined
agent.sources.seqGenSrc.type = exec
#agent.sources.seqGenSrc.command = tail -F /opt/logs/test.log|grep businessCollection|awk -F '- {' '{print "{"$2}'
agent.sources.seqGenSrc.command = tail -F /opt/logs/test.log|grep businessCollection
agent.sources.seqGenSrc.shell = /bin/bash -c
agent.sources.seqGenSrc.batchSize = 1
agent.sources.seqGenSrc.batchTimeout = 90000
......
原因：采用shell+EXEC方式的时候，flume的源码中使用的是如下的方式来获取日志
private Process process = null; //使用这种方式来执行命令。 process = Runtime.getRuntime().exec(commandArgs); //读取日志 reader = new BufferedReader( new InputStreamReader(process.getInputStream(), charset));
在一行日志超过1M后，这个代码就假死了，一直宕住，导致无法获取到数据。
针对这个问题处理方式：
方式一:修改源码的实现方式。（1.9.0的源码 对应的是源码中的flume-ng-core 项目中的org.apache.flume.source.ExecSource.java 这个类）
//process的采用如下方式获和执行命令，就改一行代码。增加.redirectErrorStream(true)后，输入流就都可以获取到，哪怕超过1M process = new ProcessBuilder(commandArgs).redirectErrorStream(true).start();
修改完成后，重新打包编译，然后将生成的jar包替换原来老的jar包。
方式二：放弃EXECSource，使用TAILDIR Source。 使用这个source时，对应的配置如下：
...... agent.sources = seqGenSrc ...... # For each one of the sources, the type is defined agent.sources.seqGenSrc.type = TAILDIR agent.sources.seqGenSrc.positionFile = ./taildir_position.json agent.sources.seqGenSrc.filegroups = seqGenSrc agent.sources.seqGenSrc.filegroups.seqGenSrc = /opt/logs/test.log agent.sources.seqGenSrc.fileHeader = false agent.sources.seqGenSrc.batchSize = 1 ......
建议采用TAILDIR Source 比较好，这个可以对多个日志进行监控和采集，而且日志采集时会记录日志采集位置到positionFile 中，这样日志采集不会重复。EXEC SOURCE在重启采集时数据会重复采集，还需要其他的方式去避免重复采集
问题二、日志超过1M后，flume的kafka sink 作为生产者发送给日志给kafka失败，kafka无法收到消息
原因：kafka 在默认情况下，只能接收1M大小以内的消息，在没有做自定义设置时。所以单条消息大于1M后是无法处理的。
处理方式如下：
1）、修改kafka 服务端server.properties文件，做如下设置(修改大小限制)
# The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer.bytes=502400 # The receive buffer (SO_RCVBUF) used by the socket server socket.receive.buffer.bytes=502400 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 message.max.bytes=5242880 replica.fetch.max.bytes=6291456
2）、修改producer.properties，做如下设置(修改大小限制)
# the maximum size of a request in bytes
max.request.size= 9242880
3）、java代码中在初始化kafka 生产者时，也需要指定max.request.size= 9242880
本文作者：张永清，转载请注明出处：一次flume exec source采集日志到kafka因为单条日志数据非常大同步失败的踩坑带来的思考
Properties properties = new Properties();
...
properties.put("max.request.size", 5242880);
...
KafkaProducer<Object,Object> kafkaProducer = new KafkaProducer<Object,Object>(properties);
4)、消费者在消费kafka数据时，也需要注意设置消费消息的大小限制
本文作者：张永清，转载请注明出处：一次flume exec source采集日志到kafka因为单条日志数据非常大同步失败的踩坑带来的思考
Properties properties = new Properties();
...
properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 6291456);
...
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
对于flume不了的同学，可以看flume 1.9中文版用户指南：https://www.h3399.cn/201906/700076.html
flume1.9 用户指南 (中文版)
概述
Apache Flume 是一个分布式, 可靠且可用的系统, 用于有效地从许多不同的 source 收集, 聚合和移动大量日志数据到集中式数据存储.
Apache Flume 的使用不仅限于日志数据聚合. 由于数据 source 是可定制的, 因此 Flume 可用于传输大量 event 数据, 包括但不限于网络流量数据, 社交媒体生成的数据, 电子邮件消息以及几乎任何可能的数据 source.
Apache Flume 是 Apache Software Foundation 的顶级项目.
系统要求
Java 运行时环境 - Java 1.8 或更高版本
内存 - 为 source,channel 或 sink 配置的内存
磁盘空间 - channel 或 sink 配置的磁盘空间
目录权限 - agent 使用的目录的读 / 写权限
架构
数据流模型
Flume event 被定义为具有字节有效负载和可选字符串属性集的数据流单元. Flume agent 是一个 (JVM) 进程, 它承载 event 从外部 source 流向下一个目标 (跃点) 的组件.

Flume source 消耗由外部 source(如 Web 服务器)传递给它的 event . 外部 source 以目标 Flume source 识别的格式向 Flume 发送 event . 例如, Avro Flume source 可用于从 Avro 客户端或从 Avrosink 发送 event 的流中的其他 Flume agent 接收 Avroevent . 可以使用 Thrift Flume Source 定义类似的流程, 以接收来自 Thrift Sink 或 Flume Thrift Rpc 客户端或 Thrift 客户端的 event , 这些客户端使用 Flume thrift 协议生成的任何语言编写. 当 Flume source 接收 event 时, 它将其存储到一个或多个 channels . 该 channel 是一个被动存储器, 可以保持 event 直到它被 Flume sink 消耗. 文件 channel 就是一个例子 - 它由本地文件系统支持. sink 从 channel 中移除 event 并将其放入外部存储库 (如 HDFS(通过 Flume HDFS sink)) 或将其转发到流中下一个 Flume agent (下一跳)的 Flume source. 给定 agent 中的 source 和 sink 与 channel 中暂存的 event 异步运行.
复杂的流程
Flume 允许用户构建多跳流, 其中 event 在到达最终目的地之前经过多个 agent . 它还允许 fan-in 和 fan-out, 上下文路由和故障跳跃的备份路由(故障转移).
可靠性
event 在每个 agent 的 channel 中进行. 然后将 event 传递到流中的下一个 agent 或终端存储库(如 HDFS). 只有将 event 存储在下一个 agent 的 channel 或终端存储库中后, 才会从 channel 中删除这些 event . 这就是 Flume 中的单跳消息传递语义如何提供流的端到端可靠性.
Flume 使用事务方法来保证 event 的可靠传递. source 和 sink 分别在事务中封装由 channel 提供的事务中放置或提供的 event 的存储 / 检索. 这可确保 event 集在流中从一个点到另一个点可靠地传递. 在多跳流的情况下, 来自前一跳的 sink 和来自下一跳的 source 都运行其事务以确保数据安全地存储在下一跳的 channel 中.
可恢复性
event 在 channel 中进行, 该 channel 管理从故障中恢复. Flume 支持由本地文件系统支持的持久文件 channel. 还有一个内存 channel, 它只是将 event 存储在内存中的队列中, 这更快, 但是当 agent 进程死亡时仍然留在内存 channel 中的任何 event 都无法恢复.
设置
设置 agent
Flume agent 配置存储在本地配置文件中. 这是一个遵循 Java 属性文件格式的文本文件. 可以在同一配置文件中指定一个或多个 agent 的配置. 配置文件包括 agent 中每个 source,sink 和 channel 的属性以及它们如何连接在一起以形成数据流.
配置单个组件
流中的每个组件 (source,sink 或 channel) 都具有特定于类型和实例化的名称, 类型和属性集. 例如, Avrosource 需要主机名 (或 IP 地址) 和端口号来接收数据. 内存 channel 可以具有最大队列大小 ("容量"),HDFS sink 需要知道文件系统 URI, 创建文件的路径, 文件轮换频率("hdfs.rollInterval") 等. 组件的所有此类属性需要在托管 Flume agent 的属性文件中设置.
将各个部分连接在一起
agent 需要知道要加载哪些组件以及它们如何连接以构成流程. 这是通过列出 agent 中每个 source,sink 和 channel 的名称, 然后为每个 sink 和 source 指定连接 channel 来完成的. 例如, agent 通过名为 file-channel 的文件 channel 将 event 从名为 avroWeb 的 Avrosource 流向 HDFS sink hdfs-cluster1. 配置文件将包含这些组件的名称和文件 channel, 作为 avroWebsource 和 hdfs-cluster1 sink 的共享 channel.
启动 agent
使用名为 flume-ng 的 shell 脚本启动 agent 程序, 该脚本位于 Flume 发行版的 bin 目录中. 您需要在命令行上指定 agent 名称, config 目录和配置文件:
$ bin/flume-ng agent -n $agent_name -c conf -f conf/flume-conf.properties.template
现在, agent 将开始运行在给定属性文件中配置的 source 和 sink.
一个简单的例子
在这里, 我们给出一个示例配置文件, 描述单节点 Flume 部署. 此配置允许用户生成 event , 然后将其记录到控制台.
# example.conf: A single-node Flume configuration # Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 # Describe the sink a1.sinks.k1.type = logger # Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
此配置定义名为 a1 的单个 agent .a1 有一个监听端口 44444 上的数据的 source, 一个缓冲内存中 event 数据的 channel, 以及一个将 event 数据记录到控制台的 sink. 配置文件命名各种组件, 然后描述其类型和配置参数. 给定的配置文件可能会定义几个命名的 agent 当一个给定的 Flume 进程启动时, 会传递一个标志, 告诉它要显示哪个命名 agent.
鉴于此配置文件, 我们可以按如下方式启动 Flume:
$ bin/flume-ng agent --conf conf --conf-file example.conf --name a1 -Dflume.root.logger=INFO,console
请注意, 在完整部署中, 我们通常会包含一个选项: --conf=<conf-dir> . 所述 <conf-dir> 目录将包括一个 shell 脚本 f lume-env.sh 和潜在的一个 log4j 的属性文件. 在这个例子中, 我们传递一个 Java 选项来强制 Flume 登录到控制台, 我们没有自定义环境脚本.
从一个单独的终端, 我们可以 telnet 端口 44444 并向 Flume 发送一个 event :
$ telnet localhost 44444 Trying 127.0.0.1... Connected to localhost.localdomain (127.0.0.1). Escape character is '^]'. Hello world! <ENTER> OK
原始的 Flume 终端将在日志消息中输出 event .
12/06/19 15:32:19 INFO source.NetcatSource: Source starting
12/06/19 15:32:19 INFO source.NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]
12/06/19 15:32:34 INFO sink.LoggerSink: Event: {
headers:{
} body: 48 65 6C 6C 6F 20 77 6F 72 6C 64 21 0D Hello world!.
}
恭喜 - 您已成功配置并部署了 Flume agent ! 后续部分更详细地介绍了 agent 配置.
在配置文件中使用环境变量
Flume 能够替换配置中的环境变量. 例如:
a1.sources = r1
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = ${
NC_PORT
}
a1.sources.r1.channels = c1
注意: 它目前仅适用于 values , 不适用于 keys . (IE. only on the "right side" of the = mark of the config lines.)
通过设置 propertiesImplementation = org.apache.flume.node.EnvVarResolverProperties, 可以通过 agent 程序调用上的 Java 系统属性启用此功能.
例如:
$ NC_PORT=44444 bin/flume-ng agent -conf conf -conf-file example.conf -name a1 -Dflume.root.logger=INFO,console -DpropertiesImplementation=org.apache.flume.node.EnvVarResolverProperties
请注意, 上面只是一个示例, 可以通过其他方式配置环境变量, 包括在 conf/flume-env.sh.
记录原始数据
在许多生产环境中记录流经摄取 pipeline 的原始数据流不是所希望的行为, 因为这可能导致泄漏敏感数据或安全相关配置 (例如密钥) 泄漏到 Flume 日志文件. 默认情况下, Flume 不会记录此类信息. 另一方面, 如果数据管道被破坏, Flume 将尝试提供调试 DEBUG 的线索.
调试 event 管道问题的一种方法是设置 连接到 Logger Sink 的附加内存 channel, 它将所有 event 数据输出到 Flume 日志. 但是, 在某些情况下, 这种方法是不够的.
为了能够记录 event 和配置相关的数据, 除了 log4j 属性外, 还必须设置一些 Java 系统属性.
要启用与配置相关的日志记录, 请设置 Java 系统属性 - Dorg.apache.flume.log.printconfig=true . 这可以在命令行上传递, 也可以在 flume-env.sh 中的 JAVA_OPTS 变量中设置.
要启用数据记录, 请 按照上述相同方式设置 Java 系统属性 -Dorg.apache.flume.log.rawdata=true . 对于大多数组件, 还必须将 log4j 日志记录级别设置为 DEBUG 或 TRACE, 以使特定于 event 的日志记录显示在 Flume 日志中.
下面是启用配置日志记录和原始数据日志记录的示例, 同时还将 Log4j 日志级别设置为 DEBUG 以用于控制台输出:
$ bin/flume-ng agent --conf conf --conf-file example.conf --name a1 -Dflume.root.logger=DEBUG,console -Dorg.apache.flume.log.printconfig=true -Dorgwdata=true
基于 Zookeeper 的配置
Flume 通过 Zookeeper 支持 agent 配置. 这是一个实验性功能. 配置文件需要在可配置前缀下的 Zookeeper 中上传. 配置文件存储在 Zookeeper 节点数据中. 以下是 agent 商 a1 和 a2 的 Zookeeper 节点树的外观
- /flume |- /a1 [Agent config file] |- /a2 [Agent config file]
上载配置文件后, 使用以下选项启动 agent
$ bin/flume-ng agent -conf conf -z zkhost:2181,zkhost1:2181 -p /flume -name a1 -Dflume.root.logger=INFO,console
| Argument Name | Default | Description |
|---|---|---|
| z | – | Zookeeper 连接字符串。以逗号分隔的主机名列表：port |
| p | /flume | Zookeeper 中的基本路径，用于存储 agent 配置 |
Flume 拥有完全基于插件的架构. 虽然 Flume 附带了许多开箱即用的 source,channels,sink,serializers 等, 但许多实现都与 Flume 分开运行. 安装第三方插件
虽然通过将自己的 jar 包添加到 flume-env.sh 文件中的 FLUME_CLASSPATH 变量中, 始终可以包含自定义 Flume 组件, 但 Flume 现在支持一个名为 plugins.d 的特殊目录, 该目录会自动获取以特定格式打包的插件. 这样可以更轻松地管理插件打包问题, 以及更简单的调试和几类问题的故障排除, 尤其是库依赖性冲突.
目录
该 plugins.d 目录位于 $FLUME_HOME/plugins.d . 在启动时, flume-ng 启动脚本在 plugins.d 目录中查找符合以下格式的插件, 并在启动 java 时将它们包含在正确的路径中.
插件的目录布局
plugins.d 中的每个插件 (子目录) 最多可以有三个子目录:
lib - the plugin's jar(s) libext - the plugin's dependency jar(s) native - any required native libraries, such as .so files
plugins.d 目录中的两个插件示例:
plugins.d/ plugins.d/custom-source-1/ plugins.d/custom-source-1/lib/my-source.jar plugins.d/custom-source-1/libext/spring-core-2.5.6.jar plugins.d/custom-source-2/ plugins.d/custom-source-2/lib/custom.jar plugins.d/custom-source-2/native/gettext.so
数据摄取
Flume 支持许多从外部来 source 摄取数据的机制.
RPC
Flume 发行版中包含的 Avro 客户端可以使用 avro RPC 机制将给定文件发送到 Flume Avrosource:
$ bin/flume-ng avro-client -H localhost -p 41414 -F /usr/logs/log.10
上面的命令会将 /usr/logs/log.10 的内容发送到监听该端口的 Flume source.
执行命令
有一个 exec source 执行给定的命令并消耗输出. 输出的单 "行" 即. 文本后跟回车符 (' r') 或换行符 (' n') 或两者一起.
网络流
Flume 支持以下机制从常用日志流类型中读取数据, 例如:
- Avro
- Thrift
- Syslog
- Netcat
设置多 agent 流程

为了跨多个 agent 或跳数据流, 先前 agent 的 sink 和当前跳的 source 需要是 avro 类型, sink 指向 source 的主机名 (或 IP 地址) 和端口.
合并
日志收集中非常常见的情况是大量日志生成客户端将数据发送到连接到存储子系统的少数消费者 agent . 例如, 从数百个 Web 服务器收集的日志发送给写入 HDFS 集群的十几个 agent .

这可以通过使用 avrosink 配置多个第一层 agent 在 Flume 中实现, 所有这些 agent 都指向单个 agent 的 avrosource(同样, 您可以在这种情况下使用 thriftsource/sink / 客户端). 第二层 agent 上的此 source 将接收的 event 合并到单个信道中, 该信道由信宿器消耗到其最终目的地.
多路复用流程
Flume 支持将 event 流多路复用到一个或多个目的地. 这是通过定义可以复制或选择性地将 event 路由到一个或多个信道的流复用器来实现的.

上面的例子显示了来自 agent "foo" 的 source 代码将流程扩展到三个不同的 channel. 扇出可以复制或多路复用. 在复制流的情况下, 每个 event 被发送到所有三个 channel. 对于多路复用情况, 当 event 的属性与预配置的值匹配时, event 将被传递到可用 channel 的子集. 例如, 如果一个名为 "txnType" 的 event 属性设置为 "customer", 那么它应该转到 channel1 和 channel3, 如果它是 "vendor", 那么它应该转到 channel2, 否则转到 channel3. 可以在 agent 的配置文件中设置映射.
配置
如前面部分所述, Flume agent 程序配置是从类似于具有分层属性设置的 Java 属性文件格式的文件中读取的.
定义流程
要在单个 agent 中定义流, 您需要通过 channel 链接 source 和 sink. 您需要列出给定 agent 的 source,sink 和 channel, 然后将 source 和 sink 指向 channel.source 实例可以指定多个 channel, 但 sink 实例只能指定一个 channel. 格式如下:
# list the sources, sinks and channels for the agent <Agent>.sources = <Source> <Agent>.sinks = <Sink> <Agent>.channels = <Channel1> <Channel2> # set channel for source <Agent>.sources.<Source>.channels = <Channel1> <Channel2> ... # set channel for sink <Agent>.sinks.<Sink>.channel = <Channel1>
例如, 名为 agent_foo 的 agent 正在从外部 avro 客户端读取数据并通过内存 channel 将其发送到 HDFS.
配置文件 weblog.config 可能如下所示:
# list the sources, sinks and channels for the agent agent_foo.sources = avro-appserver-src-1 agent_foo.sinks = hdfs-sink-1 agent_foo.channels = mem-channel-1 # set channel for source agent_foo.sources.avro-appserver-src-1.channels = mem-channel-1 # set channel for sink agent_foo.sinks.hdfs-sink-1.channel = mem-channel-1
这将使 event 从 avro-AppSrv-source 流向 hdfs-Cluster1-sink, 通过内存 channelmem-channel-1.
当使用 weblog.config 作为其配置文件启动 agent 程序时, 它将实例化该流程.
配置单个组件
定义流后, 您需要设置每个 source,sink 和 channel 的属性. 这是以相同的分层命名空间方式完成的, 您可以在其中设置组件类型以及特定于每个组件的属性的其他值:
# properties for sources <Agent>.sources.<Source>.<someProperty> = <someValue> # properties for channels <Agent>.channel.<Channel>.<someProperty> = <someValue> # properties for sinks <Agent>.sources.<Sink>.<someProperty> = <someValue>
需要为 Flume 的每个组件设置属性 "type", 以了解它需要什么类型的对象. 每个 source,sink 和 channel 类型都有自己的一组属性, 使其能够按预期运行. 所有这些都需要根据需要进行设置. 在前面的示例中, 我们有一个从 avro-AppSrv-source 到 hdfs-Cluster1-sink 的流程通过内存 channelmem-channel-1. 这是一个显示每个组件配置的示例:
agent_foo.sources = avro-AppSrv-source agent_foo.sinks = hdfs-Cluster1-sink agent_foo.channels = mem-channel-1 # set channel for sources, sinks # properties of avro-AppSrv-source agent_foo.sources.avro-AppSrv-source.type = avro agent_foo.sources.avro-AppSrv-source.bind = localhost agent_foo.sources.avro-AppSrv-source.port = 10000 # properties of mem-channel-1 agent_foo.channels.mem-channel-1.type = memory agent_foo.channels.mem-channel-1.capacity = 1000 agent_foo.channels.mem-channel-1.transactionCapacity = 100 # properties of hdfs-Cluster1-sink agent_foo.sinks.hdfs-Cluster1-sink.type = hdfs agent_foo.sinks.hdfs-Cluster1-sink.hdfs.path = hdfs://namenode/flume/webdata #...
在 agent 中添加多个流
单个 Flume agent 可以包含多个独立流. 您可以在配置中列出多个 source,sink 和 channel. 可以链接这些组件以形成多个流:
# list the sources, sinks and channels for the agent <Agent>.sources = <Source1> <Source2> <Agent>.sinks = <Sink1> <Sink2> <Agent>.channels = <Channel1> <Channel2>
然后, 您可以将 source 和 sink 链接到 channel(用于 sink)的相应 channel(用于 source), 以设置两个不同的流. 例如, 如果您需要在 agent 中设置两个流, 一个从外部 avro 客户端到外部 HDFS, 另一个从尾部输出到 avrosink, 那么这是一个配置来执行此操作:
# list the sources, sinks and channels in the agent agent_foo.sources = avro-AppSrv-source1 exec-tail-source2 agent_foo.sinks = hdfs-Cluster1-sink1 avro-forward-sink2 agent_foo.channels = mem-channel-1 file-channel-2 # flow #1 configuration agent_foo.sources.avro-AppSrv-source1.channels = mem-channel-1 agent_foo.sinks.hdfs-Cluster1-sink1.channel = mem-channel-1 # flow #2 configuration agent_foo.sources.exec-tail-source2.channels = file-channel-2 agent_foo.sinks.avro-forward-sink2.channel = file-channel-2
配置多 agent 流程
要设置多层流, 您需要有 sink 指向下一跳的 avro/thrift source. 这将导致第一个 Flume agent 将 event 转发到下一个 Flume agent . 例如, 如果您使用 avro 客户端定期向本地 Flume agent 发送文件(每个 event 1 个文件), 则此本地 agent 可以将其转发到另一个已安装存储的 agent
Weblog agent 配置:
# list sources, sinks and channels in the agent agent_foo.sources = avro-AppSrv-source agent_foo.sinks = avro-forward-sink agent_foo.channels = file-channel # define the flow agent_foo.sources.avro-AppSrv-source.channels = file-channel agent_foo.sinks.avro-forward-sink.channel = file-channel # avro sink properties agent_foo.sinks.avro-forward-sink.type = avro agent_foo.sinks.avro-forward-sink.hostname = 10.1.1.100 agent_foo.sinks.avro-forward-sink.port = 10000 # configure other pieces #...
HDFS agent 配置:
# list sources, sinks and channels in the agent agent_foo.sources = avro-collection-source agent_foo.sinks = hdfs-sink agent_foo.channels = mem-channel # define the flow agent_foo.sources.avro-collection-source.channels = mem-channel agent_foo.sinks.hdfs-sink.channel = mem-channel # avro source properties agent_foo.sources.avro-collection-source.type = avro agent_foo.sources.avro-collection-source.bind = 10.1.1.100 agent_foo.sources.avro-collection-source.port = 10000 # configure other pieces #...
在这里, 我们将 weblog agent 的 avro-forward-sink 链接到 hdfs agent 的 avro-collection-source. 这将导致来自外部应用程序服务器 source 的 event 最终存储在 HDFS 中.
扇出流量
如前一节所述, Flume 支持扇出从一个 source 到多个 channel 的流量. 扇出有两种模式 : 复制和多路复用. 在复制流程中, event 将发送到所有已配置的 channel. 在多路复用的情况下, event 仅被发送到合格 channels 的子集. 为了散开流量, 需要指定 source 的 channel 列表以及扇出它的策略. 这是通过添加可以复制或多路复用的 channel"选择器" 来完成的. 如果它是多路复用器, 则进一步指定选择规则. 如果您没有指定选择器, 那么默认情况下它会复制:
# List the sources, sinks and channels for the agent <Agent>.sources = <Source1> <Agent>.sinks = <Sink1> <Sink2> <Agent>.channels = <Channel1> <Channel2> # set list of channels for source (separated by space) <Agent>.sources.<Source1>.channels = <Channel1> <Channel2> # set channel for sinks <Agent>.sinks.<Sink1>.channel = <Channel1> <Agent>.sinks.<Sink2>.channel = <Channel2> <Agent>.sources.<Source1>.selector.type = replicating
多路复用选择具有另一组属性以分流流. 这需要指定 event 属性到 channel 集的映射. 选择器检查 event 头中的每个已配置属性. 如果它与指定的值匹配, 则该 event 将发送到映射到该值的所有 channel. 如果没有匹配项, 则将 event 发送到配置为默认值的 channel 集:
# Mapping for multiplexing selector <Agent>.sources.<Source1>.selector.type = multiplexing <Agent>.sources.<Source1>.selector.header = <someHeader> <Agent>.sources.<Source1>.selector.mapping.<Value1> = <Channel1> <Agent>.sources.<Source1>.selector.mapping.<Value2> = <Channel1> <Channel2> <Agent>.sources.<Source1>.selector.mapping.<Value3> = <Channel2> #... <Agent>.sources.<Source1>.selector.default = <Channel2>
映射允许为每个值重叠 channel.
以下示例具有多路复用到两个路径的单个流. 名为 agent_foo 的 agent 具有单个 avrosource 和两个链接到两个 sink 的 channel:
# list the sources, sinks and channels in the agent agent_foo.sources = avro-AppSrv-source1 agent_foo.sinks = hdfs-Cluster1-sink1 avro-forward-sink2 agent_foo.channels = mem-channel-1 file-channel-2 # set channels for source agent_foo.sources.avro-AppSrv-source1.channels = mem-channel-1 file-channel-2 # set channel for sinks agent_foo.sinks.hdfs-Cluster1-sink1.channel = mem-channel-1 agent_foo.sinks.avro-forward-sink2.channel = file-channel-2 # channel selector configuration agent_foo.sources.avro-AppSrv-source1.selector.type = multiplexing agent_foo.sources.avro-AppSrv-source1.selector.header = State agent_foo.sources.avro-AppSrv-source1.selector.mapping.CA = mem-channel-1 agent_foo.sources.avro-AppSrv-source1.selector.mapping.AZ = file-channel-2 agent_foo.sources.avro-AppSrv-source1.selector.mapping.NY = mem-channel-1 file-channel-2 agent_foo.sources.avro-AppSrv-source1.selector.default = mem-channel-1
选择器检查名为 "State" 的标头. 如果该值为 "CA", 则将其发送到 mem-channel-1, 如果其为 "AZ", 则将其发送到文件 channel-2, 或者如果其为 "NY" 则为两者. 如果 "状态" 标题未设置或与三者中的任何一个都不匹配, 则它将转到 mem-channel-1, 其被指定为 "default".
选择器还支持可选 channel. 要为标头指定可选 channel, 可通过以下方式使用 config 参数 "optional":
# channel selector configuration agent_foo.sources.avro-AppSrv-source1.selector.type = multiplexing agent_foo.sources.avro-AppSrv-source1.selector.header = State agent_foo.sources.avro-AppSrv-source1.selector.mapping.CA = mem-channel-1 agent_foo.sources.avro-AppSrv-source1.selector.mapping.AZ = file-channel-2 agent_foo.sources.avro-AppSrv-source1.selector.mapping.NY = mem-channel-1 file-channel-2 agent_foo.sources.avro-AppSrv-source1.selector.optional.CA = mem-channel-1 file-channel-2 agent_foo.sources.avro-AppSrv-source1.selector.mapping.AZ = file-channel-2 agent_foo.sources.avro-AppSrv-source1.selector.default = mem-channel-1
选择器将首先尝试写入所需的 channel, 如果其中一个 channel 无法使用 event , 则会使事务失败. 在所有渠道上重新尝试交易. 一旦所有必需的 channel 消耗了 event , 则选择器将尝试写入可选 channel. 任何可选 channel 使用该 event 的失败都会被忽略而不会重试.
如果可选信道与特定报头的所需信道之间存在重叠, 则认为该信道是必需的, 并且信道中的故障将导致重试所有必需信道集. 例如, 在上面的示例中, 对于标题 "CA",mem-channel-1 被认为是必需的 channel, 即使它被标记为必需和可选, 并且写入此 channel 的失败将导致该 event 在为选择器配置的所有 channel 上重试.
请注意, 如果标头没有任何所需的 channel, 则该 event 将被写入默认 channel, 并将尝试写入该标头的可选 channel. 如果未指定所需的 channel, 则指定可选 channel 仍会将 event 写入默认 channel. 如果没有将 channel 指定为默认 channel 且没有必需 channel, 则选择器将尝试将 event 写入可选 channel. 在这种情况下, 任何失败都会被忽略.
支持
多个 Flume 组件支持 SSL / TLS 协议, 以便安全地与其他系统通信.
| Component | SSL server or client |
|---|---|
| Avro Source | server |
| Avro Sink | client |
| Thrift Source | server |
| Thrift Sink | client |
| Kafka Source | client |
| Kafka Channel | client |
| Kafka Sink | client |
| HTTP Source | server |
| JMS Source | client |
| Syslog TCP Source | server |
| Multiport Syslog TCP Source | server |
SSL 兼容组件具有若干配置参数来设置 SSL, 例如启用 SSL 标志, 密钥库 / 信任库参数 (位置, 密码, 类型) 和其他 SSL 参数(例如禁用的协议)
始终在 agent 配置文件的组件级别指定为组件启用 SSL. 因此, 某些组件可能配置为使用 SSL, 而其他组件则不配置(即使具有相同的组件类型)
密钥库 / 信任库设置可以在组件级别或全局指定.
在组件级别设置的情况下, 通过组件特定参数在 agent 配置文件中配置密钥库 / 信任库. 此方法的优点是组件可以使用不同的密钥库(如果需要). 缺点是必须为 agent 配置文件中的每个组件复制密钥库参数. 组件级别设置是可选的, 但如果已定义, 则其优先级高于全局参数.
使用全局设置, 只需定义一次密钥库 / 信任库参数, 并对所有组件使用相同的设置, 这意味着更少和更集中的配置.
可以通过系统属性或通过环境变量来配置全局设置.
| 系统属性 | 环境变量 | 描述 |
|---|---|---|
| javax.net.ssl.keyStore | FLUME_SSL_KEYSTORE_PATH | 密钥库位置 |
| javax.net.ssl.keyStorePassword | FLUME_SSL_KEYSTORE_PASSWORD | 密钥库密码 |
| javax.net.ssl.keyStoreType | FLUME_SSL_KEYSTORE_TYPE | 密钥库类型（默认为 JKS） |
| javax.net.ssl.trustStore | FLUME_SSL_TRUSTSTORE_PATH | 信任库位置 |
| javax.net.ssl.trustStorePassword | FLUME_SSL_TRUSTSTORE_PASSWORD | 信任库密码 |
| javax.net.ssl.trustStoreType | FLUME_SSL_TRUSTSTORE_TYPE | 信任库类型（默认为 JKS） |
| flume.ssl.include.protocols | FLUME_SSL_INCLUDE_PROTOCOLS | 计算启用的协议时要包括的协议。逗号（，）分隔列表。如果提供，排除的协议将从此列表中排除。 |
| flume.ssl.exclude.protocols | FLUME_SSL_EXCLUDE_PROTOCOLS | 计算启用的协议时要排除的协议。逗号（，）分隔列表。 |
| flume.ssl.include.cipherSuites | FLUME_SSL_INCLUDE_CIPHERSUITES | 在计算启用的密码套件时包含的密码套件。逗号（，）分隔列表。如果提供，排除的密码套件将被排除在此列表之外。 |
| flume.ssl.exclude.cipherSuites | FLUME_SSL_EXCLUDE_CIPHERSUITES | 在计算启用的密码套件时要排除的密码套件。逗号（，）分隔列表。 |
可以在命令行上传递 SSL 系统属性, 也可以在 conf / flume-env.sh 中设置 JAVA_OPTS 环境变量(尽管使用命令行是不可取的, 因为包含密码的命令将保存到命令历史记录中.)
export JAVA_OPTS="$JAVA_OPTS -Djavax.net.ssl.keyStore=/path/to/keystore.jks" export JAVA_OPTS="$JAVA_OPTS -Djavax.net.ssl.keyStorePassword=password"
Flume 使用 JSSE(Java 安全套接字扩展)中定义的系统属性, 因此这是设置 SSL 的标准方法. 另一方面, 在系统属性中指定密码意味着可以在进程列表中看到密码. 对于不可接受的情况, 也可以在环境变量中定义参数. 在这种情况下, Flume 在内部从相应的环境变量初始化 JSSE 系统属性.
SSL 环境变量可以在启动 Flume 之前在 shell 环境中设置, 也可以在 conf / flume-env.sh 中设置(尽管使用命令行是不可取的, 因为包含密码的命令将保存到命令历史记录中.)
export FLUME_SSL_KEYSTORE_PATH=/path/to/keystore.jks
export FLUME_SSL_KEYSTORE_PASSWORD=password
** 请注意:**
必须在组件级别启用 SSL. 仅指定全局 SSL 参数不会产生任何影响.
如果在多个级别指定全局 SSL 参数, 则优先级如下(从高到低):
agent 配置中的组件参数
系统属性
环境变量
如果为组件启用了 SSL, 但未以上述任何方式指定 SSL 参数, 则
在密钥库的情况下: 配置错误
在 truststores 的情况下: 将使用默认信任库(Oracle JDK 中的 jssecacerts / cacerts)
在所有情况下, 可信任密码都是可选的. 如果未指定, 则在 JDK 打开信任库时, 不会对信任库执行完整性检查.
source 和接收批量大小和 channel 事务容量
source 和 sink 可以具有批量大小参数, 该参数确定它们在一个批次中处理的最大 event 数. 这发生在具有称为事务容量的上限的 channel 事务中. 批量大小必须小于渠道的交易容量. 有一个明确的检查, 以防止不兼容的设置. 只要读取配置, 就会进行此检查.
- Flume Source
- Avro Source
监听 Avro 端口并从外部 Avro 客户端流接收 event . 当与另一个(上一跳)Flume agent 上的内置 Avro Sink 配对时, 它可以创建分层集合拓扑. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 avro |
| bind | - | 要侦听的主机名或 IP 地址 |
| port | - | 要绑定的端口号 |
| threads | - | 生成的最大工作线程数 |
| selector.type | ||
| selector.* | ||
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* | ||
| compression-type | none | 这可以是 “none” 或“deflate”。压缩类型必须与匹配 AvroSource 的压缩类型匹配 |
| SSL | false | 将其设置为 true 以启用 SSL 加密。如果启用了 SSL，则还必须通过组件级参数（请参阅下文）或全局 SSL 参数（请参阅 SSL / TLS 支持部分）指定 “密钥库” 和“密钥库密码” 。 |
| keysore | - | 这是 Java 密钥库文件的路径。如果未在此处指定，则将使用全局密钥库（如果已定义，则配置错误）。 |
| keystore-password | - | Java 密钥库的密码。如果未在此处指定，则将使用全局密钥库密码（如果已定义，则配置错误）。 |
| keystore-type | JKS | Java 密钥库的类型。这可以是 “JKS” 或“PKCS12”。如果未在此处指定，则将使用全局密钥库类型（如果已定义，则默认为 JKS）。 |
| exclude-protocols | SSLv3 | 要排除的以空格分隔的 SSL / TLS 协议列表。除指定的协议外，将始终排除 SSLv3。 |
| include-protocols | - | 要包含的以空格分隔的 SSL / TLS 协议列表。启用的协议将是包含的协议，没有排除的协议。如果包含协议为空，则它包括每个支持的协议。 |
| exclude-cipher-suites | - | 要排除的以空格分隔的密码套件列表。 |
| include-cipher-suites | - | 以空格分隔的密码套件列表。启用的密码套件将是包含的密码套件，不包括排除的密码套件。如果 included-cipher-suites 为空，则包含每个支持的密码套件。 |
| ipFilter | false | 将此设置为 true 以启用 ipFiltering for netty |
| ipFilterRules | - | 使用此配置定义 N netty ipFilter 模式规则。 |
agent 名为 a1 的示例:
a1.sources = r1 a1.channels = c1 a1.sources.r1.type = avro a1.sources.r1.channels = c1 a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 4141
ipFilterRules 的示例
ipFilterRules 定义由逗号分隔的 N 个 netty ipFilters 模式规则必须采用此格式.
<'allow' or deny>:<'ip' or 'name' for computer name>:<pattern> or allow/deny:ip/name:pattern example: ipFilterRules=allow:ip:127.*,allow:name:localhost,deny:ip:*
请注意, 匹配的第一个规则将适用, 如下例所示, 来自 localhost 上的客户端
这将允许 localhost 上的客户端拒绝来自任何其他 ip 的客户端 "allow:name:localhost,deny:ip: 这将拒绝 localhost 上的客户端允许来自任何其他 ip 的客户端"deny:name:localhost,allow:ip:
Thrift Source
侦听 Thrift 端口并从外部 Thrift 客户端流接收 event . 当与另一个(上一跳)Flume agent 上的内置 ThriftSink 配对时, 它可以创建分层集合拓扑. 可以通过启用 kerberos 身份验证将 Thriftsource 配置为以安全模式启动. agent-principal 和 agent-keytab 是 Thriftsource 用于向 kerberos KDC 进行身份验证的属性. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要节俭 |
| bind | - | 要侦听的主机名或 IP 地址 |
| port | - | 要绑定的端口号 |
| threads | - | 生成的最大工作线程数 |
| selector.type | ||
| selector.* | ||
| interceptors | - | 空格分隔的拦截器列表 |
| interceptors.* | ||
| SSL | false | 将其设置为 true 以启用 SSL 加密。如果启用了 SSL，则还必须通过组件级参数（请参阅下文）或全局 SSL 参数（请参阅 SSL / TLS 支持部分）指定 “密钥库” 和“密钥库密码”。 |
| keystore | - | 这是 Java 密钥库文件的路径。如果未在此处指定，则将使用全局密钥库（如果已定义，则配置错误）。 |
| keystore-password | - | Java 密钥库的密码。如果未在此处指定，则将使用全局密钥库密码（如果已定义，则配置错误）。 |
| keystore-type | JKS | Java 密钥库的类型。这可以是 “JKS” 或“PKCS12”。如果未在此处指定，则将使用全局密钥库类型（如果已定义，则默认为 JKS）。 |
| exclude-protocols | 要排除的以空格分隔的 SSL / TLS 协议列表。除指定的协议外，将始终排除 SSLv3。 | |
| include-protocols | - | 要包含的以空格分隔的 SSL / TLS 协议列表。启用的协议将是包含的协议，没有排除的协议。如果包含协议为空，则它包括每个支持的协议。 |
| exclude-cipher-suites | - | 要排除的以空格分隔的密码套件列表。 |
| include-cipher-suites | - | 以空格分隔的密码套件列表。启用的密码套件将是包含的密码套件，不包括排除的密码套件。 |
| kerberos | 设置为 true 以启用 kerberos 身份验证。在 kerberos 模式下，成功进行身份验证需要 agent-principal 和 agent-keytab。安全模式下的 Thriftsource 仅接受来自已启用 kerberos 且已成功通过 kerberos KDC 验证的 Thrift 客户端的连接。 | |
| agent-principal | - | Thrift Source 使用的 kerberos 主体对 kerberos KDC 进行身份验证。 |
| agent-keytab | - | Thrift Source 与 agent 主体结合使用的 keytab 位置，用于对 kerberos KDC 进行身份验证。 |
agent 名为 a1 的示例:
a1.sources = r1 a1.channels = c1 a1.sources.r1.type = thrift a1.sources.r1.channels = c1 a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 4141 Exec Source
Exec source 在启动时运行给定的 Unix 命令, 并期望该进程在标准输出上连续生成数据 (stderr 被简单地丢弃, 除非属性 logStdErr 设置为 true). 如果进程因任何原因退出, 则 source 也会退出并且不会生成其他数据. 这意味着 cat [named pipe] 或 tail -F [file] 等配置将产生所需的结果, 而日期 可能不会 - 前两个命令产生数据流, 而后者产生单个 event 并退出. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 exec |
| command | - | 要执行的命令 |
| shell | - | 用于运行命令的 shell 调用。例如 /bin/sh -c 。仅适用于依赖 shell 功能的命令，如通配符，后退标记，管道等。 |
| restartThrottle | 10000 | 尝试重新启动之前等待的时间（以毫秒为单位） |
| restart | false | 是否应该重新执行已执行的 cmd |
| logStdErr | false | 是否应记录命令的 stderr |
| BATCHSIZE | 20 | 一次读取和发送到 channel 的最大行数 |
| batchTimeout | 3000 | 在向下游推送数据之前，如果未达到缓冲区大小，则等待的时间（以毫秒为单位） |
| selector.type | replication | 复制或多路复用 |
| selector.* | 取决于 selector.type 值 | |
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* |
警告
Exec Source 和其他异步 source 的问题在于, 如果无法将 event 放入 Channel 中, 则 source 无法保证客户端知道它. 在这种情况下, 数据将丢失. 例如, 最常请求的功能之一是 tail -F [file] 类似用例, 其中应用程序写入磁盘上的日志文件, Flume 将文件作为尾部发送, 将每一行作为 event 发送. 虽然这是可能的, 但是有一个明显的问题; 如果 channel 填满并且 Flume 无法发送 event , 会发生什么?由于某种原因, Flume 无法向编写日志文件的应用程序指示它需要保留日志或 event 尚未发送. 如果这没有意义, 您只需要知道: 当使用 Exec Source 等单向异步接口时, 您的应用程序永远无法保证已收到数据!
作为此警告的延伸 - 并且完全清楚 - 使用此 source 时, event 传递绝对没有保证.
为了获得更强的可靠性保证, 请考虑 Spooling Directory Source,Taildir Source 或通过 SDK 直接与 Flume 集成.
agent 名为 a1 的示例:
a1.sources = r1 a1.channels = c1 a1.sources.r1.type = exec a1.sources.r1.command = tail -F /var/log/secure a1.sources.r1.channels = c1
'shell'配置用于通过命令 shell(例如 Bash 或 Powershell)调用'命令'.'command'作为参数传递给'shell'执行. 这允许'命令'使用 shell 中的功能, 例如通配符, 后退标记, 管道, 循环, 条件等. 如果没有'shell'配置, 将直接调用'command'.'shell'的常用值: '/bin/sh -c', '/bin/ksh -c', 'cmd /c', 'powershell -Command', etc.
a1.sources.tailsource-1.type = exec a1.sources.tailsource-1.shell = /bin/bash -c a1.sources.tailsource-1.command = for i in /path/*.txt; do cat $i; done JMS Source
JMS Source 从 JMS 目标 (例如队列或主题) 读取消息. 作为 JMS 应用程序, 它应该与任何 JMS 提供程序一起使用, 但仅使用 ActiveMQ 进行测试. JMSsource 提供可配置的批量大小, 消息选择器, 用户 / 传递和消息到水槽 event 转换器. 请注意, 供应商提供的 JMS jar 应该包含在 Flume 类路径中, 使用 plugins.d 目录(首选), 命令行上的 - classpath 或 flume-env.sh 中的 FLUME_CLASSPATH 变量. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 jms |
| initialContextFactory | - | Inital Context Factory，例如：org.apache.activemq.jndi.ActiveMQInitialContextFactory |
| connectionFactory | - | 连接工厂应显示为的 JNDI 名称 |
| providerURL | - | JMS 提供程序 URL |
| destinationName | - | 目的地名称 |
| destinationType | - | 目的地类型（队列或主题） |
| messageSelector | - | 创建使用者时使用的消息选择器 |
| userName | - | 目标 / 提供商的用户名 |
| PASSWORDFILE | - | 包含目标 / 提供程序密码的文件 |
| BATCHSIZE | 100 | 一批中要使用的消息数 |
| converter.type | DEFAULT | 用于将消息转换为水槽 event 的类。见下文。 |
| converter.* | - | 转换器属性。 |
| converter.charset | UTF-8 | 仅限默认转换器。在将 JMS TextMessages 转换为字节数组时使用的字符集。 |
| createDurableSubscription | false | 是否创建持久订阅。持久订阅只能与 destinationType 主题一起使用。如果为 true，则必须指定 “clientId” 和“durableSubscriptionName”。 |
| clientId | - | JMS 客户端标识符在创建后立即在 Connection 上设置。持久订阅必需。 |
| durableSubscriptionName | - | 用于标识持久订阅的名称。持久订阅必需。 |
消息转换器
JMSsource 允许可插拔转换器, 尽管默认转换器可能适用于大多数用途. 默认转换器能够将字节, 文本和对象消息转换为 FlumeEvents. 在所有情况下, 消息中的属性都将作为标题添加到 FlumeEvent 中.
BytesMessage:
消息的字节被复制到 FlumeEvent 的主体. 每封邮件无法转换超过 2GB 的数据.
TextMessage 的:
消息文本转换为字节数组并复制到 FlumeEvent 的主体. 默认转换器默认使用 UTF-8, 但这是可配置的.
ObjectMessage:
Object 被写入包含在 ObjectOutputStream 中的 ByteArrayOutputStream, 并将生成的数组复制到 FlumeEvent 的主体.
agent 名为 a1 的示例:
a1.sources = r1 a1.channels = c1 a1.sources.r1.type = jms a1.sources.r1.channels = c1 a1.sources.r1.initialContextFactory = org.apache.activemq.jndi.ActiveMQInitialContextFactory a1.sources.r1.connectionFactory = GenericConnectionFactory a1.sources.r1.providerURL = tcp://mqserver:61616 a1.sources.r1.destinationName = BUSINESS_DATA a1.sources.r1.destinationType = QUEUE
SSL 和 JMS Source
JMS 客户端实现通常支持通过 JSSE(Java 安全套接字扩展)定义的某些 Java 系统属性来配置 SSL / TLS. 为 Flume 的 JVM 指定这些系统属性, JMSsource(或更准确地说是 JMSsource 使用的 JMS 客户端实现)可以通过 SSL 连接到 JMS 服务器(当然, 仅当 JMS 服务器也已设置为使用 SSL 时). 它应该可以与任何 JMS 提供程序一起使用, 并且已经过 ActiveMQ,IBM MQ 和 Oracle WebLogic 的测试.
以下部分仅介绍 Flume 方面所需的 SSL 配置步骤. 您可以在 Flume Wiki 上找到有关不同 JMS 提供程序的服务器端设置以及完整工作配置示例的更详细说明.
** SSL 传输 / 服务器身份验证:**
如果 JMS 服务器使用自签名证书或其证书由不受信任的 CA(例如公司自己的 CA)签名, 则需要设置信任库 (包含正确的证书) 并传递给 Flume. 它可以通过全局 SSL 参数完成. 有关全局 SSL 设置的更多详细信息, 请参阅 SSL / TLS 支持部分.
使用 SSL 时, 某些 JMS 提供程序需要 SSL 特定的 JNDI 初始上下文工厂和 / 或提供程序 URL 设置(例如, ActiveMQ 使用 ssl:// URL 前缀而不是 tcp://). 在这种情况下, 必须在 agent 配置文件中调整 source 属性(initialContextFactory 和 / 或 providerURL)
客户端证书身份验证(双向 SSL):
JMS Source 可以通过客户端证书身份验证而不是通常的用户 / 密码登录来对 JMS 服务器进行身份验证(使用 SSL 并且 JMS 服务器配置为接受此类身份验证时).
包含用于身份验证的 Flume 密钥的密钥库需要再次通过全局 SSL 参数进行配置. 有关全局 SSL 设置的更多详细信息, 请参阅 SSL / TLS 支持部分.
密钥库应该只包含一个密钥(如果存在多个密钥, 则将使用第一个密钥). 密钥密码必须与密钥库密码相同.
在客户端证书身份验证的情况下, 不需要在 Flume agent 配置文件中为 JMSsource 指定 userName / passwordFile 属性.
请注意:
与其他组件不同, JMS Source 没有组件级别的配置参数. 也没有启用 SSL 标志. SSL 设置由 JNDI / Provider URL 设置 (最终是 JMS 服务器设置) 以及 truststore / keystore 的存在 / 不存在控制.
Spooling Directory Source
此 source 允许您通过将要摄取的文件放入磁盘上的 "spooling" 目录来摄取数据. 此 source 将查看新文件的指定目录, 并将在新文件出现时解析 event .event 解析逻辑是可插入的. 在给定文件完全读入 channel 后, 默认情况下通过重命名文件来指示完成, 或者可以删除它或使用 trackerDir 来跟踪已处理的文件.
与 Exec source 不同, 即使 Flume 重新启动或被杀死, 此 source 也是可靠的并且不会遗漏数据. 作为这种可靠性的交换, 只有不可变的, 唯一命名的文件必须被放入 spooling directory. 中. Flume 试图检测这些问题, 如果违反则会声明失败:
如果在放入 spooling directory 后写入文件, Flume 会将错误打印到其日志文件并停止处理.
如果稍后重复使用文件名, Flume 将在其日志文件中输出错误并停止处理.
为避免上述问题, 在将文件名移动到 spooling directory 中时, 添加唯一标识符 (例如时间戳) 可能很有用.
尽管该 source 的可靠性保证, 但仍存在如果发生某些下游故障则可能重复 event 的情况. 这与其他 Flume 组件提供的保证一致.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称需要是 spooldir。 |
| spoolDir | - | 从中读取文件的目录。 |
| fileSuffix | .COMPLETED | 后缀附加到完全摄取的文件 |
| deletePolicy | never | 何时删除已完成的文件：从 never 或 immediate |
| FileHeader | false | 是否添加存储绝对路径文件名的标头。 |
| fileHeaderKey | file | 将绝对路径文件名附加到 event 标题时使用的标题键。 |
| basenameHeader | false | 是否添加存储文件基本名称的标头。 |
| basenameHeaderKey | basename | 标题将文件的基本名称附加到 event 标题时使用的标题。 |
| includePattern | .*$ | 正则表达式，指定要包含的文件。它可以与 ignorePattern 一起使用。如果一个文件同时匹配 ignorePattern 和 includePattern 正则表达式，该文件将被忽略。 |
| ignorePattern | $ | 正则表达式，指定要忽略的文件（跳过）。它可以与 includePattern 一起使用。如果一个文件同时匹配 ignorePattern 和 includePattern 正则表达式，该文件将被忽略。 |
| trackerDir | .flumespool | 用于存储与文件处理相关的元数据的目录。如果此路径不是绝对路径，则将其解释为相对于 spoolDir。 |
| trackingPolicy | rename | 跟踪策略定义如何跟踪文件处理。它可以是 “重 rename” 或“tracker_dir”。此参数仅在 deletePolicy 为 “never” 时有效。“重 rename” - 处理完文件后，会根据 fileSuffix 参数重命名。“tracker_dir” - 不重命名文件，但会在 trackerDir 中创建新的空文件。新的跟踪器文件名 source 自摄取的文件名和 fileSuffix。 |
| consumeOrder | oldest | spooling directory 中的文件将以 oldest， youngest 和 random 的方式使用。如果是 oldest 和 youngest 的，文件的最后修改时间将用于比较文件。如果出现相同，将首先消耗具有最小字典顺序的文件。在 random 的情况下，任何文件将被随机挑选。当使用 oldest 和 youngest 时，整个目录将被扫描以选择 oldest/youngest 的文件，如果存在大量文件，这可能会很慢，而使用 random 可能会导致旧文件在新文件不断进入时很晚被消耗 & nbsp;spooling directory。 |
| pollDelay | 500 | 轮询新文件时使用的延迟（以毫秒为单位）。 |
| recursiveDirectorySearch | false | 是否监视子目录以查找要读取的新文件。 |
| maxBackoff | 4000 | 如果 channel 已满，则在连续尝试写入 channel 之间等待的最长时间（以毫秒为单位）。source 将以低退避开始，并在每次 channel 抛出 ChannelException 时以指数方式增加，直到此参数指定的值。 |
| BATCHSIZE | 100 | 批量传输到 channel 的粒度 |
| inputCharset | UTF-8 | 反序列化器使用的字符集，将输入文件视为文本。 |
| decodeErrorPolicy | FAIL | 当我们在输入文件中看到不可解码的字符时该怎么办。FAIL：抛出异常并且无法解析文件。 REPLACE：用“替换字符”char 替换不可解析的字符，通常是 Unicode U+FFFD 。 IGNORE：删除不可解析的字符序列。 |
| deserializer | LINE | 指定用于将文件解析为 event 的反序列化程序。默认将每行解析为 event 。指定的类必须实现 EventDeserializer.Builder。 |
| deserializer.* | 每个 event 反序列化器不同。 | |
| bufferMaxLines | - | （Obselete）此选项现在被忽略。 |
| bufferMaxLineLength | 5000 | （已弃用）提交缓冲区中行的最大长度。请改用 deserializer.maxLineLength。 |
| selector.type | replicating | replicating or multiplexing |
| selector.* | 取决于 selector.type 值 | |
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* |
名为 agent-1 的 agent 示例:
a1.channels = ch-1 a1.sources = src-1 a1.sources.src-1.type = spooldir a1.sources.src-1.channels = ch-1 a1.sources.src-1.spoolDir = /var/log/apache/flumeSpool a1.sources.src-1.fileHeader = true Event Deserializers
以下 event 反序列化器随 Flume 一起提供.
LINE
此解串器为每行文本输入生成一个 event
| 物业名称 | 默认 | 描述 |
|---|---|---|
| deserializer.maxLineLength | 2048 | 单个 event 中包含的最大字符数。如果一行超过此长度，则会被截断，并且该行上的其余字符将出现在后续 event 中。 |
| deserializer.outputCharset | UTF-8 | 用于编码放入 channel 的 event 的字符集。 |
AVRO
此解串器能够读取 Avro 容器文件, 并在文件中为每个 Avro 记录生成一个 event . 每个 event 都使用标头进行注释, 该标头指示所使用的模式. event 的主体是二进制 Avro 记录数据, 不包括模式或容器文件元素的其余部分.
请注意, 如果假脱机目录 source 必须重试将其中一个 event 放入 channel(例如, 因为 channel 已满), 则它将重置并从最新的 Avro 容器文件同步点重试. 为了减少此类故障情况下的潜在 event 重复, 请在 Avro 输入文件中更频繁地写入同步标记.
| 物业名称 | 默认 | 描述 |
|---|---|---|
| deserializer.schemaType | HASH | 如何表示模式。默认情况下，或者 & nbsp; 指定值 HASH 时，会对 Avro 架构进行哈希处理，并将哈希值存储在 event 头 “flume.avro.schema.hash” 中的每个 event 中。如果指定了 LITERAL，则 JSON 编码的模式本身存储在 event 头 “flume.avro.schema.literal” 中的每个 event 中。与 HASH 模式相比，使用 LITERAL 模式效率相对较低。 |
BlobDeserializer>
此反序列化器为每个 event 读取二进制大对象(BLOB), 通常每个文件一个 BLOB. 例如 PDF 或 JPG 文件. 请注意, 此方法不适用于非常大的对象, 因为整个 BLOB 都缓存在 RAM 中.
| 物业名称 | 默认 | 描述 |
|---|---|---|
| 解串器 | - | 此类的 FQCN:org.apache.flume.sink.solr.morphline.BlobDeserializer$Builder |
| deserializer.maxBlobLength | 100000000 | 要读取的最大字节数和给定请求的缓冲区 |
Taildir Source
注意: 此 source 作为预览功能提供. 它不适用于 Windows.
观察指定的文件, 并在检测到添加到每个文件的新行后几乎实时地拖尾它们. 如果正在写入新行, 则此 source 将重试读取它们以等待写入完成.
此 source 是可靠的, 即使 tail 文件旋转也不会丢失数据. 它定期以 JSON 格式写入给定位置文件上每个文件的最后读取位置. 如果 Flume 由于某种原因停止或停止, 它可以从写在现有位置文件上的位置重新开始 tail.
在其他用例中, 此 source 也可以使用给定的位置文件从每个文件的任意位置开始拖尾. 当指定路径上没有位置文件时, 默认情况下它将从每个文件的第一行开始拖尾.
文件将按修改时间顺序使用. 将首先使用具有最早修改时间的文件.
此 source 不会重命名或删除或对正在挂载的文件执行任何修改. 目前此 source 不支持 tail 二进制文件. 它逐行读取文本文件.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 TAILDIR。 |
| filegroups | - | 以空格分隔的文件组列表。每个文件组都指示一组要挂起的文件。 |
| filegroups. | - | 文件组的绝对路径。正则表达式（而不是文件系统模式）只能用于文件名。 |
| positionFile | ~/.flume/taildir_position.json | 以 JSON 格式文件以记录每个尾部文件的 inode，绝对路径和最后位置。 |
| headers. | - | 标题值，使用标题键设置。可以为一个文件组指定多个标头。 |
| byteOffsetHeader | false | 是否将 tailed line 的字节偏移量添加到名为 “byteoffset” 的标头中。 |
| skipToEnd | false | 在未写入位置文件的文件的情况下是否跳过位置到 EOF。 |
| idleTimeout | 120000 | 关闭非活动文件的时间（毫秒）。如果关闭的文件附加了新行，则此 source 将自动重新打开它。 |
| writePosInterval | 3000 | 写入位置文件上每个文件的最后位置的间隔时间（ms）。 |
| BATCHSIZE | 100 | 一次读取和发送到 channel 的最大行数。使用默认值通常很好。 |
| maxBatchCount | Long.MAX_VALUE | 控制从同一文件连续读取的批次数。如果 source 正在拖尾多个文件，并且其中一个文件以快速写入，则可以防止处理其他文件，因为繁忙文件将在无限循环中读取。在这种情况下，降低此值。 |
| backoffSleepIncrement | 1000 | 在最后一次尝试未找到任何新数据时，重新尝试轮询新数据之前的时间延迟增量。 |
| maxBackoffSleep | 5000 | 每次重新尝试轮询新数据时的最大时间延迟，当最后一次尝试未找到任何新数据时。 |
| cachePatternMatching | true | 对于包含数千个文件的目录，列出目录并应用文件名正则表达式模式可能非常耗时。缓存匹配文件列表可以提高性能。消耗文件的顺序也将被缓存。要求文件系统以至少 1 秒的粒度跟踪修改时间。 |
| FileHeader | false | 是否添加存储绝对路径文件名的标头。 |
| fileHeaderKey | file | 将绝对路径文件名附加到 event 标题时使用的标题键。 |
agent 名为 a1 的示例:
a1.sources = r1 a1.channels = c1 a1.sources.r1.type = TAILDIR a1.sources.r1.channels = c1 a1.sources.r1.positionFile = /var/log/flume/taildir_position.JSON a1.sources.r1.filegroups = f1 f2 a1.sources.r1.filegroups.f1 = /var/log/test1/example.log a1.sources.r1.headers.f1.headerKey1 = value1 a1.sources.r1.filegroups.f2 = /var/log/test2/.*log.* a1.sources.r1.headers.f2.headerKey1 = value2 a1.sources.r1.headers.f2.headerKey2 = value2-2 a1.sources.r1.fileHeader = true a1.sources.ri.maxBatchCount = 1000 Kafka Source
Kafka Source 是一个 Apache Kafka 消费者, 它从 Kafka 主题中读取消息. 如果您运行了多个 Kafka source, 则可以使用相同的使用者组配置它们, 以便每个 source 都读取一组唯一的主题分区. 这目前支持 Kafka 服务器版本 0.10.1.0 或更高版本. 测试完成了 2.0.1, 这是发布时最高的可用版本.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 org.apache.flume.source.kafka.KafkaSource |
| kafka.bootstrap.servers | - | source 代码使用的 Kafka 集群中的 agent 列表 |
| kafka.consumer.group.id | flume | 独特的消费者群体。在多个 source 或 agent 中设置相同的 ID 表示它们是同一个使用者组的一部分 |
| kafka.topics | - | 以逗号分隔的主题列表，kafka 消费者将从中读取消息。 |
| kafka.topics.regex | - | 正则表达式，用于定义 source 订阅的主题集。此属性具有比 kafka.topics 更高的优先级，并覆盖 kafka.topics（如果存在）。 |
| BATCHSIZE | 1000 | 一批中写入 Channel 的最大消息数 |
| batchDurationMillis | 1000 | 将批次写入 channel 之前的最长时间（以毫秒为单位）只要达到第一个大小和时间，就会写入批次。 |
| backoffSleepIncrement | 1000 | Kafka Topic 显示为空时触发的初始和增量等待时间。等待时间将减少对空 kafka Topic 的激进 ping 操作。一秒钟是摄取用例的理想选择，但使用拦截器的低延迟操作可能需要较低的值。 |
| maxBackoffSleep | 5000 | Kafka Topic 显示为空时触发的最长等待时间。5 秒是摄取用例的理想选择，但使用拦截器的低延迟操作可能需要较低的值。 |
| useFlumeEventFormat | false | 默认情况下，event 将从 Kafka Topic 直接作为字节直接进入 event 主体。设置为 true 以读取 event 作为 Flume Avro 二进制格式。与 Kafka Sink 上的相同属性或 Kafka Channel 上的 parseAsFlumeEvent 属性一起使用时，这将保留在生成端发送的任何 Flume 标头。 |
| setTopicHeader | true | 设置为 true 时，将检索到的消息的主题存储到由 topic Header 属性定义的标头中 & nbsp;。 |
| topicHeader | topic | 如果 setTopicHeader 属性设置为 true，则定义用于存储接收消息主题名称的标头名称。如果与 Kafka SinktopicHeader 属性结合使用，应该小心，以避免将消息发送回循环中的同一主题。 |
| kafka.consumer.security.protocol | PLAINTEXT | 如果使用某种级别的安全性写入 Kafka，则设置为 SASL_PLAINTEXT,SASL_SSL 或 SSL。有关安全设置的其他信息，请参见下文。 |
| more consumer security props | 如果使用 SASL_PLAINTEXT,SASL_SSL 或 SSL，请参阅 Kafka 安全性以获取需要在使用者上设置的其他属性。 | |
| Other Kafka Consumer Properties | - | 这些属性用于配置 Kafka Consumer。可以使用 Kafka 支持的任何消费者财产。唯一的要求是使用前缀 kafka.consumer 添加属性名称 & nbsp;。例如：kafka.consumer.auto.offset.reset |
** 注意 **
Kafka Source 会覆盖两个 Kafka 使用者参数: source 和每次 batch 提交的时候会将 auto.commit.enable 设置为 "false", 并提交每个批处理. Kafka Source 至少保证一次消息检索策略.
source 启动时可以存在重复项.
Kafka Source 还提供了 key.deserializer(org.apache.kafka.common.serialization.StringSerializer)和 value.deserializer(org.apache.kafka.common.serialization.ByteArraySerializer)的默认值.
不建议修改这些参数.
不推荐使用的属性
| 属性名称 | 默认 | 描述 |
|---|---|---|
| topic | - | 使用 kafka.topics |
| groupId | flume | 使用 kafka.consumer.group.id |
| zookeeperConnect | - | 自 0.9.x 起不再受 kafka 消费者客户端的支持。使用 kafka.bootstrap.servers 与 kafka 集群建立连接 |
| migrateZookeeperOffsets | true | 如果找不到 Kafka 存储的偏移量，请在 Zookeeper 中查找偏移量并将它们提交给 Kafka。这应该是支持从旧版本的 Flume 无缝 Kafka 客户端迁移。迁移后，可以将其设置为 false，但通常不需要这样做。如果未找到 Zookeeper 偏移量，则 Kafka 配置 kafka.consumer.auto.offset.reset 定义如何处理偏移量。 有关详细信息，请查看 Kafka 文档 |
通过逗号分隔的主题列表进行主题订阅的示例:
tier1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource tier1.sources.source1.channels = channel1 tier1.sources.source1.batchSize = 5000 tier1.sources.source1.batchDurationMillis = 2000 tier1.sources.source1.kafka.Bootstrap.servers = localhost:9092 tier1.sources.source1.kafka.topics = test1, test2 tier1.sources.source1.kafka.consumer.group.id = custom.g.id
正则表达式主题订阅的示例:
ier1.source.source1.type = org.apache.flume.source.kafka.KafkaSource tier1.source.source1.channel = channel1
tier1.source.source1.kafka.Bootstrap.servers = 本地主机: 9092
tier1.source.source1.kafka .topics.regex = ^ topic [0-9] $
＃默认使用 kafka.consumer.group.id = flume
** Security 和 Kafka Source:**
Flume 和 Kafka 之间的通信渠道支持安全认证和数据加密. 对于安全身份验证, 可以使用 Kafka 0.9.0 版中的 SASL / GSSAPI(Kerberos V5)或 SSL(即使该参数名为 SSL, 实际协议是 TLS 实现)
截至目前, 数据加密仅由 SSL / TLS 提供.
将 kafka.consumer.security.protocol 设置为以下任何值意味着:
SASL_PLAINTEXT - 无数据加密的 Kerberos 或纯文本身份验证
SASL_SSL - 使用数据加密的 Kerberos 或纯文本身份验证
SSL - 基于 TLS 的加密, 带有可选的身份验证
** 警告 **
启用 SSL 时性能会下降, 其大小取决于 CPU 类型和 JVM 实现. 参考: Kafka 安全概述 和跟踪此问题的 jira: https://issues/jira/browse/KAFKA-2561
** TLS 和 Kafka Source:**
请阅读配置 Kafka 客户端 SSL 中描述的步骤, 以了解用于微调的其他配置设置, 例如以下任何一项: 安全提供程序, 密码套件, 启用的协议, 信任库或密钥库类型.
配置服务器端身份验证和数据加密的示例:
a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource a1.sources.source1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.sources.source1.kafka.topics = mytopic a1.sources.source1.kafka.consumer.group.id = flume-consumer a1.sources.source1.kafka.consumer.security.protocol = SSL # optional, the global truststore can be used alternatively a1.sources.source1.kafka.consumer.ssl.truststore.location=/path/to/truststore.jks a1.sources.source1.kafka.consumer.ssl.truststore.password=<password to access the truststore>
在此处指定信任库是可选的, 可以使用全局信任库. 有关全局 SSL 设置的更多详细信息, 请参阅 SSL / TLS 支持部分
注意: 默认情况下, 未定义属性 ssl.endpoint.identification.algorithm, 因此不会执行主机名验证. 要启用主机名验证, 请设置以下属性
a1.sources.source1.kafka.consumer.ssl.endpoint.identification.algorithm=HTTPS
启用后, 客户端将根据以下两个字段之一验证服务器的完全限定域名(FQDN):
通用名称(CN) https://tools.ietf.org/html/rfc6125#section-2.3 https://tools.ietf.org/html/rfc6125#section-2.3
主题备选名称(SAN)
如果还需要客户端身份验证, 则还需要将以下内容添加到 Flume agent 配置中, 或者可以使用全局 SSL 设置(请参阅 SSL / TLS 支持部分). 每个 Flume agent 都必须拥有其客户证书, Kafka 经纪人必须单独或通过其签名链来信任. 常见示例是由单个根 CA 签署每个客户端证书, 而后者又由 Kafka 经纪人信任.
# optional, the global keystore can be used alternatively a1.sources.source1.kafka.consumer.ssl.keystore.location=/path/to/client.keystore.jks a1.sources.source1.kafka.consumer.ssl.keystore.password=<password to access the keystore>
** Kerberos 和 Kafka source **
要将 Kafka source 与使用 Kerberos 保护的 Kafka 群集一起使用, 请为消费者设置上面提到的 consumer.security.protocol 属性.与 Kafka agent 一起使用的 Kerberos 密钥表和主体在 JAAS 文件的 "KafkaClient" 部分中指定."客户端" 部分描述了 Zookeeper 连接(如果需要). 有关 JAAS 文件内容的信息, 请参阅 Kafka doc. 可以通过 flume-env.sh 中的 JAVA_OPTS 指定此 JAAS 文件的位置以及可选的系统范围的 kerberos 配置:
JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/path/to/krb5.conf" JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/path/to/flume_jaas.conf"
使用 SASL_PLAINTEXT 的示例安全配置:
a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource a1.sources.source1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.sources.source1.kafka.topics = mytopic a1.sources.source1.kafka.consumer.group.id = flume-consumer a1.sources.source1.kafka.consumer.security.protocol = SASL_PLAINTEXT a1.sources.source1.kafka.consumer.sasl.mechanism = GSSAPI a1.sources.source1.kafka.consumer.sasl.kerberos.service.name = kafka
使用 SASL_SSL 的安全配置示例:
a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource a1.sources.source1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.sources.source1.kafka.topics = mytopic a1.sources.source1.kafka.consumer.group.id = flume-consumer a1.sources.source1.kafka.consumer.security.protocol = SASL_SSL a1.sources.source1.kafka.consumer.sasl.mechanism = GSSAPI a1.sources.source1.kafka.consumer.sasl.kerberos.service.name = kafka # optional, the global truststore can be used alternatively a1.sources.source1.kafka.consumer.ssl.truststore.location=/path/to/truststore.jks a1.sources.source1.kafka.consumer.ssl.truststore.password=<password to access the truststore>
示例 JAAS 文件. 有关其内容的参考, 请参阅 SASL 配置的 Kafka 文档中所需认证机制 (GSSAPI / PLAIN) 的客户端配置部分. 由于 Kafka Source 也可能连接到 Zookeeper 以进行偏移迁移, 因此 "Client" 部分也添加到此示例中. 除非您需要偏移迁移, 否则不需要这样做, 或者您需要此部分用于其他安全组件. 另外, 请确保 Flume 进程的操作系统用户对 jaas 和 keytab 文件具有读权限.
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/path/to/keytabs/flume.keytab"
principal="flume/flumehost1.example.com@YOURKERBEROSREALM";
};
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/path/to/keytabs/flume.keytab"
principal="flume/flumehost1.example.com@YOURKERBEROSREALM";
};
NetCat TCP source
类似于 netcat 的 source, 它侦听给定端口并将每行文本转换为 event . 像 nc -k -l [host] [port]这样的行为. 换句话说, 它打开一个指定的端口并监听数据. 期望是提供的数据是换行符分隔的文本. 每行文本都转换为 Flume event , 并通过连接的 channel 发送. 必需属性以 粗体显示
| Property Name | Default | Description |
|---|---|---|
| channels | – | |
| type | – | The component type name, needs to be netcat |
| bind | – | Host name or IP address to bind to |
| port | – | Port # to bind to |
| max-line-length | 512 | Max line length per event body (in bytes) |
| ack-every-event | true | Respond with an “OK” for every event received |
| selector.type | replicating | replicating or multiplexing |
| selector.* | Depends on the selector.type value | |
| interceptors | – | Space-separated list of interceptors |
| interceptors.* |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = netcat a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 6666 a1.sources.r1.channels = c1 NetCat UDP source
根据原始的 Netcat(TCP)source, 该 source 监听给定端口并将每行文本转换为 event 并通过连接的 channel 发送. 像 nc -u -k -l [host] [port]这样的行为. 必需属性以 粗体显示
| Property Name | Default | Description |
|---|---|---|
| channels | – | |
| type | – | The component type name, needs to be netcatudp |
| bind | – | Host name or IP address to bind to |
| port | – | Port # to bind to |
| remoteAddressHeader | – | |
| selector.type | replicating | replicating or multiplexing |
| selector.* | Depends on the selector.type value | |
| interceptors | – | Space-separated list of interceptors |
| interceptors.* |
agent 名为 a1 的示例:
a1.sources = r1 a1.channels = c1 a1.sources.r1.type = netcatudp a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 6666 a1.sources.r1.channels = c1 Sequence Generator Source
一个简单的 sequence 生成器, 它使用从 0 开始的计数器连续生成 event , 递增 1 并在 totalEvents 处停止. 无法向 channel 发送 event 时重试. 主要用于测试. 在重试期间, 它使重试消息的主体保持与以前相同, 以便在目的地重复数据删除之后, 唯一 event 的数量应等于指定的 totalEvents. 必需属性以粗体显示
| Property Name | Default | Description |
|---|---|---|
| channels | – | |
| type | – | The component type name, needs to be seq |
| selector.type | replicating or multiplexing | |
| selector.* | replicating | Depends on the selector.type value |
| interceptors | – | Space-separated list of interceptors |
| interceptors.* | ||
| batchSize | 1 | Number of events to attempt to process per request loop. |
| totalEvents | Long.MAX_VALUE | Number of unique events sent by the source. |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = seq a1.sources.r1.channels = c1 Syslog Source
读取 syslog 数据并生成 Flume event .UDP source 将整个消息视为单个 event .TCPsource 为每个由换行符 ('n') 分隔的字符串创建一个新 event
必需属性以粗体显示
Syslog TCPsource
原始的, 经过验证的 syslog TCPsource 代码
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 & nbsp;syslogtcp |
| host | - | 要绑定的主机名或 IP 地址 |
| port | - | 要绑定的端口号 |
| eventSize | 2500 | 单个 event 行的最大大小（以字节为单位） |
| keepFields | none | 将此设置为 “all” 将保留 event 正文中的 Priority,Timestamp 和 Hostname。还允许包含间隔开的字段列表。目前，可以包括以下字段：优先级，版本，时间戳，主机名。值'true'和'false'已被弃用，采用'all'和'none'。 |
| clientIPHeader | - | 如果指定，客户端的 IP 地址将使用此处指定的标头名称存储在每个 event 的标头中。这允许拦截器和 channel 选择器基于客户端的 IP 地址定制路由逻辑。不要在此处使用标准 Syslog 标头名称（如_host_），因为在这种情况下将覆盖 event 标头。 |
| clientHostnameHeader | - | 如果指定，则客户端的主机名将使用此处指定的标头名称存储在每个 event 的标头中。这允许拦截器和 channel 选择器基于客户端的主机名自定义路由逻辑。检索主机名可能涉及名称服务反向查找，这可能会影响性能。不要在此处使用标准 Syslog 标头名称（如_host_），因为在这种情况下将覆盖 event 标头。 |
| selector.type | replicating or multiplexing | |
| selector.* | replicating | 取决于 selector.type 值 |
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* | ||
| SSL | false | 将其设置为 true 以启用 SSL 加密。如果启用了 SSL，则还必须通过组件级参数（请参阅下文）或全局 SSL 参数（请参阅 SSL / TLS 支持部分）指定 “密钥库” 和“密钥库密码” 。 |
| keystore | - | 这是 Java 密钥库文件的路径。如果未在此处指定，则将使用全局密钥库（如果已定义，则配置错误）。 |
| keystore-password | - | Java 密钥库的密码。如果未在此处指定，则将使用全局密钥库密码（如果已定义，则配置错误）。 |
| keystore-type | JKS | Java 密钥库的类型。这可以是 “JKS” 或“PKCS12”。如果未在此处指定，则将使用全局密钥库类型（如果已定义，则默认为 JKS）。 |
| exclude-protocols | SSLv3 | 要排除的以空格分隔的 SSL / TLS 协议列表。除指定的协议外，将始终排除 SSLv3。 |
| include-protocols | - | 要包含的以空格分隔的 SSL / TLS 协议列表。启用的协议将是包含的协议，没有排除的协议。如果包含协议为空，则它包括每个支持的协议。 |
| exclude-cipher-suites | - | 要排除的以空格分隔的密码套件列表。 |
| include-cipher-suites | - | 以空格分隔的密码套件列表。启用的密码套件将是包含的密码套件，不包括排除的密码套件。如果 included-cipher-suites 为空，则包含每个支持的密码套件。 |
For example, a syslog TCP source for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = syslogtcp a1.sources.r1.port = 5140 a1.sources.r1.host = localhost a1.sources.r1.channels = c1
多端口 Syslog TCP source
这是 Syslog TCPsource 的更新, 更快, 多端口版本. 请注意, 端口配置设置已替换端口. 多端口功能意味着它可以以有效的方式一次监听多个端口. 此 source 使用 Apache Mina 库来执行此操作. 提供对 RFC-3164 和许多常见 RFC-5424 格式消息的支持. 还提供了配置基于每个端口的字符集的功能.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 multiport_syslogtcp |
| host | - | 要绑定的主机名或 IP 地址。 |
| ports | - | 要绑定到的空格分隔列表（一个或多个）。 |
| eventSize | 2500 | 单个 event 行的最大大小（以字节为单位）。 |
| keepFields | none | 将此设置为 “all” 将保留 event 正文中的 Priority,Timestamp 和 Hostname。还允许包含间隔开的字段列表。目前，可以包括以下字段：优先级，版本，时间戳，主机名。值'true'和'false'已被弃用，采用'all'和'none'。 |
| portHeader | - | 如果指定，端口号将使用此处指定的标头名称存储在每个 event 的标头中。这允许拦截器和 channel 选择器基于传入端口定制路由逻辑。 |
| clientIPHeader | - | 如果指定，客户端的 IP 地址将使用此处指定的标头名称存储在每个 event 的标头中。这允许拦截器和 channel 选择器基于客户端的 IP 地址定制路由逻辑。不要在此处使用标准 Syslog 标头名称（如_host_），因为在这种情况下将覆盖 event 标头。 |
| clientHostnameHeader | - | 如果指定，则客户端的主机名将使用此处指定的标头名称存储在每个 event 的标头中。这允许拦截器和 channel 选择器基于客户端的主机名自定义路由逻辑。检索主机名可能涉及名称服务反向查找，这可能会影响性能。不要在此处使用标准 Syslog 标头名称（如_host_），因为在这种情况下将覆盖 event 标头。 |
| charset.default | UTF-8 | 将 syslogevent 解析为字符串时使用的默认字符集。 |
| charset.port | - | 字符集可基于每个端口进行配置。 |
| BATCHSIZE | 100 | 每个请求循环尝试处理的最大 event 数。使用默认值通常很好。 |
| readBufferSize | 1024 | 内部 Mina 读缓冲区的大小。提供性能调整。使用默认值通常很好。 |
| numProcessors | （auto-detected） | 处理消息时系统上可用的处理器数量。默认是使用 Java Runtime API 自动检测 CPU 数量。Mina 将为每个检测到的 CPU 生成 2 个请求处理线程，这通常是合理的。 |
| selector.type | replicating | replicating, multiplexing, or custom |
| selector.* | - | 取决于 selector.type 值 |
| interceptors | - | 以空格分隔的拦截器列表。 |
| interceptors.* | ||
| SSL | false | 将其设置为 true 以启用 SSL 加密。如果启用了 SSL，则还必须通过组件级参数（请参阅下文）或全局 SSL 参数（请参阅 SSL / TLS 支持部分）指定 “密钥库” 和“密钥库密码” 。 |
| keystore | - | 这是 Java 密钥库文件的路径。如果未在此处指定，则将使用全局密钥库（如果已定义，则配置错误）。 |
| keystore-password | - | Java 密钥库的密码。如果未在此处指定，则将使用全局密钥库密码（如果已定义，则配置错误）。 |
| keystore-type | JKS | Java 密钥库的类型。这可以是 “JKS” 或“PKCS12”。如果未在此处指定，则将使用全局密钥库类型（如果已定义，则默认为 JKS）。 |
| exclude-protocols | SSLv3 | 要排除的以空格分隔的 SSL / TLS 协议列表。除指定的协议外，将始终排除 SSLv3。 |
| include-protocols | - | 要包含的以空格分隔的 SSL / TLS 协议列表。启用的协议将是包含的协议，没有排除的协议。如果包含协议为空，则它包括每个支持的协议。 |
| exclude-cipher-suites | - | 要排除的以空格分隔的密码套件列表。 |
| include-cipher-suites | - | 以空格分隔的密码套件列表。启用的密码套件将是包含的密码套件，不包括排除的密码套件。如果 included-cipher-suites 为空，则包含每个支持的密码套件。 |
For example, a multiport syslog TCP source for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = multiport_syslogtcp a1.sources.r1.channels = c1 a1.sources.r1.host = 0.0.0.0 a1.sources.r1.ports = 10001 10002 10003 a1.sources.r1.portHeader = port Syslog UDPsource
| Property Name | Default | Description |
|---|---|---|
| channels | – | |
| type | – | The component type name, needs to be syslogudp |
| host | – | Host name or IP address to bind to |
| port | – | Port # to bind to |
| keepFields | false | Setting this to true will preserve the Priority, Timestamp and Hostname in the body of the event. |
| clientIPHeader | – | If specified, the IP address of the client will be stored in the header of each event using the header name specified here. This allows for interceptors and channel selectors to customize routing logic based on the IP address of the client. Do not use the standard Syslog header names here (like _host_) because the event header will be overridden in that case. |
| clientHostnameHeader | – | If specified, the host name of the client will be stored in the header of each event using the header name specified here. This allows for interceptors and channel selectors to customize routing logic based on the host name of the client. Retrieving the host name may involve a name service reverse lookup which may affect the performance. Do not use the standard Syslog header names here (like _host_) because the event header will be overridden in that case. |
| selector.type | replicating or multiplexing | |
| selector.* | replicating | Depends on the selector.type value |
| interceptors | – | Space-separated list of interceptors |
| interceptors.* |
For example, a syslog UDP source for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = syslogudp a1.sources.r1.port = 5140 a1.sources.r1.host = localhost a1.sources.r1.channels = c1
HTTP Source < 通过 HTTP POST 和 GET 接受 Flumeevent 的 source.GET 应仅用于实验. HTTP 请求由可插入的 "处理程序" 转换为 flume event , 该处理程序必须实现 HTTPSourceHandler 接口. 该处理程序获取 HttpServletRequest 并返回 flume event 列表.从一个 Http 请求处理的所有 event 都在一个事务中提交给 channel, 从而允许在诸如文件 channel 之类的 channel 上提高效率. 如果处理程序抛出异常, 则此 source 将返回 HTTP 状态 400. 如果 channel 已满, 或者 source 无法将 event 附加到 channel, 则 source 将返回 HTTP 503 - 暂时不可用状态
在一个发布请求中发送的所有 event 都被视为一个批处理, 并在一个事务中插入到 channel 中
此 source 基于 Jetty 9.4, 并提供了设置其他 Jetty 特定参数的功能, 这些参数将直接传递给 Jetty 组件
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | 组件类型名称需要为 http | |
| port | - | source 应绑定的端口。 |
| bind | 0.0.0.0 | 要侦听的主机名或 IP 地址 |
| handler | org.apache.flume.source.http.JSONHandler | 处理程序类的 FQCN。 |
| handler.* | - | 配置处理程序的参数 |
| selector.type | replicating | replicating or multiplexing |
| selector.* | 取决于 selector.type 值 | |
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* | ||
| SSL | 假 | 将属性设置为 true，以启用 SSL.HTTP Source 不支持 SSLv3。 |
| exclude-protocols | 在 SSLv3 | 要排除的以空格分隔的 SSL / TLS 协议列表。除指定的协议外，将始终排除 SSLv3。 |
| include-protocols | - | 要包含的以空格分隔的 SSL / TLS 协议列表。启用的协议将是包含的协议，没有排除的协议。如果包含协议为空，则它包括每个支持的协议。 |
| exclude-cipher-suites | - | 要排除的以空格分隔的密码套件列表。 |
| include-cipher-suites | - | 以空格分隔的密码套件列表。启用的密码套件将是包含的密码套件，不包括排除的密码套件。 |
| keystore | 密钥库的位置，包括密钥库文件名。如果启用了 SSL 但未在此处指定密钥库，则将使用全局密钥库（如果已定义，则配置错误）。 | |
| keystore-password | 密钥库密码。如果启用了 SSL 但未在此处指定密钥库密码，则将使用全局密钥库密码（如果已定义，则配置错误）。 | |
| keystore-type | JKS | 密钥库类型。这可以是 “JKS” 或“PKCS12”。 |
| QueuedThreadPool.* | 要在 org.eclipse.jetty.util.thread.QueuedThreadPool 上设置的 Jetty 特定设置。NB QueuedThreadPool 仅在设置了此类的至少一个属性时使用。 | |
| HttpConfiguration.* | 要在 org.eclipse.jetty.server.HttpConfiguration 上设置 Jetty 特定设置 | |
| SslContextFactory.* | 要在 org.eclipse.jetty.util.ssl.SslContextFactory 上设置的 Jetty 特定设置（仅在 ssl 设置为 true 时适用）。 | |
| ServerConnector.* | 要在 org.eclipse.jetty.server.ServerConnector 上设置的 Jetty 特定设置 |
不推荐使用的属性
| 物业名称 | 默认 | 描述 |
|---|---|---|
| keystorePassword | - | 使用密钥库密码。不推荐的值将被新的值覆盖。 |
| excludeProtocols | SSLv3 | 使用 exclude-protocols。不推荐的值将被新的值覆盖。 |
| enableSSL | false | 使用 ssl。不推荐的值将被新的值覆盖。 |
NB 使用上面列出的对象上的 setter-methods 设置 Jetty 特定设置. 有关完整的详细信息, 请参阅这些类的 Javadoc(, , 和 )
使用特定于 Jetty 的设置时, 上面命名的属性将优先(例如, excludeProtocols 将优先于 SslContextFactory.ExcludeProtocols). 所有房产都是小写的.
An example http source for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = http a1.sources.r1.port = 5140 a1.sources.r1.channels = c1 a1.sources.r1.handler = org.example.REST.RestHandler a1.sources.r1.handler.nickname = random props a1.sources.r1.HttpConfiguration.sendServerVersion = false a1.sources.r1.ServerConnector.idleTimeout = 300 JSONHandler
开箱即用的处理程序可以处理以 JSON 格式表示的 event , 并支持 UTF-8,UTF-16 和 UTF-32 字符集. 处理程序接受一个 event 数组(即使只有一个 event ,event 必须在数组中发送), 并根据请求中指定的编码将它们转换为 Flumeevent . 如果未指定编码, 则假定为 UTF-8.JSON 处理程序支持 UTF-8,UTF-16 和 UTF-32.event 表示如下.
[{
"headers" : {
"timestamp" : "434324343",
"host" : "random_host.example.com"
},
"body" : "random_body"
},
{
"headers" : {
"namenode" : "namenode.example.com",
"datanode" : "random_datanode.example.com"
},
"body" : "really_random_body"
}]
要设置 charset, 请求必须具有指定为 application/JSON 的 内容类型 ; charset = UTF-8(根据需要用 UTF-16 或 UTF-32 替换 UTF-8)
以此处理程序所期望的格式创建 event 的一种方法是使用 Flume SDK 中提供的 JSONEvent 并使用 Google Gson 使用 Gson#fromJson(Object,Type)方法创建 JSON 字符串. 要作为 event 列表的此方法的第二个参数传递的类型标记可以通过以下方式创建:
Type type = new TypeToken<List<JSONEvent>>() {
}.getType();
BlobHandler
默认情况下, HTTPSource 将 JSON 输入拆分为 Flumeevent . 作为替代方案, BlobHandler 是 HTTPSource 的处理程序, 它返回包含请求参数的 event 以及使用此请求上载的二进制大对象(BLOB). 例如 PDF 或 JPG 文件. 请注意, 此方法不适用于非常大的对象, 因为它会将整个 BLOB 缓存在 RAM 中.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| handler | - | 此类的 FQCN:org.apache.flume.sink.solr.morphline.BlobHandler |
| handler.maxBlobLength | 100000000 | 要读取的最大字节数和给定请求的缓冲区 |
Stress source
StressSource 是一个内部负载生成 source 实现, 对压力测试非常有用. 它允许用户使用空标头配置 event 有效负载的大小. 用户可以配置要发送的 event 总数以及要传递的最大成功 event 数.
必需属性以粗体显示
| 物业名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件类型名称，需要是 org.apache.flume.source.StressSource |
| size | 500 | 每个 event 的有效载荷大小。单位：字节 |
| maxTotalEvents | -1 | 要发送的最大 event 数 |
| maxSuccessfulEvents | -1 | 成功发送的最大 event 数 |
| BATCHSIZE | 1 | 一批中要发送的 event 数 |
| maxEventsPerSecond | 0 | 设置为大于零的整数时，在 source 上强制执行速率限制。 |
Example for agent named a1: a1.sources = stresssource-1 a1.channels = memoryChannel-1 a1.sources.stresssource-1.type = org.apache.flume.source.StressSource a1.sources.stresssource-1.size = 10240 a1.sources.stresssource-1.maxTotalEvents = 1000000 a1.sources.stresssource-1.channels = memoryChannel-1 Legacy Sources
legacy source 允许 Flume 1.x agent 从 Flume 0.9.4 agent 接收 event . 它接受 Flume 0.9.4 格式的 event , 将它们转换为 Flume 1.0 格式, 并将它们存储在连接的 channel 中. 0.9.4event 属性 (如 timestamp,pri,host,nanos 等) 将转换为 1.xevent 头属性. 旧版 source 支持 Avro 和 Thrift RPC 连接. 要在两个 Flume 版本之间使用此桥接, 您需要使用 avroLegacy 或 thriftLegacysource 启动 Flume 1.x agent .0.9.4 agent 应该让 agent Sink 指向 1.x agent 的主机 / 端口.
注意
Flume 1.x 的可靠性语义与 Flume 0.9.x 的可靠性语义不同. 旧版 source 不支持 Flume 0.9.x agent 的 E2E 或 DFO 模式. 唯一支持的 0.9.x 模式是尽力而为, 尽管 1.x 流的可靠性设置将适用于传统 source 保存到 Flume 1.xchannel 后的 event
必需属性以粗体显示
Avro Legacy Source
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 org.apache.flume.source.avroLegacy.AvroLegacySource |
| host | - | 要绑定的主机名或 IP 地址 |
| port | - | 要听的端口＃ |
| selector.type | replicating or multiplexing | |
| selector.* | replicating | 取决于 selector.type 值 |
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = org.apache.flume.source.avroLegacy.AvroLegacySource a1.sources.r1.host = 0.0.0.0 a1.sources.r1.bind = 6666 a1.sources.r1.channels = c1 Thrift Legacy Source
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是 org.apache.flume.source.thriftLegacy.ThriftLegacySource |
| host | - | 要绑定的主机名或 IP 地址 |
| port | - | 要听的端口＃ |
| selector.type | replicating or multiplexing | |
| selector.* | replicating | 取决于 selector.type 值 |
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = org.apache.flume.source.thriftLegacy.ThriftLegacySource a1.sources.r1.host = 0.0.0.0 a1.sources.r1.bind = 6666 a1.sources.r1.channels = c1 Custom Source
自定义 source 是您自己的 Source 接口实现. 启动 Flume agent 时, 自定义 source 的类及其依赖项必须包含在 agent 程序的类路径中. 自定义 source 的类型是其 FQCN.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channels | - | |
| type | - | 组件类型名称，需要是您的 FQCN |
| selector.type | replicating or multiplexing | |
| selector.* | replicating | 取决于 selector.type 值 |
| interceptors | - | 以空格分隔的拦截器列表 |
| interceptors.* |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = org.example.MySource a1.sources.r1.channels = c1 Scribe Source
Scribe 是另一种摄取系统. 要采用现有的 Scribe 摄取系统, Flume 应该使用基于 Thrift 的 ScribeSource 和兼容的传输协议. 要部署 Scribe, 请遵循 Facebook 的指南. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件类型名称，需要是 org.apache.flume.source.scribe.ScribeSource |
| port | 1499 | 应该连接 Scribe 的端口 |
| maxReadBufferBytes | 16384000 | Thrift 默认 FrameBuffer 大小 |
| workerThreads | 5 | 在 Thrift 中处理线程数 |
| selector.type | ||
| selector.* |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.type = org.apache.flume.source.scribe.ScribeSource a1.sources.r1.port = 1463 a1.sources.r1.workerThreads = 5 a1.sources.r1.channels = c1 Flume Sinks HDFS Sink
此 sink 将 event 写入 Hadoop 分布式文件系统 (HDFS). 它目前支持创建文本和序列文件. 它支持两种文件 type 的压缩. 可以根据经过的时间或数据大小或 event 数量定期滚动文件(关闭当前文件并创建新文件). 它还根据 event source 自的时间戳或机器等属性对数据进行分区 / 分区. HDFS 目录路径可能包含格式转义序列, 将由 HDFS sink 替换, 以生成用于存储 event 的目录 / 文件名. 使用此 sink 需要安装 hadoop, 以便 Flume 可以使用 Hadoop jar 与 HDFS 集群进行通信. 请注意, 需要支持 sync() 调用的 Hadoop 版本.
以下是支持的转义序列:
| 别号 | 描述 |
|---|---|
| ％{host} | 名为 “host” 的 event 标头的替换值。支持任意标题名称。 |
| ％T | Unix 时间，以毫秒为单位 |
| ％a | locale 的工作日短名称（周一，周二，......） |
| ％A | locale 的完整工作日名称（周一，周二......） |
| ％b | locale 的短月名（Jan,Feb，...） |
| ％B | locale 的长月名（1 月，2 月......） |
| ％C | locale 的日期和时间（2005 年 3 月 3 日 23:05:25） |
| ％d | 每月的一天（01） |
| ％E | 没有填充的月份日（1） |
| ％d | 日期; 与％m /％d /％y 相同 |
| ％H | 小时（00..23） |
| ％I | 小时（01..12） |
| ％j | 一年中的一天（001..366） |
| ％K | 小时（0..23） |
| ％M | 月（01..12） |
| ％N | 没有填充的月份（1..12） |
| ％M | 分钟（00..59） |
| ％p | locale 相当于 am 或 pm |
| ％S | 自 1970-01-01 00:00:00 UTC 以来的秒数 |
| ％S | 第二（00..59） |
| ％Y | 年份的最后两位数（00..99） |
| ％Y | 年（2010 年） |
| ％Z | + hhmm 数字时区（例如，-0400） |
| ％[localhost] | 替换运行 agent 程序的主机的 hostname |
| ％[IP] | 替换运行 agent 程序的主机的 IP 地址 |
| ％[FQDN] | 替换运行 agent 程序的主机的规范 hostname |
注意: 转义字符串％[localhost],％[IP]和％[FQDN]都依赖于 Java 获取 hostname 的能力, 这在某些网络环境中可能会失败.
正在使用的文件将在名称末尾包含 ".tmp". 文件关闭后, 将删除此扩展程序. 这允许排除目录中的部分完整文件. 必需属性以粗体显示
注意
对于所有与时间相关的转义序列, event 标题中必须存在带有 "timestamp" 键的标头(除非 hdfs.useLocalTimeStamp 设置为 true).自动添加此方法的一种方法是使用 TimestampInterceptor.
| 名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 hdfs |
| hdfs.path | - | HDFS 目录路径（ eg hdfs://namenode/flume/webdata/ ） |
| hdfs.filePrefix | FlumeData | 名称前缀为 Flume 在 hdfs 目录中创建的文件 |
| hdfs.fileSuffix | - | 附加到文件的后缀（例如. avro - 注意：不会自动添加句点） |
| hdfs.inUsePrefix | - | 用于水槽主动写入的临时文件的前缀 |
| hdfs.inUseSuffix | .TMP | 用于临时文件的后缀，flume 主动写入 |
| hdfs.emptyInUseSuffix | false | 如果为 false，则在写入输出时使用 hdfs.inUseSuffix。关闭输出后，hdfs.inUseSuffix 将从输出文件名中删除。如果为 true，则忽略 hdfs.inUseSuffix 参数，而是使用空字符串。 |
| hdfs.rollInterval | 30 | 滚动当前文件之前等待的秒数（0 = 根据时间间隔从不滚动） |
| hdfs.rollSize | 1024 | 触发滚动的文件大小，以字节为单位（0：永不基于文件大小滚动） |
| hdfs.rollCount | 10 | 在滚动之前写入文件的 event 数（0 = 从不基于 event 数滚动） |
| hdfs.idleTimeout | 0 | 超时后非活动文件关闭（0 = 禁用自动关闭空闲文件） |
| hdfs.batchSize | 100 | 在将文件刷新到 HDFS 之前写入文件的 event 数 |
| hdfs.codeC | - | 压缩编解码器。以下之一：gzip,bzip2,lzo,lzop,snappy |
| hdfs.fileType | SequenceFile | 文件格式：当前 SequenceFile,DataStream 或 CompressedStream （1)DataStream 不会压缩输出文件，请不要设置 codeC(2)CompressedStream 需要使用可用的 codeC 设置 hdfs.codeC |
| hdfs.maxOpenFiles | 5000 | 仅允许此数量的打开文件。如果超过此数量，则关闭最旧的文件。 |
| hdfs.minBlockReplicas | - | 指定每个 HDFS 块的最小副本数。如果未指定，则它来自类路径中的默认 Hadoop 配置。 |
| hdfs.writeFormat | 可写 | 序列文件记录的格式。一个文本或可写。在使用 Flume 创建数据文件之前设置为 Text，否则 Apache Impala（孵化）或 Apache Hive 无法读取这些文件。 |
| hdfs.threadsPoolSize | 10 | HDFS IO 操作的每个 HDFS sink 的线程数（open, write, etc. ） |
| hdfs.rollTimerPoolSize | 1 | 每个 HDFS sink 用于调度定时文件滚动的线程数 |
| hdfs.kerberosPrincipal | - | 用于访问安全 HDFS 的 Kerberos 用户主体 |
| hdfs.kerberosKeytab | - | 用于访问安全 HDFS 的 Kerberos 密钥表 |
| hdfs.proxyUser | ||
| hdfs.round | false | 是否应将时间戳向下舍入（如果为 true，则影响除％t 之外的所有基于时间的转义序列） |
| hdfs.roundValue | 1 | 舍入到此最高倍（在使用 hdfs.roundUnit 配置的单位中），小于当前时间。 |
| hdfs.roundUnit | second | 舍入值的单位 - second, minute or hour. |
| hdfs.timeZone | Local Time | 应该用于解析目录路径的时区名称，例如 America/Los_Angeles。 |
| hdfs.useLocalTimeStamp | false | 替换转义序列时，请使用本地时间（而不是 event 头中的时间戳）。 |
| hdfs.closeTries | 0 | 启动近距离尝试后，sink 必须尝试重命名文件的次数。如果设置为 1，则此 sink 将不会重新尝试失败的重命名（例如，由于 NameNode 或 DataNode 失败），并且可能使文件处于打开状态，扩展名为. tmp。如果设置为 0,sink 将尝试重命名该文件，直到最终重命名该文件（它将尝试的次数没有限制）。如果关闭调用失败但数据将保持不变，则文件可能仍保持打开状态，在这种情况下，只有在 Flume 重启后文件才会关闭。 |
| hdfs.retryInterval | 180 | 连续尝试关闭文件之间的时间（以秒为单位）。每次关闭调用都会花费多次 RPC 往返 Namenode，因此将此设置得太低会导致名称节点上的大量负载。如果设置为 0 或更小，则如果第一次尝试失败，sink 将不会尝试关闭文件，并且可能使文件保持打开状态或扩展名为 “.tmp”。 |
| serializer | TEXT | 其他可能的选项包括 avro_event 或 EventSerializer.Builder 接口的实现的完全限定类名 & nbsp;。 |
| serializer.* |
不推荐使用的属性:
| 名称 | 默认 | 描述 |
| hdfs.callTimeout | 30000 | HDFS 操作允许的毫秒数，例如 open,write,flush,close。 如果发生许多 HDFS 超时操作，则应增加此数量。 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = hdfs a1.sinks.k1.channel = c1 a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S a1.sinks.k1.hdfs.filePrefix = events- a1.sinks.k1.hdfs.round = true a1.sinks.k1.hdfs.roundValue = 10 a1.sinks.k1.hdfs.roundUnit = minute
以上配置将时间戳向下舍入到最后 10 分钟. 例如, 时间戳为 2012 年 6 月 12 日上午 11:54:34 的 event 将导致 hdfs 路径变为
- /flume/events/2012-06-12/1150/00.
- Hive Sink
此 sink 将包含分隔文本或 JSON 数据的 event 直接流式传输到 Hive 表或分区. event 使用 Hive 事务编写. 一旦将一组 event 提交给 Hive, 它们就会立即显示给 Hive 查询. 水槽将流入的分区既可以预先创建, 也可以选择 Flume 创建它们, 如果它们缺失的话. 传入 event 数据中的字段将映射到 Hive 表中的相应列
| 名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称需要是 hive |
| hive.metastore | - | Hive Metastore URI（ eg thrift://a.b.com:9083 ） |
| hive.database | - | Hive 数据库名称 |
| hive.table | - | Hive 表名 |
| hive.partition | - | 逗号分隔标识要写入的分区的分区值列表。可能包含转义序列。例如：如果表格被分区（大陆：字符串，国家：字符串，时间：字符串），那么'亚洲，印度，2014-02-26-01-21'将表示大陆 = 亚洲，国家 = 印度，时间 = 2014 -02-26-01-21 |
| hive.txnsPerBatchAsk | 100 | Hive 向 Flume 等流媒体客户端授予一批交易而非单笔交易。此设置配置每个事务批处理所需的事务数。来自单个批次中所有事务的数据最终都在一个文件中。Flume 将在批处理中的每个事务中写入最多 batchSizeevent 。此设置与 batchSize 一起提供对每个文件大小的控制。请注意，最终 Hive 会将这些文件透明地压缩为更大的文件。 |
| heartBeatInterval | 240 | （以秒为单位）发送到 Hive 的连续心跳之间的间隔，以防止未使用的事务过期。将此值设置为 0 可禁用心跳。 |
| autoCreatePartitions | true | Flume 将自动创建必要的 Hive 分区以进行流式传输 |
| BATCHSIZE | 15000 | 在单个 Hive 事务中写入 Hive 的最大 event 数 |
| maxOpenConnections | 500 | 仅允许此数量的打开连接。如果超过此数量，则关闭最近最少使用的连接。 |
| callTimeout | 10000 | （以毫秒为单位）Hive 和 HDFS I/O 操作的超时，例如 openTxn,write,commit,abort。 |
| serializer | Serializer 负责解析 event 中的字段并将它们映射到 hive 表中的列。serializer 器的选择取决于 event 中数据的格式。支持的序列化程序：DELIMITED 和 JSON | |
| roundUnit | minute | 舍入值的单位 - 秒，分钟或小时。 |
| roundValue | 1 | 舍入到此最高倍数（在使用 hive.roundUnit 配置的单位中），小于当前时间 |
| 时区 | Local Time | 应该用于解析分区中转义序列的时区名称，例如 America / Los_Angeles。 |
| useLocalTimeStamp | false | 替换转义序列时，请使用本地时间（而不是 event 头中的时间戳）。 |
为 Hive sink 提供了以下序列化程序:
JSON: 处理 UTF8 编码的 JSON(严格语法)event , 不需要配置. JSON 中的对象名称直接映射到 Hive 表中具有相同名称的列. 内部使用 org.apache.hive.hcatalog.data.JsonSerDe, 但独立于 Hive 表的 Serde. 此序列化程序需要安装 HCatalog.
DELIMITED: 处理简单的分隔文本 event . 内部使用 LazySimpleSerde, 但独立于 Hive 表的 Serde.
| 名称 | 默认 | 描述 |
|---|---|---|
| serializer.delimiter | ， | （type：字符串）传入数据中的字段分隔符。要使用特殊字符，请用双引号括起来，例如 “ ” |
| serializer.fieldnames | - | 从输入字段到 hive 表中的列的映射。指定为 hive 表列名称的逗号分隔列表（无空格），按发生顺序标识输入字段。要跳过字段，请保留未指定的列名称。例如。'time ,, ip,message'表示输入映射到 hive 表中的 time,ip 和 message 列的第 1，第 3 和第 4 个字段。 |
| serializer.serdeSeparator | Ctrl-A | （type：字符）自定义基础 serde 使用的分隔符。如果 serializer.fieldnames 中的字段与表列的顺序相同，则 serializer.delimiter 与 serializer.serdeSeparator 相同，并且 serializer.fieldnames 中的字段数小于或等于表的数量，可以提高效率列，因为传入 event 正文中的字段不需要重新排序以匹配表列的顺序。对于' '这样的特殊字符使用单引号。确保输入字段不包含此字符。注意：如果 serializer.delimiter 是单个字符，最好将其设置为相同的字符 |
以下是支持的转义序列:
| 别号 | 描述 |
|---|---|
| ％{host} | 名为 “host” 的 event 标头的替换值。支持任意标题名称。 |
| ％T | Unix 时间，以毫秒为单位 |
| ％a | locale 的工作日短名称（周一，周二，......） |
| ％A | locale 的完整工作日名称（周一，周二......） |
| ％b | locale 的短月名（Jan,Feb，...） |
| ％B | locale 的长月名（1 月，2 月......） |
| ％C | locale 的日期和时间（2005 年 3 月 3 日 23:05:25） |
| ％d | 每月的一天（01） |
| ％d | 日期; 与％m /％d /％y 相同 |
| ％H | 小时（00..23） |
| ％I | 小时（01..12） |
| ％j | 一年中的一天（001..366） |
| ％K | 小时（0..23） |
| ％M | 月（01..12） |
| ％M | 分钟（00..59） |
| ％p | locale 相当于 am 或 pm |
| ％S | 自 1970-01-01 00:00:00 UTC 以来的秒数 |
| ％S | 第二（00..59） |
| ％Y | 年份的最后两位数（00..99） |
| ％Y | 年（2010 年） |
| ％Z | + hhmm 数字时区（例如，-0400） |
注意
对于所有与时间相关的转义序列, event 标题中必须存在具有键 "timestamp" 的标头(除非 useLocalTimeStamp 设置为 true). 自动添加此方法的一种方法是使用 TimestampInterceptor.
示例 Hive 表:
create table weblogs ( id int , msg string )
partitioned by (continent string, country string, time string)
clustered by (id) into 5 buckets
stored as orc;
Example for agent named a1:
a1.channels = c1
a1.channels.c1.type = memory
a1.sinks = k1
a1.sinks.k1.type = hive
a1.sinks.k1.channel = c1
a1.sinks.k1.hive.metastore = thrift://127.0.0.1:9083
a1.sinks.k1.hive.database = logsdb
a1.sinks.k1.hive.table = weblogs
a1.sinks.k1.hive.partition = asia,%{country},%y-%m-%d-%H-%M
a1.sinks.k1.useLocalTimeStamp = false
a1.sinks.k1.round = true
a1.sinks.k1.roundValue = 10
a1.sinks.k1.roundUnit = minute
a1.sinks.k1.serializer = DELIMITED
a1.sinks.k1.serializer.delimiter = " "
a1.sinks.k1.serializer.serdeSeparator = ' '
a1.sinks.k1.serializer.fieldnames =id,,msg
以上配置将时间戳向下舍入到最后 10 分钟. 例如, 将时间戳标头设置为 2012 年 6 月 12 日上午 11:54:34 且 "country" 标头设置为 "india" 的 event 将评估为分区(continent ='asia',country ='india',time ='2012-06-12-11-50'. 序列化程序配置为接受包含三个字段的制表符分隔输入并跳过第二个字段.
Logger Sink
在 INFO 级别记录 event . 通常用于测试 / 调试目的. 必需属性以粗体显示. 此 sink 是唯一的例外, 它不需要在 "记录原始数据" 部分中说明的额外配置.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 & nbsp; logger |
| maxBytesToLog | 16 | 要记录的 event 主体的最大字节数 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = logger a1.sinks.k1.channel = c1 Avro Sink
这个 Sink 形成了 Flume 的分层收集支持的一半. 发送到此 sink 的 Flume event 将转换为 Avro event 并发送到配置的 hostname/port pair .event 将从配置的 channel 中批量获取配置的批处理大小. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 avro。 |
| hostname | - | 要绑定的 hostname 或 IP 地址。 |
| port | - | 要听的端口＃。 |
| batch-size | 100 | 要一起批量发送的 event 数。 |
| connect-timeout | 20000 | 允许第一个（握手）请求的时间量（ms）。 |
| request-timeout | 20000 | 在第一个之后允许请求的时间量（ms）。 |
| reset-connection-interval | none | 重置连接到下一跳之前的时间量。这将迫使 Avro Sink 重新连接到下一跳。这将允许 sink 在添加新闻主机时连接到硬件负载平衡器后面的主机，而无需重新启动 agent 。 |
| compression-type | none | 这可以是 “ none ” 或“ deflate ”。压缩 type 必须与匹配 AvroSource 的压缩 type 匹配 |
| compression-level | 6 | 压缩 event 的压缩级别。0 = 无压缩，1-9 是压缩。数字越大，压缩越多 |
| SSL | false | 设置为 true 以启用此 AvroSink 的 SSL。配置 SSL 时，您可以选择设置“truststore”，“truststore-password”，“truststore-type”，并指定是否“trust-all-certs”。 |
| trust-all-certs | false | 如果将此设置为 true，则不会检查远程服务器（Avrosource）的 SSL 服务器证书。这不应该在生产中使用，因为它使攻击者更容易执行中间人攻击并 “监听” 加密连接。 |
| truststore | - | 自定义 Java 信任库文件的路径。Flume 使用此文件中的证书颁发机构信息来确定是否应该信任远程 Avro Source 的 SSL 身份验证凭据。如果未指定，则将使用全局密钥库。如果未指定全局密钥库，则将使用缺省 Java JSSE 证书颁发机构文件（通常为 Oracle JRE 中的 “jssecacerts” 或“cacerts”）。 |
| truststore-password | - | 信任库的密码。如果未指定，则将使用全局密钥库密码（如果已定义）。 |
| truststore-type | JKS | Java 信任库的 type。这可以是 “JKS” 或其他受支持的 Java 信任库 type。如果未指定，则将使用全局密钥库 type（如果已定义，则 defautl 为 JKS）。 |
| exclude-protocols | 在 SSLv3 | 要排除的以空格分隔的 SSL/TLS 协议列表。除指定的协议外，将始终排除 SSLv3。 |
| maxIoWorkers | 2 * 机器中可用处理器的数量 | I / O 工作线程的最大数量。这是在 NettyAvroRpcClient NioClientSocketChannelFactory 上配置的。 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = avro a1.sinks.k1.channel = c1 a1.sinks.k1.hostname = 10.10.10.10 a1.sinks.k1.port = 4545 Thrift Sink
这个水槽形成了 Flume 的分层收集支持的一半. 发送到此 sink 的 Flume event 将转换为 Thrift event 并发送到配置的 hostname/port pair.event 将从配置的 channel 中批量获取配置的批处理大小.
可以通过启用 kerberos 身份验证将 Thrift sink 配置为以安全模式启动. 要与以安全模式启动的 Thriftsource 通信, Thriftsink 也应该以安全模式运行. client-principal 和 client-keytab 是 Thriftsink 用于向 kerberos KDC 进行身份验证的属性. server-principal 表示 Thriftsource 的主体, 此 sink 配置为以安全模式连接. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要节俭。 |
| hostname | - | 要绑定的 hostname 或 IP 地址。 |
| port | - | 要听的端口＃。 |
| batch-size | 100 | 要一起批量发送的 event 数。 |
| connect-timeout | 20000 | 允许第一个（握手）请求的时间量（ms）。 |
| request-timeout | 20000 | 在第一个之后允许请求的时间量（ms）。 |
| connection-reset-interval | none | 重置连接到下一跳之前的时间量。这将迫使 Thrift Sink 重新连接到下一跳。这将允许 sink 在添加新闻主机时连接到硬件负载平衡器后面的主机，而无需重新启动 agent 。 |
| SSL | false | 设置为 true 以为此 ThriftSink 启用 SSL。配置 SSL 时，您可以选择设置 “truststore”，“truststore-password” 和“truststore-type” |
| truststore | - | 自定义 Java 信任库文件的路径。Flume 使用此文件中的证书颁发机构信息来确定是否应该信任远程 Thrift Source 的 SSL 身份验证凭据。如果未指定，则将使用全局密钥库。如果未指定全局密钥库，则将使用缺省 Java JSSE 证书颁发机构文件（通常为 Oracle JRE 中的 “jssecacerts” 或“cacerts”）。 |
| truststore-password | - | 信任库的密码。如果未指定，则将使用全局密钥库密码（如果已定义）。 |
| truststore-type | JKS | Java 信任库的 type。这可以是 “JKS” 或其他受支持的 Java 信任库 type。如果未指定，则将使用全局密钥库 type（如果已定义，则 defautl 为 JKS）。 |
| exclude-protocols | SSLv3 | 要排除的以空格分隔的 SSL / TLS 协议列表 |
| Kerberos 的 | false | 设置为 true 以启用 kerberos 身份验证。在 kerberos 模式下，需要 client-principal,client-keytab 和 server-principal 才能成功进行身份验证并与启用 kerberos 的 Thrift Source 进行通信。 |
| client-principal | - | Thrift Sink 使用的 kerberos 校长对 kerberos KDC 进行身份验证。 |
| client-keytab | - | Thrift Sink 与客户端主体结合使用的 keytab 位置，用于对 kerberos KDC 进行身份验证。 |
| server-principal | - | Thrift Sink 配置为连接到的 Thrift Source 的 kerberos 主体。 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = thrift a1.sinks.k1.channel = c1 a1.sinks.k1.hostname = 10.10.10.10 a1.sinks.k1.port = 4545 IRC Sink
IRC sink 从附加 channel 接收消息, 并将这些消息中继到已配置的 IRC 目标. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 irc |
| hostname | - | 要连接的 hostname 或 IP 地址 |
| port | 6667 | 要连接的远程主机的端口号 |
| nick | - | 昵称 |
| user | - | 用户名 |
| password | - | 用户密码 |
| chan | - | channel |
| name | ||
| splitlines | - | （布尔值） |
| splitChars 中 | n | 行分隔符（如果你要在配置文件中输入默认值，那么你需要转义反斜杠，如下所示：“ n”） |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = irc a1.sinks.k1.channel = c1 a1.sinks.k1.hostname = irc.yourdomain.com a1.sinks.k1.nick = flume a1.sinks.k1.chan = #flume File Roll Sink
在本地文件系统上存储 event . 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 file_roll。 |
| sink.directory | - | 将存储文件的目录 |
| sink.pathManager | DEFAULT | 要使用的 PathManager 实现。 |
| sink.pathManager.extension | - | 如果使用默认的 PathManager，则为文件扩展名。 |
| sink.pathManager.prefix | - | 如果使用默认 PathManager，则添加到文件名开头的字符串 |
| sink.rollInterval | 30 | 每 30 秒滚动一次文件。指定 0 将禁用滚动并导致所有 event 都写入单个文件。 |
| sink.serializer | TEXT | 其他可能的选项包括 avro_event 或 EventSerializer.Builder 接口实现的 FQCN。 |
| sink.batchSize | 100 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = file_roll a1.sinks.k1.channel = c1 a1.sinks.k1.sink.directory = /var/log/flume Null Sink
丢弃从 channel 收到的所有 event . 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称必须为 null。 |
| BATCHSIZE | 100 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = null a1.sinks.k1.channel = c1 HBase Sink
此 sink 将数据写入 HBase.Hbase 配置是从类路径中遇到的第一个 hbase-site.xml 中获取的. 实现由配置指定的 HbaseEventSerializer 的类用于将 event 转换为 HBase put 和 / 或 increments . 然后将这些 put 和 increments 写入 HBase. 该 sink 提供与 HBase 相同的一致性保证, HBase 是当前行的原子性. 如果 Hbase 无法写入某些 event , 则 sink 将重播该事务中的所有 event
HBase Sink 支持将数据写入安全 HBase. 要写入安全 HBase, agent 程序正在运行的用户必须具有对 sink 配置为写入的表的写入权限. 可以在配置中指定用于对 KDC 进行身份验证的主体和密钥表. Flume agent 程序类路径中的 hbase-site.xml 必须将身份验证设置为 kerberos(有关如何执行此操作的详细信息, 请参阅 HBase 文档)
为方便起见, Flume 提供了两个序列化器.
SimpleHbaseEventSerializer(org.apache.flume.sink.hbase.SimpleHbaseEventSerializer)按原样将 event 主体写入 HBase, 并可选择增加 Hbase 中的列. 这主要是一个示例实现.
RegexHbaseEventSerializer(org.apache.flume.sink.hbase.RegexHbaseEventSerializer)根据给定的正则表达式打破 event 体, 并将每个部分写入不同的列.
type 是 FQCN:org.apache.flume.sink.hbase.HBaseSink.
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 hbase |
| table | - | Hbase 中要写入的表的名称。 |
| ColumnFamily | - | Hbase 中的列族写入。 |
| zookeeperQuorum | - | 法定人数规格。这是 hbase-site.xml 中属性 hbase.zookeeper.quorum 的值 |
| znodeParent | /hbase | -ROOT - 区域的 znode 的基本路径。价值 zookeeper.znode.parent 在 HBase 的 - site.xml 中 |
| BATCHSIZE | 100 | 每个 txn 要写入的 event 数。 |
| coalesceIncrements | false | 如果 sink 将多个增量合并到每个批次的单元格中。如果有限数量的单元格有多个增量，这可能会提供更好的性能。 |
| serializer | org.apache.flume.sink.hbase.SimpleHbaseEventSerializer | 默认增量列 =“iCol”，有效负载列 =“pCol”。 |
| serializer.* | - | 要传递给序列化程序的属性。 |
| kerberosPrincipal | - | 用于访问安全 HBase 的 Kerberos 用户主体 |
| kerberosKeytab | - | 用于访问安全 HBase 的 Kerberos 密钥表 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = hbase a1.sinks.k1.table = foo_table a1.sinks.k1.columnFamily = bar_cf a1.sinks.k1.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializer a1.sinks.k1.channel = c1 HBase2 Sink
HBase2Sink 相当于 HBase 版本 2 的 HBaseSink. 提供的功能和配置参数与 HBaseSink 的情况相同(除了 sinktype 中的 hbase2 标签和包 / 类名称)
type 是 FQCN:org.apache.flume.sink.hbase2.HBase2Sink
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 hbase2 |
| table | - | HBase 中要写入的表的名称。 |
| ColumnFamily | - | HBase 中的列族写入。 |
| zookeeperQuorum | - | 这是 hbase-site.xml 中属性 hbase.zookeeper.quorum 的值 |
| znodeParent | /hbase | -ROOT - 区域的 znode 的基本路径。价值 zookeeper.znode.parent 在 HBase 的 - site.xml 中 |
| BATCHSIZE | 100 | 每个 txn 要写入的 event 数。 |
| coalesceIncrements | false | 如果 sink 将多个增量合并到每个批次的单元格中。如果有限数量的单元格有多个增量，这可能会提供更好的性能。 |
| serializer | org.apache.flume.sink.hbase2.SimpleHBase2EventSerializer | 默认增量列 =“iCol”，有效负载列 =“pCol”。 |
| serializer.* | - | 要传递给序列化程序的属性。 |
| kerberosPrincipal | - | 用于访问安全 HBase 的 Kerberos 用户主体 |
| kerberosKeytab | - | 用于访问安全 HBase 的 Kerberos 密钥表 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = hbase2 a1.sinks.k1.table = foo_table a1.sinks.k1.columnFamily = bar_cf a1.sinks.k1.serializer = org.apache.flume.sink.hbase2.RegexHBase2EventSerializer a1.sinks.k1.channel = c1 AsyncHBaseSink
此 sink 使用异步模型将数据写入 HBase. 实现由配置指定的 AsyncHbaseEventSerializer 的类用于将 event 转换为 HBase put 和 / 或 increments. 然后将这些放置和增量写入 HBase. 此 sink 使用 Asynchbase API https://github.com/OpenTSDB/asynchbase 写入 HBase. 该 sink 提供与 HBase 相同的一致性保证, HBase 是当前行的原子性. 如果 Hbase 无法写入某些 event , 则 sink 将重播该事务中的所有 event .AsyncHBaseSink 只能与 HBase 1.x 一起使用. AsyncHBaseSink 使用的异步客户端库不适用于 HBase 2.type 为 FQCN:org.apache.flume.sink.hbase.AsyncHBaseSink. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 asynchbase |
| table | - | Hbase 中要写入的表的名称。 |
| zookeeperQuorum | - | 法定人数规格。这是 hbase-site.xml 中属性 hbase.zookeeper.quorum 的值 |
| znodeParent | /hbase | -ROOT - 区域的 znode 的基本路径。价值 zookeeper.znode.parent 在 HBase 的 - site.xml 中 |
| ColumnFamily | - | Hbase 中的列族写入。 |
| batchSize | 100 | 每个 txn 要写入的 event 数。 |
| coalesceIncrements | false | 如果 sink 将多个增量合并到每个批次的单元格中。如果有限数量的单元格有多个增量，这可能会提供更好的性能。 |
| timeout | 60000 | sink 为事务中的所有 event 等待来自 hbase 的 acks 的时间长度（以毫秒为单位）。 |
| serializer | org.apache.flume.sink.hbase.SimpleAsyncHbaseEventSerializer | |
| serializer.* | - | 要传递给序列化程序的属性。 |
| async.* | - | 要传递给 asyncHbase 库的属性。这些属性优先于旧的 zookeeperQuorum 和 znodeParent 值。您可以在 AsyncHBase 的文档页面中找到可用属性的列表 & nbsp;。 |
请注意, 此 sink 在配置中获取 Zookeeper Quorum 和父 znode 信息. 可以在 flume 配置文件中指定 Zookeeper Quorum 和父节点配置. 或者, 这些配置值取自类路径中的第一个 hbase-site.xml 文件.
如果配置中未提供这些, 则 sink 将从类路径中的第一个 hbase-site.xml 文件中读取此信息.
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = asynchbase a1.sinks.k1.table = foo_table a1.sinks.k1.columnFamily = bar_cf a1.sinks.k1.serializer = org.apache.flume.sink.hbase.SimpleAsyncHbaseEventSerializer a1.sinks.k1.channel = c1 MorphlineSolrSink
此 sink 从 Flume event 中提取数据, 对其进行转换, 并将其近乎实时地加载到 Apache Solr 服务器中, 后者又向最终用户或搜索应用程序提供查询.
此 sink 非常适合将原始数据流式传输到 HDFS(通过 HdfsSink)并同时提取, 转换并将相同数据加载到 Solr(通过 MorphlineSolrSink)的用例. 特别是, 此 sink 可以处理来自不同数据 source 的任意异构原始数据, 并将其转换为对搜索应用程序有用的数据模型
ETL 功能可使用 morphline 配置文件进行自定义, 该文件定义了一系列转换命令, 用于将 event 记录从一个命令传递到另一个命令
Morphlines 可以看作是 Unix 管道的演变, 其中数据模型被推广为使用通用记录流, 包括任意二进制有效载荷. morphline 命令有点像 Flume 拦截器. Morphlines 可以嵌入到 Flume 等 Hadoop 组件中.
用于解析和转换一组标准数据格式 (如日志文件, Avro,CSV, 文本, HTML,xml,PDF,Word,Excel 等) 的命令是开箱即用的, 还有其他自定义命令和解析器用于其他数据格式可以添加为 morphline 插件. 可以索引任何 type 的数据格式, 并且可以生成任何 type 的 Solr 模式的任何 Solr 文档, 并且可以注册和执行任何自定义 ETL 逻辑.
Morphlines 操纵连续的记录流. 数据模型可以描述如下: 记录是一组命名字段, 其中每个字段具有一个或多个值的有序列表. 值可以是任何 Java 对象. 也就是说, 记录本质上是一个哈希表, 其中每个哈希表条目包含一个 String 键和一个 Java 对象列表作为值.(该实现使用 Guava 的 ArrayListMultimap, 它是一个 ListMultimap). 请注意, 字段可以具有多个值, 并且任何两个记录都不需要使用公共字段名称.
这个 sink 将 Flumeevent 的主体填充到_attachment_body 中记录字段中, 并将 Flumeevent 的标头复制到同名的记录字段中. 然后命令可以对此数据执行操作.
支持路由到 SolrCloud 集群以提高可伸缩性. 索引负载可以分布在大量 MorphlineSolrSink 上, 以提高可伸缩性. 可以跨多个 MorphlineSolrSink 复制索引负载以实现高可用性, 例如使用 Flume 功能, 例如负载平衡 sink 处理器. MorphlineInterceptor 还可以帮助实现到多个 Solr 集合的动态路由(例如, 用于多租户)
必须将环境所需的 morphline 和 Solr jar 放在 Apache Flume 安装的 lib 目录中.
type 是 FQCN:org.apache.flume.sink.Solr.morphline.MorphlineSolrSink
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 org.apache.flume.sink.solr.morphline.MorphlineSolrSink |
| morphlineFile | - | 本地文件系统与 morphline 配置文件的相对或绝对路径。示例：/etc/flume-ng/conf/morphline.conf |
| morphlineId | null | 如果 morphline 配置文件中有多个 morphlines，则用于标识 morphline 的可选名称 |
| batchSize | 1000 | 每个水槽事务要采取的最大 event 数。 |
| batchDurationMillis | 1000 | 每个水槽事务的最大持续时间（ms）。事务在此持续时间之后或超过 batchSize 时提交，以先到者为准。 |
| handlerClass | org.apache.flume.sink.solr.morphline.MorphlineHandlerImpl | 实现 org.apache.flume.sink.solr.morphline.MorphlineHandler 的类的 FQCN |
| isProductionMode | false | 应该为关键任务，大规模在线生产系统启用此标志，这些系统需要在发生不可恢复的异常时无需停机即可取得进展。与未知 Solr 架构字段相关的损坏或格式错误的解析器输入数据，解析器错误和错误会产生不可恢复的异常。 |
| recoverableExceptionClasses | org.apache.solr.client.solrj.SolrServerException | 以逗号分隔的可恢复异常列表，这些异常往往是暂时的，在这种情况下，可以重试相应的任务。示例包括网络连接错误，超时等。当生产模式标志设置为 true 时，使用此参数配置的可恢复异常将不会被忽略，因此将导致重试。 |
| isIgnoringRecoverableExceptions | false | 如果不可恢复的异常被意外错误分类为可恢复，则应启用此标志。这使得 sink 能够取得进展并避免永远重试 event 。 |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = org.apache.flume.sink.Solr.morphline.MorphlineSolrSink a1.sinks.k1.channel = c1 a1.sinks.k1.morphlineFile = /etc/flume-ng/conf/morphline.conf # a1.sinks.k1.morphlineId = morphline1 # a1.sinks.k1.batchSize = 1000 # a1.sinks.k1.batchDurationMillis = 1000 ElasticSearchSink
此 sink 将数据写入弹性搜索集群. 默认情况下, 将写入 event 以便 Kibana http://kibana.org/ 图形界面可以显示它们 - 就像 logstash 一样 https://logstash.net/ 编写它们一样
必须将环境所需的 Elasticsearch 和 lucene-core jar 放在 Apache Flume 安装的 lib 目录中. Elasticsearch 要求客户端 JAR 的主要版本与服务器的主要版本匹配, 并且两者都运行相同的 JVM 次要版本. 如果这不正确, 将出现 SerializationExceptions. 要选择所需的版本, 请首先确定 Elasticsearch 的版本以及目标群集正在运行的 JVM 版本. 然后选择与主要版本匹配的 Elasticsearch 客户端库.0.19.x 客户端可以与 0.19.x 群集通信; 0.20.x 可以与 0.20.x 对话, 0.90.x 可以与 0.90.x 对话. 确定 Elasticsearch 版本后, 读取 pom.xml 文件以确定要使用的正确 lucene-core JAR 版本.
event 将每天写入新索引. 名称将是 < indexName> -yyyy-MM-dd, 其中 < indexName > 是 indexName 参数. sink 将在午夜 UTC 开始写入新索引.
默认情况下, ElasticSearchLogStashEventSerializer 会为弹性搜索序列化 event . 可以使用 serializer 参数覆盖此行为. 此参数接受 org.apache.flume.sink.Elasticsearch.ElasticSearchEventSerializer 或 org.apache.flume.sink.Elasticsearch.ElasticSearchIndexRequestBuilderFactory 的实现. 不推荐使用 ElasticSearchEventSerializer 来支持更强大的 ElasticSearchIndexRequestBuilderFactory
type 是 FQCN:org.apache.flume.sink.Elasticsearch.ElasticSearchSink
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是 org.apache.flume.sink.elasticsearch.ElasticSearchSink |
| hostname | - | 逗号分隔的 hostname:port 列表，如果端口不存在，将使用默认端口'9300' |
| indexName | flume | 将附加日期的索引的名称。示例'flume' - >'flume-yyyy-MM-dd'支持任意标题替换，例如。％{header} 替换为命名 event 标头的值 |
| indexType | logs | 索引文档的 type，默认为'log'支持任意标头替换，例如。％{header} 替换为命名 event 标头的值 |
| clusterName | elasticsearch | 要连接的 ElasticSearch 集群的名称 |
| batchSize | 100 | 每个 txn 要写入的 event 数。 |
| ttl | - | TTL 以天为单位，设置时会导致过期文档自动删除，如果没有设置，文档将永远不会被自动删除。TTL 在早期的整数形式中都被接受，例如 a1.sink.k1.ttl = 5，还有限定符 ms（毫秒），s（秒），m（分钟），h（小时），d（天）和 w（周）。示例 a1.sink.k1.ttl = 5d 将 TTL 设置为 5 天。按照 < a href="http://www.elasticsearch.org/guide/reference/mapping/ttl-field/以获取更多信息。" ztid="2867" ow="472" oh="36">http://www.elasticsearch.org/guide/reference/mapping/ttl-field / 以获取更多信息。 |
| serializer | org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer | 要使用的 ElasticSearchIndexRequestBuilderFactory 或 ElasticSearchEventSerializer。接受任一类的实现，但首选 ElasticSearchIndexRequestBuilderFactory。 |
| serializer.* | - | 要传递给序列化程序的属性。 |
注意
使用标头替换可以方便地使用 event 标头的值来动态决定在存储 event 时要使用的 indexName 和 indexType. 使用此功能时应谨慎, 因为 event 提交者现在可以控制 indexName 和 indexType. 此外, 如果使用 Elasticsearch REST 客户端, 则 event 提交者可以控制所使用的 URL 路径
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = Elasticsearch a1.sinks.k1.hostNames = 127.0.0.1:9200,127.0.0.2:9300 a1.sinks.k1.indexName = foo_index a1.sinks.k1.indexType = bar_type a1.sinks.k1.clusterName = foobar_cluster a1.sinks.k1.batchSize = 500 a1.sinks.k1.ttl = 5d a1.sinks.k1.serializer = org.apache.flume.sink.Elasticsearch.ElasticSearchDynamicSerializer a1.sinks.k1.channel = c1 Kite Dataset Sink
将 event 写入 Kite 数据集的 http://kitesdk.org/docs/current/guide/ 实验 sink. 此 sink 将反序列化每个传入 event 的主体, 并将结果记录存储在风筝数据集中. 它通过按 URI 加载数据集来确定目标数据集.
唯一支持的序列化的 Avro 和记录模式必须在 event 标题使用或者被通过, flume.avro.schema.literal 与 JSON 模式表示或 flume.avro.schema.url 有一个网址, 该模式可能会发现(hdfs:/ ... 支持 URI). 这与使用 deserializer.schemaType = LITERAL 的 Log4jAppender 水槽客户端和 false 脱机目录 source 的 Avro 反序列化器 兼容.
注 1: 不支持 flume.avro.schema.hash 标头. 注意 2: 在某些情况下, 在超过滚动间隔后可能会略微发生文件滚动. 但是, 此延迟不会超过 5 秒. 在大多数情况下, 延迟是可以忽略的.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 必须是 org.apache.flume.sink.kite.DatasetSink |
| kite.dataset.uri | - | 要打开的数据集的 URI |
| kite.repo.uri | - | 要打开的存储库的 URI（不建议使用; 请改用 kite.dataset.uri） |
| kite.dataset.namespace | - | 数据集的命名空间，其中将写入记录（不建议使用; 请改用 kite.dataset.uri） |
| kite.dataset.name | - | 将写入记录的数据集的名称（不建议使用; 请改用 kite.dataset.uri） |
| kite.batchSize | 100 | 每批中要处理的记录数 |
| kite.rollInterval | 30 | 释放数据文件之前的最长等待时间（秒） |
| kite.flushable.commitOnBatch | true | 如果为 true，将提交 Flume 事务，并在每批 kite.batchSize 记录上刷新 writer。此设置仅适用于可刷新数据集。如果为 & nbsp;true，则可以将具有提交数据的临时文件保留在数据集目录中。需要手动恢复这些文件，以使数据对 DatasetReaders 可见。 |
| kite.syncable.syncOnBatch | true | 控制 sink 在提交事务时是否也将同步数据。此设置仅适用于可同步数据集。同步 gaurentees 数据将写入远程系统上的稳定存储，同时只刷新数据已离开 Flume 的客户端缓冲区的 gaurentees。当 kite.flushable.commitOnBatch 属性设置为 false 时，此属性也必须设置为 false。 |
| kite.entityParser | Avro | 将 Flume Events 变为 Kite 实体的 Parser 。有效值是 avro 和 EntityParser.Builder 接口的实现的完全限定类名。 |
| kite.failurePolicy | retry | 政策处理不可恢复的错误，如缺少 & nbsp; 架构在 event 头。默认值 retry 将使当前批处理失败，并再次尝试与旧行为匹配。其他有效值是保存，这将写原始 event 到 kite.error.dataset.uri 数据集，以及一个实现的完全限定类名 FailurePolicy.Builder 接口。 |
| kite.error.dataset.uri | - | 当 kite.failurePolicy 设置为 save 时，保存失败 event 的数据集的 URI 。需要的时候 kite.failurePolicy 设置为保存。 |
| auth.kerberosPrincipal | - | 用于 HDFS 安全身份验证的 Kerberos 用户主体 |
| auth.kerberosKeytab | - | 主体的 Kerberos keytab 位置（本地 FS） |
| auth.proxyUser | - | HDFS 操作的有效用户，如果与 kerberos 主体不同 |
Kafka Sink
这是一个 Flume Sink 实现, 可以将数据发布到 Kafka http://kafka.apache.org/ 主题. 其中一个目标是将 Flume 与 Kafka 集成, 以便基于拉的处理系统可以处理来自各种 Flume source 的数据.
这目前支持 Kafka 服务器版本 0.10.1.0 或更高版本. 测试完成了 2.0.1, 这是发布时最高的可用版本
必需属性以粗体字体标记
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 必须设置为 org.apache.flume.sink.kafka.KafkaSink |
| kafka.bootstrap.servers | - | Kafka-Sink 将连接到的经纪人列表，以获取主题分区列表这可以是经纪人的部分列表，但我们建议至少两个用于 HA。格式为逗号分隔的 hostname:port 列表 |
| kafka.topic | default-flume-topic | Kafka 中将发布消息的主题。如果配置了此参数，则会将消息发布到此主题。如果 event 标头包含 “主题” 字段，则 event 将发布到该主题，从而覆盖此处配置的主题。支持任意头部替换，例如。％{header}将替换为名为 “header” 的 event 标头的值。（如果使用替换，建议将 Kafka agent 的 “auto.create.topics.enable” 属性设置为 true。） |
| flumeBatchSize | 100 | 一批中要处理的消息数。较大批量可提高吞吐量，同时增加延迟。 |
| kafka.producer.acks | 1 | 在考虑成功写入之前，有多少副本必须确认消息。接受的值为 0（从不等待确认），1（仅等待前导），-1（等待所有副本）将此值设置为 - 1，以避免在某些领导失败的情况下丢失数据。 |
| useFlumeEventFormat | false | 默认情况下，event 直接从 event 主体作为字节放到 Kafka 主题上。设置为 true 以将 event 存储为 Flume Avro 二进制格式。与 KafkaSource 上的相同属性或 Kafka Channel 上的 parseAsFlumeEvent 属性一起使用时，这将保留生成端的任何 Flume 头。 |
| defaultPartitionId | - | 指定要发送到此 channel 中的所有 event 的 Kafka 分区 ID（整数），除非被 partitionIdHeader 覆盖。默认情况下，如果未设置此属性，则 event 将由 Kafka Producer 的分区程序分发 - 包括按键（如果指定）（或由 kafka.partitioner.class 指定的分区程序）。 |
| partitionIdHeader | - | 设置后，sink 将使用 event 标头中使用此属性的值命名的字段的值，并将消息发送到主题的指定分区。如果该值表示无效分区，则将抛出 EventDeliveryException。如果存在标头值，则此设置将覆盖 defaultPartitionId。 |
| allowTopicOverride | true | 设置后，sink 将允许将消息生成到 topicHeader 属性指定的主题中（如果提供）。 |
| topicHeader | topic | 当与 allowTopicOverride 一起设置时，将生成一个消息，该消息将使用此属性的值命名为标头的值。与 Kafka Source topicHeader 属性一起使用时应小心，以避免创建环回。 |
| kafka.producer.security.protocol | PLAINTEXT | 如果使用某种级别的安全性写入 Kafka，则设置为 SASL_PLAINTEXT,SASL_SSL 或 SSL。有关安全设置的其他信息，请参见下文。 |
| more producer security props | 如果使用 SASL_PLAINTEXT，则 SASL_SSL 或 SSL 会引用 Kafka 安全性以获取需要在生产者上设置的其他属性。 | |
| 其他 Kafka Producer Properties | - | 这些属性用于配置 Kafka Producer。可以使用 Kafka 支持的任何生产者属性。唯一的要求是使用前缀 kafka.producer 添加属性名称 & nbsp;。例如：kafka.producer.linger.ms |
注意
Kafka Sink 使用 Flume Event 标头中的 topic 和 key 属性将 event 发送到 Kafka. 如果 topic 中存在 headers , 则会将 event 发送到该特定 topic, 从而覆盖为 Sink 配置的主题. 如果 key 在 topic 存在, 关键还是使用 Kafkatopic 分区分区数据. 具有相同密钥的 event 将发送到同一分区. 如果密钥为空, 则将 event 发送到随机分区.
Kafka sink 还为 key.serializer(org.apache.kafka.common.serialization.StringSerializer)和 value.serializer(org.apache.kafka.common.serialization.ByteArraySerializer)
提供默认值. 不建议修改这些参数
不推荐使用的属性:
| 属性名称 | 默认 | 描述 |
|---|---|---|
| brokerList | - | 使用 kafka.bootstrap.servers |
| topic | default-flume-topic | 使用 kafka.topic |
| BATCHSIZE | 100 | 使用 kafka.flumeBatchSize |
| requiredAcks | 1 | 使用 kafka.producer.acks |
下面给出 Kafka sink 的示例配置. 以 kafka 生产者前缀 kafka.producer 开头的属性. 创建 Kafka 生成器时传递的属性不限于此示例中给出的属性. 此外, 可以在此处包含您的自定义属性, 并通过作为方法参数传入的 Flume Context 对象在预处理器中访问它们.
a1.sinks.k1.channel = c1 a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = mytopic a1.sinks.k1.kafka.Bootstrap.servers = localhost:9092 a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks = 1 a1.sinks.k1.kafka.producer.linger.ms = 1 a1.sinks.k1.kafka.producer.compression.type = snappy
安全和 Kafka Sink
Flume 和 Kafka 之间的通信 channel 支持安全认证和数据加密. 对于安全身份验证, 可以使用 Kafka 0.9.0 版中的 SASL / GSSAPI(Kerberos V5)或 SSL(即使该参数名为 SSL, 实际协议是 TLS 实现)
截至目前, 数据加密仅由 SSL / TLS 提供.
将 kafka.producer.security.protocol 设置为以下任何值意味着:
SASL_PLAINTEXT - 无数据加密的 Kerberos 或纯文本身份验证
SASL_SSL - 使用数据加密的 Kerberos 或纯文本身份验证
SSL - 基于 TLS 的加密, 带有可选的身份验证
警告
启用 SSL 时性能会下降, 其大小取决于 CPUtype 和 JVM 实现. 参考: Kafka 安全概述 和跟踪此问题的 jira: https://issues/jira/browse/KAFKA-2561
TLS 和 Kafka Sink:
请阅读配置 Kafka 客户端 SSL 中描述的步骤, 以了解用于微调的其他配置设置, 例如以下任何一项: 安全提供程序, 密码套件, 启用的协议, 信任库或密钥库 type.
配置服务器端身份验证和数据加密的示例.
在此处指定信任库是可选的, 可以使用全局信任库. 有关全局 SSL 设置的更多详细信息, 请参阅 SSL / TLS 支持部分.
注意: 默认情况下, 未定义属性 ssl.endpoint.identification.algorithm, 因此不会执行 hostname 验证. 要启用 hostname 验证, 请设置以下属性
a1.sinks.sink1.kafka.producer.ssl.endpoint.identification.algorithm = HTTPS
启用后, 客户端将根据以下两个字段之一验证服务器的完全限定域名(FQDN):
通用名称(CN) https://tools.ietf.org/html/rfc6125#section-2.3 https://tools.ietf.org/html/rfc6125#section-2.3
主题备选名称(SAN)
如果还需要客户端身份验证, 则还需要将以下内容添加到 Flume agent 配置中, 或者可以使用全局 SSL 设置(请参阅 SSL / TLS 支持部分). 每个 Flume agent 都必须拥有其客户证书, Kafka 经纪人必须单独或通过其签名链来信任. 常见示例是由单个根 CA 签署每个客户端证书, 而后者又由 Kafka 经纪人信任
# optional, the global keystore can be used alternatively a1.sinks.sink1.kafka.producer.ssl.keystore.location = /path/to/client.keystore.jks a1.sinks.sink1.kafka.producer.ssl.keystore.password = <password to access the keystore>
如果密钥库和密钥使用不同的密码保护, 则 ssl.key.password 属性将为生产者密钥库提供所需的额外密钥
a1.sinks.sink1.kafka.producer.ssl.key.password = <password to access the key>
Kerberos 和 Kafka Sink
要将 Kafkasink 与使用 Kerberos 保护的 Kafka 集群一起使用, 请为生产者设置上面提到的 producer.security.protocol 属性. 与 Kafka agent 一起使用的 Kerberos 密钥表和主体在 JAAS 文件的 "KafkaClient" 部分中指定."客户端" 部分描述了 Zookeeper 连接(如果需要). 有关 JAAS 文件内容的信息, 请参阅 Kafka doc. 可以通过 flume-env.sh 中的 JAVA_OPTS 指定此 JAAS 文件的位置以及可选的系统范围的 kerberos 配置:
JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/path/to/krb5.conf" JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/path/to/flume_jaas.conf"
使用 SASL_PLAINTEXT 的示例安全配置:
a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.sink1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.sinks.sink1.kafka.topic = mytopic a1.sinks.sink1.kafka.producer.security.protocol = SASL_PLAINTEXT a1.sinks.sink1.kafka.producer.sasl.mechanism = GSSAPI a1.sinks.sink1.kafka.producer.sasl.kerberos.service.name = kafka
使用 SASL_SSL 的安全配置示例:
a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.sink1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.sinks.sink1.kafka.topic = mytopic a1.sinks.sink1.kafka.producer.security.protocol = SASL_SSL a1.sinks.sink1.kafka.producer.sasl.mechanism = GSSAPI a1.sinks.sink1.kafka.producer.sasl.kerberos.service.name = kafka # optional, the global truststore can be used alternatively a1.sinks.sink1.kafka.producer.ssl.truststore.location = /path/to/truststore.jks a1.sinks.sink1.kafka.producer.ssl.truststore.password = <password to access the truststore>
示例 JAAS 文件. 有关其内容的参考, 请参阅 SASL 配置的 Kafka 文档中所需认证机制 (GSSAPI / PLAIN) 的客户端配置部分. 与 Kafka Source 或 Kafka Channel 不同, 不需要 "Client" 部分, 除非其他连接组件需要它. 另外, 请确保 Flume 进程的操作系统用户对 jaas 和 keytab 文件具有读权限.
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/path/to/keytabs/flume.keytab"
principal="flume/flumehost1.example.com@YOURKERBEROSREALM";
};
HTTP Sink
此 sink 的行为是它将从 channel 获取 event , 并使用 HTTP POST 请求将这些 event 发送到远程服务. event 内容作为 POST 正文发送
此 sink 的错误处理行为取决于目标服务器返回的 HTTP 响应. sink 退避 / 就绪状态是可配置的, 事务提交 / 回滚结果以及 event 是否有助于成功的 event 排放计数也是可配置的
状态代码不可读的服务器返回的任何格式错误的 HTTP 响应都将导致退避信号, 并且不会从该 channel 中消耗该 event
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称需要为 http。 |
| endpoint | - | POST 到的完全限定的 URL 端点 |
| connectTimeout | 5000 | 套接字 connect-timeout（以毫秒为单位） |
| request-timeout | 5000 | 最大请求处理时间（以毫秒为单位） |
| contentTypeHeader | text/plain | HTTP Content-Type 标头 |
| acceptHeader | text/plain | HTTP Accept 标头值 |
| defaultBackoff | true | 是否在接收所有 HTTP 状态代码时默认退避 |
| defaultRollback | true | 是否在接收所有 HTTP 状态代码时默认回滚 |
| defaultIncrementMetrics | false | 是否在接收所有 HTTP 状态代码时默认增加指标 |
| backoff.CODE | - | 为个人（即 200）代码或组（即 2XX）代码配置特定退避 |
| rollback.CODE | - | 为单个（即 200）代码或组（即 2XX）代码配置特定回滚 |
| incrementMetrics.CODE | - | 配置单个（即 200）代码或组（即 2XX）代码的特定度量增量 |
请注意, 最具体的 HTTP 状态代码匹配用于 backoff,rollback 和 incrementMetrics 配置选项. 如果存在 2XX 和 200 状态代码的配置值, 则 200 个 HTTP 代码将使用 200 值, 而 201-299 范围内的所有其他 HTTP 代码将使用 2XX 值.
消耗任何空或空 event 而不向 HTTP 端点发出任何请求.
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = http a1.sinks.k1.channel = c1 a1.sinks.k1.endpoint = http://localhost:8080/someuri a1.sinks.k1.connectTimeout = 2000 a1.sinks.k1.requestTimeout = 2000 a1.sinks.k1.acceptHeader = application/JSON a1.sinks.k1.contentTypeHeader = application/JSON a1.sinks.k1.defaultBackoff = true a1.sinks.k1.defaultRollback = true a1.sinks.k1.defaultIncrementMetrics = false a1.sinks.k1.backoff.4XX = false a1.sinks.k1.rollback.4XX = false a1.sinks.k1.incrementMetrics.4XX = true a1.sinks.k1.backoff.200 = false a1.sinks.k1.rollback.200 = false a1.sinks.k1.incrementMetrics.200 = true Custom Sink
自定义 sink 是您自己的 Sink 接口实现. 启动 Flume agent 程序时, 必须在 agent 程序的类路径中包含自定义 sink 的类及其依赖项.自定义 sink 的 type 是其 FQCN. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| channel | - | |
| type | - | 组件 type 名称，需要是您的 FQCN |
Example for agent named a1: a1.channels = c1 a1.sinks = k1 a1.sinks.k1.type = org.example.MySink a1.sinks.k1.channel = c1 Flume Channels
channel 是 event 在 agent 上暂存的存储库. Source 添加 event ,Sink 删除它.
Memory Channel
event 存储在具有可配置最大大小的内存中队列中. 它非常适合需要更高吞吐量的流量, 并且在 agent 发生故障时准备丢失分阶段数据. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称，需要是 & nbsp;memory |
| capacity | 100 | channel 中存储的最大 event 数 |
| transactionCapacity | 100 | 每个事务 channel 从 source 或提供给 sink 的最大 event 数 |
| keep-alive | 3 | 添加或删除 event 的超时（以秒为单位） |
| byteCapacityBufferPercentage | 20 | 定义 byteCapacity 与 channel 中所有 event 的估计总大小之间的缓冲区百分比，以计算标头中的数据。见下文。 |
| byteCapacity | 见说明 | 允许的最大内存总字节数，作为此 channel 中所有 event 的总和。该实现仅计算 Event 主体，这也是提供 byteCapacityBufferPercentage 配置参数的原因。默认值为计算值，等于 JVM 可用的最大内存的 80％（即命令行传递的 - Xmx 值的 80％）。请注意，如果在单个 JVM 上有多个内存 channel，并且它们碰巧保持相同的物理 event （即，如果您使用来自单个 source 的复制 channel 选择器），那么这些 event 大小可能会被重复计算以用于 channelbyteCapacity 目的。将此值设置为 0 将导致此值回退到大约 200 GB 的内部硬限制。 |
Example for agent named a1: a1.channels = c1 a1.channels.c1.type = memory a1.channels.c1.capacity = 10000 a1.channels.c1.transactionCapacity = 10000 a1.channels.c1.byteCapacityBufferPercentage = 20 a1.channels.c1.byteCapacity = 800000 JDBC Channel
event 存储在由数据库支持的持久存储中. JDBC Channel 当前支持嵌入式 Derby. 这是一个持久的 channel, 非常适合可恢复性很重要的流程. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称，需要是 & nbsp; jdbc |
| db.type | DERBY | 数据库供应商需要是 DERBY。 |
| driver.class | org.apache.derby.jdbc.EmbeddedDriver | 供应商的 JDBC 驱动程序的类 |
| driver.url | （由其他属性建造） | JDBC 连接 URL |
| db.username | “sa” | 数据库连接的用户标识 |
| db.password | - | 数据库连接的密码 |
| connection.properties.file | - | JDBC 连接属性文件路径 |
| create.schema | true | 如果为 true，则创建 db 模式（如果不存在） |
| create.index | true | 创建索引以加快查找速度 |
| create.foreignkey | true | |
| transaction.isolation | “READ_COMMITTED” | db 会话的隔离级别 READ_UNCOMMITTED,READ_COMMITTED,SERIALIZABLE,REPEATABLE_READ |
| maximum.connections | 10 | 允许 db 的最大连接数 |
| 最大容量 | 0（无限制） | channel 中的最大 event 数 |
| sysprop。* | 数据库供应商特定属性 | |
| sysprop.user.home | 存储嵌入式 Derby 数据库的主路径 |
Example for agent named a1: a1.channels = c1 a1.channels.c1.type = jdbc Kafka Channel
event 存储在 Kafka 群集中(必须单独安装).Kafka 提供高可用性和复制, 因此如果 agent 或 kafka agent 崩溃, event 可立即用于其他 sinks
Kafka 频道可用于多种场景:
使用 Flume source 和 sink - 它为 event 提供了可靠且高度可用的 channel
使用 Flume source 和拦截器但没有 sink - 它允许将 Flumeevent 写入 Kafka 主题, 供其他应用程序使用
使用 Flume sink, 但没有 source - 它是一种低延迟, 容错的方式将 event 从 Kafka 发送到 Flumesink, 如 HDFS,HBase 或 Solr
这目前支持 Kafka 服务器版本 0.10.1.0 或更高版本. 测试完成了 2.0.1, 这是发布时最高的可用版本.
配置参数组织如下:
通常与 channel 相关的配置值应用于 channel 配置级别, 例如: a1.channel.k1.type =
与 Kafka 相关的配置值或 Channel 运行的前缀以 "kafka." 为前缀(这对 CommonClient Configs 是有效的), 例如: a1.channel.k1.kafka.topic 和 a1.channel.k1.kafka.Bootstrap.servers. 这与 hdfssink 的运行方式没有什么不同
特定于生产者 / 消费者的属性以 kafka.producer 或 kafka.consumer 为前缀
在可能的情况下, 使用 Kafka 参数名称, 例如: Bootstrap.servers 和 acks
此版本的 flume 与以前的版本向后兼容, 但是下表中显示了已弃用的属性, 并且在配置文件中存在时会在启动时记录警告消息.
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称，需要是 org.apache.flume.channel.kafka.KafkaChannel |
| kafka.bootstrap.servers | - | channel 使用的 Kafka 集群中的经纪商列表这可以是经纪人的部分列表，但我们建议至少两个用于 HA。格式为逗号分隔的 hostname:port 列表 |
| kafka.topic | 水槽沟道 | 频道将使用的 Kafka 主题 |
| kafka.consumer.group.id | 水槽 | channel 用于向 Kafka 注册的消费者群组 ID。多个 channel 必须使用相同的主题和组，以确保当一个 agent 程序发生故障时，另一个 agent 程序可以获取数据请注意，具有相同 ID 的非 channel 使用者可能会导致数据丢失。 |
| parseAsFlumeEvent | 真正 | 期望在频道中使用 FlumeEvent 架构的 Avro 基准。如果 Flume source 写入 channel，则应该为 true; 如果其他生成器正在写入 channel 正在使用的主题，则应为 false。通过使用 flume-ng-sdk 工件提供的 org.apache.flume.source.avro.AvroFlumeEvent，可以在 Flume 之外解析到 Kafka 的 Flume source 消息 |
| 的 pollTimeout | 500 | 在消费者的 “poll（）” 调用中等待的时间量（以毫秒为单位）。https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long） |
| defaultPartitionId | - | 指定要发送到此 channel 中的所有 event 的 Kafka 分区 ID（整数），除非被 partitionIdHeader 覆盖。默认情况下，如果未设置此属性，则 event 将由 Kafka Producer 的分区程序分发 - 包括按键（如果指定）（或由 kafka.partitioner.class 指定的分区程序）。 |
| partitionIdHeader | - | 设置时，生产者将从 event 头中获取使用此属性的值命名的字段的值，并将消息发送到主题的指定分区。如果该值表示无效分区，则该 event 将不被接受。如果存在标头值，则此设置将覆盖 defaultPartitionId。 |
| kafka.consumer.auto.offset.reset | 最新 | 当 Kafka 中没有初始偏移量或服务器上当前偏移量不再存在时（例如因为该数据已被删除）该怎么办：最早：自动将偏移量重置为最早的最新偏移量：自动重置偏移量到最新的偏移量无：如果没有为消费者的组找到任何其他偏移量，则向消费者抛出异常：向消费者抛出异常。 |
| kafka.producer.security.protocol | 纯文本 | 如果使用某种级别的安全性写入 Kafka，则设置为 SASL_PLAINTEXT,SASL_SSL 或 SSL。有关安全设置的其他信息，请参见下文。 |
| kafka.consumer.security.protocol | 纯文本 | 与 kafka.producer.security.protocol 相同，但是用于从 Kafka 阅读 / 消费。 |
| 更多生产者 / 消费者安全道具 | 如果使用 SASL_PLAINTEXT，则 SASL_SSL 或 SSL 会引用 Kafka 安全性以获取需要在生产者 / 消费者上设置的其他属性。 |
不推荐使用的属性:
| 属性名称 | 默认 | 描述 |
|---|---|---|
| brokerList | - | channel 使用的 Kafka 集群中的经纪商列表这可以是经纪人的部分列表，但我们建议至少两个用于 HA。格式为逗号分隔的 hostname:port 列表 |
| 话题 | 水槽沟道 | 使用 kafka.topic |
| 的 groupId | 水槽 | 使用 kafka.consumer.group.id |
| readSmallestOffset | false | 使用 kafka.consumer.auto.offset.reset |
| migrateZookeeperOffsets | 真正 | 如果找不到 Kafka 存储的偏移量，请在 Zookeeper 中查找偏移量并将它们提交给 Kafka。这应该是支持从旧版本的 Flume 无缝 Kafka 客户端迁移。迁移后，可以将其设置为 false，但通常不需要这样做。如果未找到 Zookeeper 偏移量，则 kafka.consumer.auto.offset.reset 配置定义如何处理偏移量。 |
注意
Example for agent named a1: a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel a1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9092,kafka-2:9092,kafka-3:9092 a1.channels.channel1.kafka.topic = channel1 a1.channels.channel1.kafka.consumer.group.id = flume-consumer
安全和 Kafka 频道:
Flume 和 Kafka 之间的通信 channel 支持安全认证和数据加密. 对于安全身份验证, 可以使用 Kafka 0.9.0 版中的 SASL / GSSAPI(Kerberos V5)或 SSL(即使该参数名为 SSL, 实际协议是 TLS 实现)
截至目前, 数据加密仅由 SSL / TLS 提供
将 kafka.producer | consumer.security.protocol 设置为以下任何值意味着:
SASL_PLAINTEXT - 无数据加密的 Kerberos 或纯文本身份验证
SASL_SSL - 使用数据加密的 Kerberos 或纯文本身份验证
SSL - 基于 TLS 的加密, 带有可选的身份验证.
警告
启用 SSL 时性能会下降, 其大小取决于 CPUtype 和 JVM 实现. 参考: Kafka 安全概述 和跟踪此问题的 jira: https://issues/jira/browse/KAFKA-2561
** TLS 和 Kafka 频道:**
请阅读配置 Kafka 客户端 SSL 中描述的步骤 以了解用于微调的其他配置设置, 例如以下任何一项: 安全提供程序, 密码套件, 启用的协议, 信任库或密钥库 type.
配置服务器端身份验证和数据加密的示例.
a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel a1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.channels.channel1.kafka.topic = channel1 a1.channels.channel1.kafka.consumer.group.id = flume-consumer a1.channels.channel1.kafka.producer.security.protocol = SSL # optional, the global truststore can be used alternatively a1.channels.channel1.kafka.producer.ssl.truststore.location = /path/to/truststore.jks a1.channels.channel1.kafka.producer.ssl.truststore.password = <password to access the truststore> a1.channels.channel1.kafka.consumer.security.protocol = SSL # optional, the global truststore can be used alternatively a1.channels.channel1.kafka.consumer.ssl.truststore.location = /path/to/truststore.jks a1.channels.channel1.kafka.consumer.ssl.truststore.password = <password to access the truststore>
在此处指定信任库是可选的, 可以使用全局信任库. 有关全局 SSL 设置的更多详细信息, 请参阅 SSL / TLS 支持部分.
注意: 默认情况下, 未定义属性 ssl.endpoint.identification.algorithm, 因此不会执行 hostname 验证. 要启用 hostname 验证, 请设置以下属性
- a1.channels.channel1.kafka.producer.ssl.endpoint.identification.algorithm = HTTPS
- a1.channels.channel1.kafka.consumer.ssl.endpoint.identification.algorithm = HTTPS
启用后, 客户端将根据以下两个字段之一验证服务器的完全限定域名(FQDN):
通用名称(CN) https://tools.ietf.org/html/rfc6125#section-2.3 https://tools.ietf.org/html/rfc6125#section-2.3
主题备选名称(SAN)
如果还需要客户端身份验证, 则还需要将以下内容添加到 Flume agent 配置中, 或者可以使用全局 SSL 设置(请参阅 SSL / TLS 支持部分). 每个 Flume agent 都必须拥有其客户证书, Kafka 经纪人必须单独或通过其签名链来信任. 常见示例是由单个根 CA 签署每个客户端证书, 而后者又由 Kafka 经纪人信任.
# optional, the global keystore can be used alternatively a1.channels.channel1.kafka.producer.ssl.keystore.location = /path/to/client.keystore.jks a1.channels.channel1.kafka.producer.ssl.keystore.password = <password to access the keystore> # optional, the global keystore can be used alternatively a1.channels.channel1.kafka.consumer.ssl.keystore.location = /path/to/client.keystore.jks a1.channels.channel1.kafka.consumer.ssl.keystore.password = <password to access the keystore>
如果密钥库和密钥使用不同的密码保护, 则 ssl.key.password 属性将为使用者和生产者密钥库提供所需的额外密钥:
a1.channels.channel1.kafka.producer.ssl.key.password = <password to access the key> a1.channels.channel1.kafka.consumer.ssl.key.password = <password to access the key>
** Kerberos 和 Kafka 频道:**
要将 Kafkachannel 与使用 Kerberos 保护的 Kafka 群集一起使用, 请为生产者和 / 或使用者设置上面提到的 producer / consumer.security.protocol 属性. 与 Kafka agent 一起使用的 Kerberos 密钥表和主体在 JAAS 文件的 "KafkaClient" 部分中指定."客户端" 部分描述了 Zookeeper 连接(如果需要). 有关 JAAS 文件内容的信息, 请参阅 Kafka doc. 可以通过 flume-env.sh 中的 JAVA_OPTS 指定此 JAAS 文件的位置以及可选的系统范围的 kerberos 配置:
JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/path/to/krb5.conf" JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/path/to/flume_jaas.conf"
使用 SASL_PLAINTEXT 的示例安全配置:
a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel a1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.channels.channel1.kafka.topic = channel1 a1.channels.channel1.kafka.consumer.group.id = flume-consumer a1.channels.channel1.kafka.producer.security.protocol = SASL_PLAINTEXT a1.channels.channel1.kafka.producer.sasl.mechanism = GSSAPI a1.channels.channel1.kafka.producer.sasl.kerberos.service.name = kafka a1.channels.channel1.kafka.consumer.security.protocol = SASL_PLAINTEXT a1.channels.channel1.kafka.consumer.sasl.mechanism = GSSAPI a1.channels.channel1.kafka.consumer.sasl.kerberos.service.name = kafka
使用 SASL_SSL 的安全配置示例:
a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel a1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093 a1.channels.channel1.kafka.topic = channel1 a1.channels.channel1.kafka.consumer.group.id = flume-consumer a1.channels.channel1.kafka.producer.security.protocol = SASL_SSL a1.channels.channel1.kafka.producer.sasl.mechanism = GSSAPI a1.channels.channel1.kafka.producer.sasl.kerberos.service.name = kafka # optional, the global truststore can be used alternatively a1.channels.channel1.kafka.producer.ssl.truststore.location = /path/to/truststore.jks a1.channels.channel1.kafka.producer.ssl.truststore.password = <password to access the truststore> a1.channels.channel1.kafka.consumer.security.protocol = SASL_SSL a1.channels.channel1.kafka.consumer.sasl.mechanism = GSSAPI a1.channels.channel1.kafka.consumer.sasl.kerberos.service.name = kafka # optional, the global truststore can be used alternatively a1.channels.channel1.kafka.consumer.ssl.truststore.location = /path/to/truststore.jks a1.channels.channel1.kafka.consumer.ssl.truststore.password = <password to access the truststore>
示例 JAAS 文件. 有关其内容的参考, 请参阅 SASL 配置的 Kafka 文档中所需认证机制 (GSSAPI / PLAIN) 的客户端配置部分. 由于 Kafka Source 也可能连接到 Zookeeper 以进行偏移迁移, 因此 "Client" 部分也添加到此示例中. 除非您需要偏移迁移, 否则不需要这样做, 或者您需要此部分用于其他安全组件. 另外, 请确保 Flume 进程的操作系统用户对 jaas 和 keytab 文件具有读权限.
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/path/to/keytabs/flume.keytab"
principal="flume/flumehost1.example.com@YOURKERBEROSREALM";
};
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/path/to/keytabs/flume.keytab"
principal="flume/flumehost1.example.com@YOURKERBEROSREALM";
};
File Channel
必需属性以粗体显示
| 属性名称默认值 | 描述 | |
|---|---|---|
| type | - | 组件 type 名称，需要是 & nbsp; file 。 |
| checkpointDir | ~/.flume/file-channel/checkpoint | 将存储检查点文件的目录 |
| useDualCheckpoints | false | 备份检查点。如果将其设置为 true，则 & nbsp; 必须设置 backupCheckpointDir |
| backupCheckpointDir | - | 备份检查点的目录。此目录不能与数据目录或检查点目录相同 |
| dataDirs | ~/.flume/file-channel/data | 逗号分隔的目录列表，用于存储日志文件。在不同磁盘上使用多个目录可以提高文件 channel 的性能 |
| transactionCapacity | 10000 | channel 支持的最大事务大小 |
| checkpointInterval | 30000 | 检查点之间的时间量（以毫秒为单位） |
| maxFileSize 为 | 2146435071 | 单个日志文件的最大大小（以字节为单位） 2G |
| minimumRequiredSpace | 524288000 | 最小所需可用空间（以字节为单位）。为避免数据损坏，当可用空间低于此值时，文件 channel 将停止接受接收 / 放置请求 500M |
| capacity | 1000000 | channel 的最大容量 & nbsp; 100 万 |
| keep-alive | 3 | 等待放置操作的时间量（以秒为单位） |
| use-log-replay-v1 | false | 推荐：使用原有的重播逻辑 |
| use-fast-replay | false | 推荐：不使用队列重播 |
| checkpointOnClose | true | 控制是否在关闭 channel 时创建检查点。通过避免重放，在关闭时创建检查点可以提高文件 channel 的后续启动速度。 |
| encryption.activeKey | - | 用于加密新数据的密钥名称 |
| encryption.cipherProvider | - | 密码提供程序 type，支持的 type:AESCTRNOPADDING |
| encryption.keyProvider | - | 密钥提供程序 type，支持的 type:JCEKSFILE |
| encryption.keyProvider.keyStoreFile | - | 密钥库文件的路径 |
| encrpytion.keyProvider.keyStorePasswordFile | - | 密钥库密码文件的路径 |
| encryption.keyProvider.keys | - | 所有键的列表（例如 activeKey 设置的历史记录） |
| encyption.keyProvider.keys.*.passwordFile | - | 可选密钥密码文件的路径 |
注意
默认情况下, 文件 channel 使用上面指定的用户主目录内的检查点和数据目录的路径. 因此, 如果 agent 中有多个活动的文件 channel 实例, 则只有一个实例可以锁定目录并导致其他 channel 初始化失败.
因此, 有必要为所有已配置的 channel 提供显式路径, 最好是在不同的磁盘上
此外, 由于文件 channel 将在每次提交后同步到磁盘, 因此将其与将 event 一起批处理的 sink/source 耦合可能是必要的, 以便在多个磁盘不可用于检查点和数据目录时提供良好的性能.
Example for agent named a1: a1.channels = c1 a1.channels.c1.type = file a1.channels.c1.checkpointDir = /mnt/flume/checkpoint a1.channels.c1.dataDirs = /mnt/flume/data
加密
以下是一些示例配置:
使用密钥存储密码分隔生成密码密钥:
keytool -genseckey -alias key-0 -keypass keyPassword -keyalg AES -keysize 128 -validity 9000 -keystore test.keystore -storetype jceks -storepass keyStorePassword
使用与密钥库密码相同的密码生成密钥:
keytool -genseckey -alias key-1 -keyalg AES -keysize 128 -validity 9000 -keystore src/test/resources/test.keystore -storetype jceks -storepass keyStorePassword a1.channels.c1.encryption.activeKey = key-0 a1.channels.c1.encryption.cipherProvider = AESCTRNOPADDING a1.channels.c1.encryption.keyProvider = key-provider-0 a1.channels.c1.encryption.keyProvider = JCEKSFILE a1.channels.c1.encryption.keyProvider.keyStoreFile = /path/to/my.keystore a1.channels.c1.encryption.keyProvider.keyStorePasswordFile = /path/to/my.keystore.password a1.channels.c1.encryption.keyProvider.keys = key-0
false 设您已使用密钥 0 输出密钥, 并且应使用密钥 1 加密新文件:
a1.channels.c1.encryption.activeKey = key-1 a1.channels.c1.encryption.cipherProvider = AESCTRNOPADDING a1.channels.c1.encryption.keyProvider = JCEKSFILE a1.channels.c1.encryption.keyProvider.keyStoreFile = /path/to/my.keystore a1.channels.c1.encryption.keyProvider.keyStorePasswordFile = /path/to/my.keystore.password a1.channels.c1.encryption.keyProvider.keys = key-0 key-1
与上面相同的场景, 但 key-0 有自己的密码:
a1.channels.c1.encryption.activeKey = key-1 a1.channels.c1.encryption.cipherProvider = AESCTRNOPADDING a1.channels.c1.encryption.keyProvider = JCEKSFILE a1.channels.c1.encryption.keyProvider.keyStoreFile = /path/to/my.keystore a1.channels.c1.encryption.keyProvider.keyStorePasswordFile = /path/to/my.keystore.password a1.channels.c1.encryption.keyProvider.keys = key-0 key-1 a1.channels.c1.encryption.keyProvider.keys.key-0.passwordFile = /path/to/key-0.password Custom Channel
自定义 channel 是您自己的 Channel 接口实现. 启动 Flume agent 程序时, 必须在 agent 程序的类路径中包含自定义 channel 的类及其依赖项. 自定义 channel 的 type 是其 FQCN. 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称需要是 FQCN |
Example for agent named a1: a1.channels = c1 a1.channels.c1.type = org.example.MyChannel Flume Channel Selectors
如果未指定 type, 则默认为 "replicating"
Replicating Channel Selector (default)
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| selector.type | replicating | 组件 type 名称需要 & nbsp; replicating |
| selector.optional | - | 要标记为可选的 channel 集 |
Example for agent named a1 and it's source called r1: a1.sources = r1 a1.channels = c1 c2 c3 a1.sources.r1.selector.type = replicating a1.sources.r1.channels = c1 c2 c3 a1.sources.r1.selector.optional = c3
在上面的配置中, c3 是可选的 channel.
无法写入 c3 只是被忽略了. 由于 c1 和 c2 未标记为可选, 因此无法写入这些 channel 将导致事务失败.
Multiplexing Channel Selector (多路复用)
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| selector.type | replicating | 组件 type 名称需要进行多路复用 (multiplexing) |
| selector.header | flume.selector.header | |
| selector.default | - | |
| selector.mapping。* | - |
Example for agent named a1 and it's source called r1: a1.sources = r1 a1.channels = c1 c2 c3 c4 a1.sources.r1.selector.type = multiplexing a1.sources.r1.selector.header = state a1.sources.r1.selector.mapping.CZ = c1 a1.sources.r1.selector.mapping.US = c2 c3 a1.sources.r1.selector.default = c4 Custom Channel Selector
自定义 channel 选择器是您自己的 Channelelector 接口实现. 启动 Flume agent 程序时, 自定义 channel 选择器的类及其依赖项必须包含在 agent 程序的类路径中. 自定义 channel 选择器的 type 是其 FQCN.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| selector.type | - | 组件 type 名称，需要是您的 FQCN |
Example for agent named a1 and its source called r1: a1.sources = r1 a1.channels = c1 a1.sources.r1.selector.type = org.example.MyChannelSelector Flume Sink Processors
接收组允许用户将多个 sink 分组到一个实体中. sink 处理器可用于在组内的所有 sink 上提供负载平衡功能, 或在时间故障的情况下实现从一个 sink 到另一个 sink 的故障转移.
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| sinks | - | 以空格分隔的参与组的 sink 列表 |
| processor.type | default | 组件 type 名称需要是 default,failover 或 load_balance |
Example for agent named a1: a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 a1.sinkgroups.g1.processor.type = load_balance Default Sink Processor
默认 sink 只接受一个 sink. 用户不必为单个 sink 创建处理器(sink 组). 相反, 用户可以遵循本用户指南中上面解释的 source - channel - sink 模式
Failover Sink Processor
故障转移 sink 维护一个优先级的 sink 列表, 保证只要有一个可用的 event 将被处理(传递)
故障转移机制的工作原理是将故障 sink 降级到池中, 在池中为它们分配一个冷却期, 在重试之前随顺序故障而增加. sink 成功发送 event 后, 它将恢复到实时池. sink 具有与之相关的优先级, 数量越大, 优先级越高. 如果在发送 event 时 sink 发生故障, 则接下来将尝试下一个具有最高优先级的 sink 以发送 event . 例如, 在优先级为 80 的 sink 之前激活优先级为 100 的 sink. 如果未指定优先级, 则根据配置中指定 sink 的顺序确定 Sinks 优先级.
要进行配置, 请将 sink 组处理器设置为故障转移并为所有单个 sink 设置优先级. 所有指定的优先级必须是唯一的 此外, 可以使用 maxpenalty 属性设置故障转移时间的上限(以毫秒为单位)
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| sinks | - | 以空格分隔的参与组的 sink 列表 |
| processor.type | default | 组件 type 名称需要进行故障转移 & nbsp; failover |
| processor.priority | - | 优先价值。 必须是与当前 sink 组关联的 sink 实例之一。较高优先级值 Sink 较早被激活。绝对值越大表示优先级越高 |
| processor.maxpenalty | 30000 | 失败的 sink 的最大退避时间（以毫秒为单位） |
Example for agent named a1: a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 a1.sinkgroups.g1.processor.type = failover a1.sinkgroups.g1.processor.priority.k1 = 5 a1.sinkgroups.g1.processor.priority.k2 = 10 a1.sinkgroups.g1.processor.maxpenalty = 10000 Load balancing Sink Processor(负载均衡接收处理器)
负载平衡 sink 处理器提供了在多个 sink 上进行负载均衡流量的功能. 它维护一个索引的活动 sink 列表, 必须在其上分配负载. 实现支持使用 round_robin 或 random 机制分配负载. 选择机制的选择默认为 round_robin 类型, 但可以通过配置覆盖. 通过从 AbstractSinkSelector 继承的自定义类支持自定义选择机制.
调用时, 此选择器使用其配置的选择机制选择下一个 sink 并调用它. 对于 round_robin 和 random 如果所选的 sink 无法传递 event , 则处理器通过其配置的选择机制选择下一个可用的 sink. 此实现不会将失败的 sink 列入黑名单, 而是继续乐观地尝试每个可用的 sink. 如果所有 sink 调用都导致失败, 则选择器将故障传播到 sink 运行器.
如果启用了 backoff , 则 sink 处理器会将失败的 sink 列入黑名单, 将其删除以供给定超时的选择. 当超时结束时, 如果 sink 仍然没有响应, 则超时会以指数方式增加, 以避免在无响应的 sink 上长时间等待时卡住. 在禁用此功能的情况下, 在循环中, 所有失败的 sink 负载将被传递到下一个 sink, 因此不均衡
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| processor.sink | - | 以空格分隔的参与组的 sink 列表 |
| processor.type | default | 组件 type 名称需要是 load_balance |
| processor.backoff | false | 失败的 sink 是否会以指数方式退回。 |
| processor.selector | round_robin | 选择机制。必须是 round_robin,random 或自定义类的 FQCN，它继承自 AbstractSinkelector |
| processor.selector.maxTimeOut | 30000 | 由退避选择器用于限制指数 backoff （以毫秒为单位） |
Example for agent named a1: a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 a1.sinkgroups.g1.processor.type = load_balance a1.sinkgroups.g1.processor.backoff = true a1.sinkgroups.g1.processor.selector = random Custom Sink Processor
目前不支持自定义 sink 处理器
Event Serializers
该 file_roll sink 和 HDFS sink 都支持 EventSerializer 接口. 下面提供了随 Flume 一起提供的 Event Serializer 的详细信息.
Body Text Serializer
别名: text . 此拦截器将 event 的主体写入输出流, 而不进行任何转换或修改. event 标题将被忽略. 配置选项如下:
| 属性名称 | 默认 | 描述 |
|---|---|---|
| appendNewline | 真正 | 是否在写入时将换行符附加到每个 event 。由于遗留原因，默认值为 truefalse 定 event 不包含换行符。 |
Example for agent named a1: a1.sinks = k1 a1.sinks.k1.type = file_roll a1.sinks.k1.channel = c1 a1.sinks.k1.sink.directory = /var/log/flume a1.sinks.k1.sink.serializer = text a1.sinks.k1.sink.serializer.appendNewline = false "Flume Event" Avro Event Serializer
Alias: avro_event.
此拦截器将 Flumeevent 序列化为 Avro 容器文件. 使用的模式与 Avro RPC 机制中用于 Flumeevent 的模式相同.
此序列化程序继承自 AbstractAvroEventSerializer 类
配置选项如下:
| 属性名称 | 默认 | 描述 |
|---|---|---|
| syncIntervalBytes | 2048000 | Avro 同步间隔，大约为字节。 2M ??? |
| compressionCodec | null | Avro 压缩编解码器。有关受支持的编解码器，请参阅 Avro 的 CodecFactory 文档。 |
Example for agent named a1: a1.sinks.k1.type = hdfs a1.sinks.k1.channel = c1 a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S a1.sinks.k1.serializer = avro_event a1.sinks.k1.serializer.compressionCodec = snappy
Avro event 序列化器
别名: 此序列化程序没有别名, 必须使用完全限定的类名类名指定.
这将 Flume event 序列化为 Avro 容器文件, 如 "Flume Event" Avro Event Serializer, 但记录模式是可配置的. 记录模式可以指定为 Flume 配置属性, 也可以在 event 头中传递.
要将记录模式作为 Flume 配置的一部分传递, 请使用下面列出的属性 schemaURL.
要在 event 标头中传递记录模式, 请指定 包含模式的 JSON 格式表示的 event 标头 flume.avro.schema.literal 或包含可以找到模式的 URL 的 flume.avro.schema.url( hdfs:/... URIs are supported I)
此序列化程序继承自 AbstractAvroEventSerializer 类.
配置选项如下:
| 属性名称 | 默认 | 描述 |
|---|---|---|
| syncIntervalBytes | 2048000 | Avro 同步间隔，大约为字节。 |
| compressionCodec | null | Avro 压缩编解码器。有关受支持的编解码器，请参阅 Avro 的 CodecFactory 文档。 |
| schemaURL | Avro 架构 URL。标题中指定的模式 ovverride 此选项。 |
Example for agent named a1: a1.sinks.k1.type = hdfs a1.sinks.k1.channel = c1 a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S a1.sinks.k1.serializer = org.apache.flume.sink.hdfs.AvroEventSerializer$Builder a1.sinks.k1.serializer.compressionCodec = snappy a1.sinks.k1.serializer.schemaURL = hdfs://namenode/path/to/schema.avsc Flume Interceptors
Flume 具有在运行中修改 / 删除 event 的能力. 这是在拦截器的帮助下完成的. 拦截器是实现 org.apache.flume.interceptor.Interceptor 的类接口. 拦截器可以根据拦截器开发者选择的任何标准修改甚至删除 event .Flume 支持拦截器的链接. 通过在配置中指定拦截器构建器类名列表, 可以实现此目的. 拦截器在 source 配置中指定为以空格分隔的列表. 指定拦截器的顺序是它们被调用的顺序. 一个拦截器返回的 event 列表将传递给链中的下一个拦截器. 拦截器可以修改或删除 event . 如果拦截器需要删除 event , 它就不会在它返回的列表中返回该 event . 如果要删除所有 event , 则只返回一个空列表. 拦截器是命名组件, 下面是它们如何通过配置创建的示例:
a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.interceptors = i1 i2
a1.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.HostInterceptor$Builder
a1.sources.r1.interceptors.i1.preserveExisting = false
a1.sources.r1.interceptors.i1.hostHeader = hostname
a1.sources.r1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
a1.sinks.k1.filePrefix = FlumeData.%{
CollectorHost
}.%Y-%m-%d
a1.sinks.k1.channel = c1
请注意, 拦截器构建器将传递给 type config 参数. 拦截器本身是可配置的, 可以传递配置值, 就像传递给任何其他可配置组件一样. 在上面的示例中, event 首先传递给 HostInterceptor, 然后 HostInterceptor 返回的 event 传递给 TimestampInterceptor. 您可以指定完全限定的类名 (FQCN) 或别名时间戳. 如果您有多个收集器写入相同的 HDFS 路径, 那么您也可以使用 HostInterceptor
时间戳拦截器 (Timestamp Interceptor)
此拦截器将 event 标头插入到 event 标头中, 以毫秒为单位处理 event . 此拦截器插入带有键时间戳 (或由 header 属性指定) 的标头, 其值为相关时间戳. 如果已在配置中存在, 则此拦截器可以保留现有时间戳
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是时间戳或 FQCN |
| headerName | timestamp | 用于放置生成的时间戳的标头的名称。 |
| preserveExisting | false | 如果时间戳已存在，是否应保留 - true 或 false |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.channels = c1 a1.sources.r1.type = seq a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = timestamp
主机拦截器(Host Interceptor)
此拦截器插入运行此 agent 程序的主机的 hostname 或 IP 地址. 它根据配置插入带有密钥主机或已配置密钥的标头, 其值为主机的 hostname 或 IP 地址.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 & nbsp;host |
| preserveExisting | false | 如果主机头已存在，是否应保留 - true 或 false |
| useIP | true | 如果为 true，请使用 IP 地址，否则使用 hostname。 |
| hostHeader | host | 要使用的标头密钥。 |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = host
静态拦截器 (Static Interceptor)
静态拦截器允许用户将具有静态值的静态头附加到所有 event
当前实现不允许一次指定多个标头. 相反, 用户可能会链接多个静态拦截器, 每个静态拦截器定义一个
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 & nbsp;static |
| preserveExisting | true | 如果已配置的标头已存在，则应保留它 - true 或 false |
| key | key | 应创建的标头的名称 |
| value | value | 应该创建的静态值 |
Example for agent named a1: a1.sources = r1 a1.channels = c1 a1.sources.r1.channels = c1 a1.sources.r1.type = seq a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = static a1.sources.r1.interceptors.i1.key = datacenter a1.sources.r1.interceptors.i1.value = NEW_YORK Remove Header Interceptor (删除标题拦截器)
此拦截器通过删除一个或多个 headers 来操纵 Flume event headers. 它可以删除静态定义的 headers, 基于正则表达式的 headers 或列表中的 headers. 如果这些都没有定义, 或者没有符合条件的 headers, 则不会修改 Flume 事件.
请注意, 如果只需要删除一个 headers, 则按名称指指定 header 可提供相对于其他两种方法的性能优势.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 remove_header |
| withName | - | 要删除的 header 的名称 |
| fromList | - | 要删除的 header 列表，使用 fromListSeparator 指定的分隔符分隔 |
| fromListSeparator | S * S * | 正则表达式，用于分隔 fromList 指定的列表中的多个 header 名称。默认值是由任意数量的空白字符包围的逗号 |
| matching | - | 将删除名称与此正则表达式匹配的所有 header |
UUID 拦截器 (UUID Interceptor)
此拦截器在所有截获的 event 上设置通用唯一标识符. 示例 UUID 是 b5755073-77a9-43c1-8fad-b7a586fc1b97, 表示 128 位值.
如果 event 的应用程序级别唯一键不可用, 请考虑使用 UUIDInterceptor 自动为 event 分配 UUID. 一旦 UUID 进入 Flume 网络就将其分配给 event 非常重要; 也就是说, 在流量的第一个 Flume Source 中. 这使得在 Flume 网络中面对复制和重新发送时 event 的后续重复数据删除可以实现高可用性和高性能. 如果应用程序级别密钥可用, 则优于自动生成的 UUID, 因为它使用所述公知的应用程序级别密钥在数据存储中启用后续更新和 event 删除.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 org.apache.flume.sink.solr.morphline.UUIDInterceptor $ Builder |
| headerName | id | 要修改的 Flume header 的名称 |
| preserveExisting | true | 如果 UUID 标头已存在，是否应保留 - true 或 false |
| prefix | “” | 前缀字符串常量，用于预先添加到每个生成的 UUID |
Morphline Interceptor (形态拦截器)
此拦截器通过 morphline 配置文件过滤 event , 该文件定义了一系列转换命令, 用于将记录从一个命令传递到另一个命令. 例如, morphline 可以忽略某些 event 或通过基于正则表达式的模式匹配来更改或插入某些 event 头, 或者它可以通过 Apache Tika 在截获的 event 上自动检测和设置 MIMEtype. 例如, 这种数据包嗅探可用于 Flume 拓扑中基于内容的动态路由. MorphlineInterceptor 还可以帮助实现到多个 Apache Solr 集合的动态路由(例如, 用于多租户)
目前, 存在一个限制, 即拦截器的形态线不能为每个输入 event 生成多个输出记录. 此拦截器不适用于重型 ETL 处理 - 如果需要, 请考虑将 ETL 处理从 Flume Source 移至 Flume Sink, 例如移至 MorphlineSolrSink.
必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 org.apache.flume.sink.solr.morphline.MorphlineInterceptor $ Builder |
| morphlineFile | - | 本地文件系统与 morphline 配置文件的相对或绝对路径。示例：/etc/flume-ng/conf/morphline.conf |
| morphlineId | null | 如果 morphline 配置文件中有多个 morphlines，则用于标识 morphline 的可选名称 |
Sample flume.conf file: a1.sources.avroSrc.interceptors = morphlineinterceptor a1.sources.avroSrc.interceptors.morphlineinterceptor.type = org.apache.flume.sink.Solr.morphline.MorphlineInterceptor$Builder a1.sources.avroSrc.interceptors.morphlineinterceptor.morphlineFile = /etc/flume-ng/conf/morphline.conf a1.sources.avroSrc.interceptors.morphlineinterceptor.morphlineId = morphline1
搜索和替换拦截器 (Search and Replace Interceptor)
此拦截器基于 Java 正则表达式提供简单的基于字符串的搜索和替换功能. 还可以进行回溯 / 群组捕捉. 此拦截器使用与 Java Matcher.replaceAll()方法中相同的规则
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 search_replace |
| searchPattern | - | 要搜索和替换的模式。 |
| replaceString | - | 替换字符串。 |
| charset | UTF-8 | event 正文的字符集。默认 false 设为 UTF-8。 |
Example configuration: a1.sources.avroSrc.interceptors = search-replace a1.sources.avroSrc.interceptors.search-replace.type = search_replace # Remove leading alphanumeric characters in an event body. a1.sources.avroSrc.interceptors.search-replace.searchPattern = ^[A-Za-z0-9_]+ a1.sources.avroSrc.interceptors.search-replace.replaceString = Another example: a1.sources.avroSrc.interceptors = search-replace a1.sources.avroSrc.interceptors.search-replace.type = search_replace # Use grouping operators to reorder and munge words on a line. a1.sources.avroSrc.interceptors.search-replace.searchPattern = The quick brown ([a-z]+) jumped over the lazy ([a-z]+) a1.sources.avroSrc.interceptors.search-replace.replaceString = The hungry $2 ate the careless $1
正则表达式过滤拦截器 (Regex Filtering Interceptor)
此拦截器通过将 event 主体解释为文本并将文本与配置的正则表达式进行匹配来有选择地过滤 event . 提供的正则表达式可用于包括 event 或排除 event
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 regex_filter |
| regex | ".*" | 用于匹配 event 的正则表达式 |
| excludeEvents | false | 如果为 true，则 regex 确定要排除的 event ，否则 regex 确定要包括的 event 。 |
正则表达式提取器拦截器(Regex Extractor Interceptor)
此拦截器使用指定的正则表达式提取正则表达式匹配组, 并将匹配组作为标题附加到 event 上. 它还支持可插入序列化程序, 用于在将匹配组添加为 event 标头之前对其进行格式化.
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 regex_extractor |
| regex | - | 用于匹配 event 的正则表达式 |
| serializer | - | 以空格分隔的序列化程序列表，用于映射与标题名称匹配并序列化其值。（参见下面的示例）Flume 为以下序列化程序提供内置支持： org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer |
| serializer 器。.TYPE | default | 必须是 default （org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer），org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer，或实现 org.apache.flume.interceptor.RegexExtractorInterceptorSerializer 的自定义类的 FQCN |
| serializers..name | - | |
| serializers.* | - | Serializer-specific properties |
序列化器用于将匹配映射到标题名称和格式化标题值; 默认情况下, 您只需指定标题名称, 将使用默认的 org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer. 此序列化程序只是将匹配映射到指定的标题名称, 并在正则表达式提取时传递值. 您可以使用完全限定的类名 (FQCN) 将自定义序列化程序实现插入到提取器中, 以便以您喜欢的方式格式化匹配.
例 1:
如果 Flumeevent 主体包含 1:2:3.4foobar5 , 则使用以下配置
a1.sources.r1.interceptors.i1.regex = (\d):(\d):(\d) a1.sources.r1.interceptors.i1.serializers = s1 s2 s3 a1.sources.r1.interceptors.i1.serializers.s1.name = one a1.sources.r1.interceptors.i1.serializers.s2.name = two a1.sources.r1.interceptors.i1.serializers.s3.name = three
提取的 event 将包含相同的主体, 但后面的标题将添加一个 => 1, 两个 => 2, 三个 => 3
例 2:
如果 Flumeevent 正文包含 2012-10-18 18:47:57,614 某些 日志 行, 则使用以下配置
a1.sources.r1.interceptors.i1.regex = ^(?:\n)?(\d\d\d\d-\d\d-\d\d\s\d\d:\d\d) a1.sources.r1.interceptors.i1.serializers = s1 a1.sources.r1.interceptors.i1.serializers.s1.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer a1.sources.r1.interceptors.i1.serializers.s1.name = timestamp a1.sources.r1.interceptors.i1.serializers.s1.pattern = yyyy-MM-dd HH:mm
提取的 event 将包含相同的主体, 但后面的标题将添加 timestamp => 1350611220000
Flume 属性
| 属性名称 | 默认 | 描述 |
|---|---|---|
| flume.called.from.service | - | 如果指定了此属性，则 Flume agent 将继续轮询配置文件，即使在预期位置找不到配置文件也是如此。否则，如果配置在预期位置不存在，Flume agent 将终止。设置此属性时不需要属性值（例如，只需指定 - Dflume.called.from.service 即可） |
属性: flume.called.from.service
Flume 每隔 30 秒定期轮询一次指定配置文件的更改. 如果首次轮询现有文件, 或者自上次轮询以来现有文件的修改日期发生更改, Flume agent 将从配置文件加载新配置. 重命名或移动文件不会更改其修改时间. 当 Flume agent 轮询一个不存在的文件时, 会发生以下两种情况之一:
当 agent 首次轮询不存在的配置文件时, agent 将根据 flume.called.from.service 属性执行操作. 如果设置了属性, 则 agent 将继续轮询(始终在同一时间 - 每 30 秒). 如果未设置该属性, 则 agent 会立即终止
当 agent 轮询一个不存在的配置文件并且这不是第一次轮询文件时, agent 不会对此轮询周期进行配置更改. agent 继续轮询而不是终止.
配置过滤器
Flume 提供了一种工具, 用于以配置过滤器的形式将敏感或生成的数据注入配置. 配置密钥可以设置为配置属性的值, 它将由配置过滤器替换为它所代表的值
配置过滤器的常见用法
格式类似于 Java 表达式语言, 但它目前不是一个完全有效的 EL 表达式解析器, 只是一种看起来像它的格式
<agent_name>
.configfilters =
<filter_name>
<agent_name>
.configfilters.
<filter_name>
.type =
<filter_type>
<agent_name>
.sources.
<source_name>
.parameter = ${
<filter_name>
['
<key_for_sensitive_or_generated_data>
']}
<agent_name>
.sinks.
<sink_name>
.parameter = ${
<filter_name>
['
<key_for_sensitive_or_generated_data>
']}
<agent_name>
.
<component_type>
.
<component_name>
.parameter = ${
<filter_name>
['
<key_for_sensitive_or_generated_data>
']} #or
<agent_name>
.
<component_type>
.
<component_name>
.parameter = ${
<filter_name>
["
<key_for_sensitive_or_generated_data>
"]} #or
<agent_name>
.
<component_type>
.
<component_name>
.parameter = ${
<filter_name>
[
<key_for_sensitive_or_generated_data>
]} #or
<agent_name>
.
<component_type>
.
<component_name>
.parameter = some_constant_data${
<filter_name>
[
<key_for_sensitive_or_generated_data>
]}
环境变量配置过滤器
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 env |
示例, 要在配置中隐藏密码, 请设置其值, 如以下示例所示
a1.sources = r1
a1.channels = c1
a1.configfilters = f1
a1.configfilters.f1.type = env
a1.sources.r1.channels = c1
a1.sources.r1.type = http
a1.sources.r1.keystorePassword = ${
f1['my_keystore_password']
}
这里 a1.source.r1.keystorePassword 配置属性将获取 my_keystore_password 环境变量的值. 设置环境变量的一种方法是运行如下所示的 flume agent:
$ my_keystore_password = Secret123 bin / flume -ng agent --conf conf --conf-file example.conf ...
外部处理配置过滤器(External Process Config Filter)
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是外部的 |
| command | - | 将执行以获取给定键的值的命令。该命令将被调用为：
并且预期返回单行值，退出代码为 0。 |
| charset | UTF-8 | 返回字符串的字符集。 |
示例, 要在配置中隐藏密码, 请设置其值, 如以下示例所示:
a1.sources = r1
a1.channels = c1
a1.configfilters = f1
a1.configfilters.f1.type = external
a1.configfilters.f1.command = /usr/bin/passwordResolver.sh
a1.configfilters.f1.charset = UTF-8
a1.sources.r1.channels = c1
a1.sources.r1.type = http
a1.sources.r1.keystorePassword = ${
f1['my_keystore_password']
}
在此示例中, flume 将运行以下命令以获取值
$ /usr/bin/passwordResolver.sh my_keystore_password
该 passwordResolver.sh 将返回 Secret123 与退出代码 0.
示例, 要生成滚动文件 sink 的目录的一部分, 请设置其值, 如以下示例所示:
a1.sources = r1
a1.channels = c1
a1.configfilters = f1
a1.configfilters.f1.type = external
a1.configfilters.f1.command = /usr/bin/generateUniqId.sh
a1.configfilters.f1.charset = UTF-8
a1.sinks = k1
a1.sinks.k1.type = file_roll
a1.sinks.k1.channel = c1
a1.sinks.k1.sink.directory = /var/log/flume/agent_${
f1['agent_name']
}
在此示例中, flume 将运行以下命令以获取值
$ /usr/bin/generateUniqId.sh agent_name
该 generateUniqId.sh 将返回 1234 与退出代码 0.
Hadoop 凭据存储配置过滤器
此功能的类路径上需要一个 hadoop-common 库(2.6 + 版本). 如果安装了 hadoop, 则 agent 会自动将其添加到类路径中
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 hadoop |
| credential.provider.path | - | 提供者路径。请参阅 hadoop 文档_here:https：//hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/CredentialProviderAPI.html#Configuring_the_Provider_Path |
| credstore.java-keystore-provider.password-file | - | 如果文件用于存储密码，则为密码文件的名称。该文件必须在类路径上。可以使用 HADOOP_CREDSTORE_PASSWORD 环境变量设置提供程序密码，也可以将其设置为空。 |
示例, 要在配置中隐藏密码, 请设置其值, 如以下示例所示:
a1.sources = r1
a1.channels = c1
a1.configfilters = f1
a1.configfilters.f1.type = hadoop
a1.configfilters.f1.credential.provider.path = jceks://file/<path_to_jceks file>
a1.sources.r1.channels = c1
a1.sources.r1.type = http
a1.sources.r1.keystorePassword = ${
f1['my_keystore_password']
}
Log4j 追加 (Log4J Appender)
将 Log4jevent 附加到 flume agent 的 avrosource. 使用此 appender 的客户端必须在类路径中包含 flume-ng-sdk(例如, flume-ng-sdk-1.9.0.jar). 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| hostname | - | 使用 avrosource 运行远程 Flume agent 的 hostname。 |
| port | - | 远程 Flume agent 的 avrosource 正在侦听的端口。 |
| UnsafeMode | false | 如果为 true，则 appender 将不会在发送 event 失败时抛出异常。 |
| AvroReflectionEnabled | false | 使用 Avro Reflection 来序列化 Log4jevent 。（当用户记录字符串时不要使用） |
| AvroSchemaUrl | - | 可从中检索 Avro 架构的 URL。 |
示例 log4j.properties 文件:
#... log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender log4j.appender.flume.Hostname = example.com log4j.appender.flume.Port = 41414 log4j.appender.flume.UnsafeMode = true # configure a class's logger to output to the flume appender log4j.logger.org.example.MyClass = DEBUG,flume #...
默认情况下, 通过调用 toString()或使用 Log4j 布局 (如果指定) 将每个 event 转换为字符串.
如果 event 是 org.apache.avro.generic.GenericRecord,org.apache.avro.specific.SpecificRecord 的实例 , 或者属性 AvroReflectionEnabled 设置为 true, 则将使用 Avro 序列化序列化 event .
使用 Avro 架构序列化每个 event 效率很低, 因此最好提供一个架构 URL, 下游 sink(通常是 HDFSsink)可以从该架构 URL 检索架构.如果未指定 AvroSchemaUrl, 则架构将作为 Flume 标头包含在内.
示例 log4j.properties 文件配置为使用 Avro 序列化:
#... log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender log4j.appender.flume.Hostname = example.com log4j.appender.flume.Port = 41414 log4j.appender.flume.AvroReflectionEnabled = true log4j.appender.flume.AvroSchemaUrl = hdfs://namenode/path/to/schema.avsc # configure a class's logger to output to the flume appender log4j.logger.org.example.MyClass = DEBUG,flume #...
负载均衡 Log4j 追加 (Load Balancing Log4J Appender)
将 Log4jevent 追加到 flume agent 的 avrosource 列表中. 使用此 appender 的客户端必须在类路径中包含 flume-ng-sdk(例如, flume-ng-sdk-1.9.0.jar). 该 appender 支持循环和随机方案, 用于执行负载平衡. 它还支持可配置的退避超时, 以便从主机集临时删除向下 agent . 必需属性以粗体显示
| 属性名称 | 默认 | 描述 |
|---|---|---|
| Hosts | - | 一个以空格分隔的 host:port 列表，Flume（通过 AvroSource）正在侦听 event |
| Selector | ROUND_ROBIN | 选择机制。必须是 ROUND_ROBIN,RANDOM 或自定义 FQDN 到继承自 LoadBalancingSelector 的类。 |
| MaxBackoff | - | 一个 long 值，表示负载平衡客户端将从无法使用 event 的节点退回的最长时间（以毫秒为单位）。默认为无退避 |
| UnsafeMode | false | 如果为 true，则 appender 将不会在发送 event 失败时抛出异常。 |
| AvroReflectionEnabled | false | 使用 Avro Reflection 来序列化 Log4jevent 。 |
| AvroSchemaUrl | - | 可从中检索 Avro 架构的 URL。 |
使用默认值配置的示例 log4j.properties 文件:
#... log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppender log4j.appender.out2.Hosts = localhost:25430 localhost:25431 # configure a class's logger to output to the flume appender log4j.logger.org.example.MyClass = DEBUG,flume #...
使用 RANDOM 负载平衡配置的示例 log4j.properties 文件:
#... log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppender log4j.appender.out2.Hosts = localhost:25430 localhost:25431 log4j.appender.out2.Selector = RANDOM # configure a class's logger to output to the flume appender log4j.logger.org.example.MyClass = DEBUG,flume #...
使用退避配置的示例 log4j.properties 文件:
#... log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppender log4j.appender.out2.Hosts = localhost:25430 localhost:25431 localhost:25432 log4j.appender.out2.Selector = ROUND_ROBIN log4j.appender.out2.MaxBackoff = 30000 # configure a class's logger to output to the flume appender log4j.logger.org.example.MyClass = DEBUG,flume #...
安全
HDFSsink,HBasesink,Thriftsource,Thriftsink 和 Kite Datasetsink 都支持 Kerberos 身份验证. 请参阅相应的部分以配置与 Kerberos 相关的选项.
Flume agent 将作为单个主体对 kerberos KDC 进行身份验证, 这将由需要进行 kerberos 身份验证的不同组件使用. 为 Thriftsource,Thriftsink,HDFSsink,HBasesink 和 DataSetsink 配置的主体和 keytab 应该相同, 否则组件将无法启动.
监控
Flume 中的监控仍在进行中. 变化可能经常发生. 几个 Flume 组件向 JMX 平台 MBean 服务器报告度量标准. 可以使用 Jconsole 查询这些指标.
可用的组件指标
下表显示了可用于组件的度量标准. 每个组件仅维护一组度量, 由 "x" 表示, 未维护的值显示默认值, 即 0. 这些表告诉您可以在何处获得有意义的数据. 度量标准的名称应该足够描述, 有关更多信息, 您必须深入了解组件的 source 代码.
Sources1
| Avro | EXEC | HTTP | JMS | Kafka | MultiportSyslogTCP | Scribe | |
|---|---|---|---|---|---|---|---|
| AppendAcceptedCount | X | ||||||
| AppendBatchAcceptedCount | X | X | X | ||||
| AppendBatchReceivedCount | X | X | X | ||||
| AppendReceivedCount | X | ||||||
| ChannelWriteFail | X | X | X | X | X | X | |
| EventAcceptedCount | X | X | X | X | X | X | X |
| EventReadFail | X | X | X | X | X | ||
| EventReceivedCount | X | X | X | X | X | X | X |
| GenericProcessingFail | X | X | |||||
| KafkaCommitTimer | X | ||||||
| KafkaEmptyCount | X | ||||||
| KafkaEventGetTimer | X | ||||||
| OpenConnectionCount | X |
Sources 2
| SequenceGenerator | SpoolDirectory | SyslogTcp | SyslogUDP | Taildir | Thrift | |
|---|---|---|---|---|---|---|
| AppendAcceptedCount | X | |||||
| AppendBatchAcceptedCount | X | X | X | X | ||
| AppendBatchReceivedCount | X | X | X | |||
| AppendReceivedCount | X | |||||
| ChannelWriteFail | X | X | X | X | X | X |
| EventAcceptedCount | X | X | X | X | X | X |
| EventReadFail | X | X | X | X | ||
| EventReceivedCount | X | X | X | X | X | |
| GenericProcessingFail | X | X | ||||
| KafkaCommitTimer | ||||||
| KafkaEmptyCount | ||||||
| KafkaEventGetTimer | ||||||
| OpenConnectionCount |
Sinks 1
| Avro/Thrift | AsyncHBase | ElasticSearch | HBase 的 | HBase2 | |
|---|---|---|---|---|---|
| BatchCompleteCount | X | X | X | X | X |
| BatchEmptyCount | X | X | X | X | X |
| BatchUnderflowCount | X | X | X | X | X |
| ChannelReadFail | X | X | |||
| ConnectionClosedCount | X | X | X | X | X |
| ConnectionCreatedCount | X | X | X | X | X |
| ConnectionFailedCount | X | X | X | X | X |
| EventDrainAttemptCount | X | X | X | X | X |
| EventDrainSuccessCount | X | X | X | X | X |
| EventWriteFail | X | X | |||
| KafkaEventSendTimer | |||||
| RollbackCount |
Sinks 2
| HDFSEvent | Hive | HTTP | Kafka | Morphline | RollingFile | |
|---|---|---|---|---|---|---|
| BatchCompleteCount | X | X | X | |||
| BatchEmptyCount | X | X | X | X | ||
| BatchUnderflowCount | X | X | X | X | ||
| ChannelReadFail | X | X | X | X | X | X |
| ConnectionClosedCount | X | X | X | |||
| ConnectionCreatedCount | X | X | X | |||
| ConnectionFailedCount | X | X | X | |||
| EventDrainAttemptCount | X | X | X | X | X | |
| EventDrainSuccessCount | X | X | X | X | X | X |
| EventWriteFail | X | X | X | X | X | X |
| KafkaEventSendTimer | X | |||||
| RollbackCount | X |
Channels
| File | Kafka | Memory | PseudoTxnMemory | SpillableMemory | |
|---|---|---|---|---|---|
| ChannelCapacity | X | X | X | ||
| Channelize | X | X | X | X | |
| CheckpointBackupWriteErrorCount | X | ||||
| CheckpointWriteErrorCount | X | ||||
| EventPutAttemptCount | X | X | X | X | X |
| EventPutErrorCount | X | ||||
| EventPutSuccessCount | X | X | X | X | X |
| EventTakeAttemptCount | X | X | X | X | X |
| EventTakeErrorCount | X | ||||
| EventTakeSuccessCount | X | X | X | X | X |
| KafkaCommitTimer | X | ||||
| KafkaEventGetTimer | X | ||||
| KafkaEventSendTimer | X | ||||
| Open | X | ||||
| RollbackCounter | X | ||||
| Unhealthy | X |
JMX 报告
可以通过使用 flume-env.sh 在 JAVA_OPTS 环境变量中指定 JMX 参数来启用 JMX 报告, 如
export JAVA_OPTS ="- Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port = 5445 -Dcom.sun.management.jmxremote.authenticate = false -Dcom.sun.management.jmxremote.ssl = false"
注意: 上面的示例禁用安全性. 要启用安全性, 请参阅
Ganglia 报告
Flume 还可以将这些指标报告给 Ganglia 3 或 Ganglia 3.1 元节点. 要向 Ganglia 报告指标, 必须使用此支持启动水槽 agent . 必须通过传递以下参数来启动 Flume agent , 作为以 flume.monitoring 为前缀的系统属性., 可以在 flume-env.sh 中指定:
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 ganglia |
| hosts | - | 逗号分隔的 hostname 列表： Ganglia 服务器的端口 |
| pollFrequency | 60 | 连续向 Ganglia 服务器报告之间的时间（以秒为单位） |
| isGanglia3 | false | Ganglia 服务器版本为 3. 默认情况下，Flume 以 Ganglia 3.1 格式发送 |
我们可以通过 Ganglia 支持启动 Flume, 如下所示:
$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=com.example:1234,com.example2:5455
JSON 报告
Flume 还可以以 JSON 格式报告指标. 为了以 JSON 格式启用报告, Flume 在可配置端口上托管 Web 服务器. Flume 以以下 JSON 格式报告指标:
{
"typeName1.componentName1" : {
"metric1" : "metricValue1", "metric2" : "metricValue2"
},
"typeName2.componentName2" : {
"metric3" : "metricValue3", "metric4" : "metricValue4"
}
}
这是一个例子:
{
"CHANNEL.fileChannel":{"EventPutSuccessCount":"468085",
"Type":"CHANNEL",
"StopTime":"0",
"EventPutAttemptCount":"468086",
"ChannelSize":"233428",
"StartTime":"1344882233070",
"EventTakeSuccessCount":"458200",
"ChannelCapacity":"600000",
"EventTakeAttemptCount":"458288"},
"CHANNEL.memChannel":{"EventPutSuccessCount":"22948908",
"Type":"CHANNEL",
"StopTime":"0",
"EventPutAttemptCount":"22948908",
"ChannelSize":"5",
"StartTime":"1344882209413",
"EventTakeSuccessCount":"22948900",
"ChannelCapacity":"100",
"EventTakeAttemptCount":"22948908"}
}
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 http |
| port | 41414 | 启动服务器的端口。 |
$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=http -Dflume.monitoring.port=34545
然后, 可以在 http://<hostname>:<port>/metrics 网页上获得度量标准. 自定义组件可以报告上面 Ganglia 部分中提到的指标.
自定义报告
可以通过编写执行报告的服务器向其他系统报告指标. 任何报告类都必须实现 org.apache.flume.instrumentation.MonitorService 接口 . 这样的类可以与 GangliaServer 用于报告的方式相同. 他们可以轮询平台 mbean 服务器以轮询 mbeans 以获取指标. 例如, 如果可以使用名为 HTTPReporting 的 HTTP 监视服务, 如下所示:
$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=com.example.reporting.HTTPReporting -Dflume.monitoring.node=com.example:332
| 属性名称 | 默认 | 描述 |
|---|---|---|
| type | - | 组件 type 名称必须是 FQCN |
从自定义组件报告指标
任何自定义 flume 组件都应继承自 org.apache.flume.instrumentation.MonitoredCounterGroup 类. 然后, 该类应为其公开的每个度量标准提供 getter 方法. 请参阅下面的代码. MonitoredCounterGroup 需要一个属性列表, 其度量由此类公开. 截至目前, 此类仅支持将度量标准公开为长值.
public class SinkCounter extends MonitoredCounterGroup implements
SinkCounterMBean {
private static final String COUNTER_CONNECTION_CREATED =
"sink.connection.creation.count";
private static final String COUNTER_CONNECTION_CLOSED =
"sink.connection.closed.count";
private static final String COUNTER_CONNECTION_FAILED =
"sink.connection.failed.count";
private static final String COUNTER_BATCH_EMPTY =
"sink.batch.empty";
private static final String COUNTER_BATCH_UNDERFLOW =
"sink.batch.underflow";
private static final String COUNTER_BATCH_COMPLETE =
"sink.batch.complete";
private static final String COUNTER_EVENT_DRAIN_ATTEMPT =
"sink.event.drain.attempt";
private static final String COUNTER_EVENT_DRAIN_SUCCESS =
"sink.event.drain.sucess";
private static final String[] ATTRIBUTES = {
COUNTER_CONNECTION_CREATED, COUNTER_CONNECTION_CLOSED,
COUNTER_CONNECTION_FAILED, COUNTER_BATCH_EMPTY,
COUNTER_BATCH_UNDERFLOW, COUNTER_BATCH_COMPLETE,
COUNTER_EVENT_DRAIN_ATTEMPT, COUNTER_EVENT_DRAIN_SUCCESS
};
public SinkCounter(String name) {
super(MonitoredCounterGroup.Type.SINK, name, ATTRIBUTES);
}
@Override
public long getConnectionCreatedCount() {
return get(COUNTER_CONNECTION_CREATED);
}
public long incrementConnectionCreatedCount() {
return increment(COUNTER_CONNECTION_CREATED);
}
}
工具
文件 channel 完整性工具
文件 channel 完整性工具可验证文件 channel 中各个 event 的完整性, 并删除损坏的 event
这些工具可以按如下方式运行:
$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir -e org.apache.flume.MyEventValidator -DmaxSize 2000
其中 datadir 是要验证的数据目录的逗号分隔列表.
以下是可用选项
| 选项名称 | 描述 |
|---|---|
| h/help | 显示帮助 |
| l/dataDirs | 以逗号分隔的工具必须验证的数据目录列表 |
event 验证工具
event 验证器工具可用于以特定于应用程序的方式验证文件 channelevent . 该工具在每个 event 上应用用户提供程序验证登录, 并删除未向逻辑确认的 event
这些工具可以按如下方式运行:
$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir -e org.apache.flume.MyEventValidator -DmaxSize 2000
其中 datadir 是要验证的数据目录的逗号分隔列表.
以下是可用选项
| 选项名称 | 描述 |
|---|---|
| h/help | 显示帮助 |
| l/dataDirs | 以逗号分隔的工具必须验证的数据目录列表 |
| e/eventValidator | 完全合格的 event 验证器实现名称。jar 必须在 Flume 类路径上 |
Event 验证器实现必须实现 EventValidator 接口. 建议不要从实现中抛出任何异常, 因为它们被视为无效 event . 其他参数可以通过 - D 选项传递给 EventValitor 实现.
让我们看一个基于简单大小的 event 验证器的示例, 它将拒绝大于指定的最大大小的 event
public static class MyEventValidator implements EventValidator {
private int value = 0;
private MyEventValidator(int val) {
value = val;
}
@Override
public boolean validateEvent(Event event) {
return event.getBody() <= value;
}
public static class Builder implements EventValidator.Builder {
private int sizeValidator = 0;
@Override
public EventValidator build() {
return new DummyEventVerifier(sizeValidator);
}
@Override
public void configure(Context context) {
binaryValidator = context.getInteger("maxSize");
}
}
}
拓扑设计注意事项
Flume 非常灵活, 允许大量可能的部署方案. 如果您计划在大型生产部署中使用 Flume, 那么花一些时间考虑如何根据 Flume 拓扑来表达您的问题是明智的. 本节介绍一些注意事项
Flume 是否适合您的问题?
如果您需要将文本日志数据提取到 Hadoop / HDFS 中, 那么 Flume 最适合您的问题, 完全停止. 对于其他用例, 以下是一些指导原则:
Flume 旨在通过相对稳定, 可能复杂的拓扑来传输和摄取定期生成的 event 数据."event 数据" 的概念定义非常广泛. 对 Flume 来说, event 只是一个普通的字节. event 的大小有一些限制 - 例如, 它不能大于您可以存储在内存中或单个机器上的磁盘上 - 但实际上, 水槽 event 可以是从文本日志条目到图像文件的所有内容. event 的关键属性是它们以连续的流式方式生成. 如果您的数据没有定期生成(即您尝试将大量数据批量加载到 Hadoop 集群中), 那么 Flume 仍然可以正常工作, 但对您的情况来说可能有点过头了. Flume 喜欢相对稳定的拓扑结构. 您的拓扑不需要是不可变的, 因为 Flume 可以处理拓扑中的更改而不会丢失数据, 并且还可以容忍由于故障转移或配置而定期重新配置. 如果您每天都要更改拓扑结构, 那么它可能无法正常工作, 因为重新配置需要一些思考和开销.
Flume 中的流量可靠性
Flume 流的可靠性取决于几个因素. 通过调整这些因素, 您可以使用 Flume 实现各种可靠性选项.
您使用什么 type 的 channel .Flume 具有持久的 channel(将数据保存到磁盘的 channel)和非持久 channel(如果机器出现故障将丢失数据). 持久 channel 使用基于磁盘的存储, 存储在此类 channel 中的数据将在机器重启或非磁盘相关故障中持续存在.
您的 channel 是否已为工作量充分配置. Flume 中的 channel 充当各种跳跃的缓冲区. 这些缓冲区具有固定容量, 一旦该容量已满, 您将在流中的早期点创建背压. 如果此压力传播到流量 source,Flume 将变得不可用并可能丢失数据.
您是否使用冗余拓扑. Flume 让你在冗余拓扑中复制流. 这可以提供非常容易的容错 source, 并且可以克服磁盘或机器故障.
考虑 Flume 拓扑结构中可靠性的最佳方法是考虑各种故障情况及其结果. 如果磁盘发生故障会怎么样? 如果机器出现故障会怎样? 如果您的终端 sink(例如 HDFS)停机一段时间并且您有背压, 会发生什么? 可能的设计空间巨大, 但您需要提出的基本问题只是极少数.
Flow 拓扑设计
设计 Flume 拓扑的第一步是枚举数据的所有 source 和目标(终端 sink). 这些将定义拓扑的边缘点. 下一个考虑因素是是否引入中间聚合层或 event 路由. 如果要从大量 source 中收集数据, 则聚合数据以简化终端 sink 的提取可能会有所帮助. 聚合层还可以通过充当缓冲区来消除 source 的突发性或 sink 的不可用性. 如果要在不同位置之间路由数据, 您可能还希望在不同点分割流: 这会创建可能本身包含聚合点的子拓扑.
调整 Flume 部署的大小
一旦了解了拓扑的外观, 下一个问题就是需要多少硬件和网络容量. 首先, 量化您生成的数据量. 这并不总是一项简单的任务! 大多数数据流是突发性的 (例如, 由于昼夜模式) 并且可能是不可预测的. 一个好的起点是考虑每个拓扑层中的最大吞吐量, 包括每秒 event 数和每秒字节数. 一旦知道给定层的所需吞吐量, 就可以计算该层所需节点数的下限. 要确定可达到的吞吐量, 最好使用合成或采样 event 数据在硬件上试验 Flume. 通常, 基于磁盘的 channel 应该获得 10 的 MB / s, 而基于内存的 channel 应该达到 100 的 MB / s 或更多. 但性能会有很大差异, 具体取决于硬件和操作环境.
调整聚合吞吐量可以为每层提供所需节点数量的下限. 有几个原因需要额外的节点, 例如增加冗余和更好地吸收负载中的突发.
故障排除
处理 agent 失败
如果 Flume agent 程序关闭, 则该 agent 程序上托管的所有流程都将中止. 重新启动 agent 后, 将恢复流程. 使用文件 channel 或其他稳定 channel 的流将从中断处继续处理 event . 如果无法在同一硬件上重新启动 agent , 则可以选择将数据库迁移到另一个硬件并设置新的 Flume agent , 该 agent 可以继续处理 db 中保存的 event . 可以利用数据库 HA 期货将 Flume agent 移动到另一个主机.
兼容性
HDFS
目前 Flume 支持 HDFS 0.20.2 和 0.23.
组件摘要
| Component Interface | Type Alias | Implementation Class |
|---|---|---|
| org.apache.flume.Channel | memory | org.apache.flume.channel.MemoryChannel |
| org.apache.flume.Channel | jdbc | org.apache.flume.channel.jdbc.JdbcChannel |
| org.apache.flume.Channel | file | org.apache.flume.channel.file.FileChannel |
| org.apache.flume.Channel | – | org.apache.flume.channel.PseudoTxnMemoryChannel |
| org.apache.flume.Channel | – | org.example.MyChannel |
| org.apache.flume.Source | avro | org.apache.flume.source.AvroSource |
| org.apache.flume.Source | netcat | org.apache.flume.source.NetcatSource |
| org.apache.flume.Source | seq | org.apache.flume.source.SequenceGeneratorSource |
| org.apache.flume.Source | exec | org.apache.flume.source.ExecSource |
| org.apache.flume.Source | syslogtcp | org.apache.flume.source.SyslogTcpSource |
| org.apache.flume.Source | multiport_syslogtcp | org.apache.flume.source.MultiportSyslogTCPSource |
| org.apache.flume.Source | syslogudp | org.apache.flume.source.SyslogUDPSource |
| org.apache.flume.Source | spooldir | org.apache.flume.source.SpoolDirectorySource |
| org.apache.flume.Source | http | org.apache.flume.source.http.HTTPSource |
| org.apache.flume.Source | thrift | org.apache.flume.source.ThriftSource |
| org.apache.flume.Source | jms | org.apache.flume.source.jms.JMSSource |
| org.apache.flume.Source | – | org.apache.flume.source.avroLegacy.AvroLegacySource |
| org.apache.flume.Source | – | org.apache.flume.source.thriftLegacy.ThriftLegacySource |
| org.apache.flume.Source | – | org.example.MySource |
| org.apache.flume.Sink | null | org.apache.flume.sink.NullSink |
| org.apache.flume.Sink | logger | org.apache.flume.sink.LoggerSink |
| org.apache.flume.Sink | avro | org.apache.flume.sink.AvroSink |
| org.apache.flume.Sink | hdfs | org.apache.flume.sink.hdfs.HDFSEventSink |
| org.apache.flume.Sink | hbase | org.apache.flume.sink.hbase.HBaseSink |
| org.apache.flume.Sink | hbase2 | org.apache.flume.sink.hbase2.HBase2Sink |
| org.apache.flume.Sink | asynchbase | org.apache.flume.sink.hbase.AsyncHBaseSink |
| org.apache.flume.Sink | elasticsearch | org.apache.flume.sink.elasticsearch.ElasticSearchSink |
| org.apache.flume.Sink | file_roll | org.apache.flume.sink.RollingFileSink |
| org.apache.flume.Sink | irc | org.apache.flume.sink.irc.IRCSink |
| org.apache.flume.Sink | thrift | org.apache.flume.sink.ThriftSink |
| org.apache.flume.Sink | – | org.example.MySink |
| org.apache.flume.ChannelSelector | replicating | org.apache.flume.channel.ReplicatingChannelSelector |
| org.apache.flume.ChannelSelector | multiplexing | org.apache.flume.channel.MultiplexingChannelSelector |
| org.apache.flume.ChannelSelector | – | org.example.MyChannelSelector |
| org.apache.flume.SinkProcessor | default | org.apache.flume.sink.DefaultSinkProcessor |
| org.apache.flume.SinkProcessor | failover | org.apache.flume.sink.FailoverSinkProcessor |
| org.apache.flume.SinkProcessor | load_balance | org.apache.flume.sink.LoadBalancingSinkProcessor |
| org.apache.flume.SinkProcessor | – | |
| org.apache.flume.interceptor.Interceptor | timestamp | org.apache.flume.interceptor.TimestampInterceptor$Builder |
| org.apache.flume.interceptor.Interceptor | host | org.apache.flume.interceptor.HostInterceptor$Builder |
| org.apache.flume.interceptor.Interceptor | static | org.apache.flume.interceptor.StaticInterceptor$Builder |
| org.apache.flume.interceptor.Interceptor | regex_filter | org.apache.flume.interceptor.RegexFilteringInterceptor$Builder |
| org.apache.flume.interceptor.Interceptor | regex_extractor | org.apache.flume.interceptor.RegexFilteringInterceptor$Builder |
| org.apache.flume.channel.file.encryption.KeyProvider$Builder | jceksfile | org.apache.flume.channel.file.encryption.JCEFileKeyProvider |
| org.apache.flume.channel.file.encryption.KeyProvider$Builder | – | org.example.MyKeyProvider |
| org.apache.flume.channel.file.encryption.CipherProvider | aesctrnopadding | org.apache.flume.channel.file.encryption.AESCTRNoPaddingProvider |
| org.apache.flume.channel.file.encryption.CipherProvider | – | org.example.MyCipherProvider |
| org.apache.flume.serialization.EventSerializer$Builder | text | org.apache.flume.serialization.BodyTextEventSerializer$Builder |
| org.apache.flume.serialization.EventSerializer$Builder | avro_event | org.apache.flume.serialization.FlumeEventAvroEventSerializer$Builder |
| org.apache.flume.serialization.EventSerializer$Builder | – | org.example.MyEventSerializer$Builder |
别名约定
这些别名的约定在上面的组件特定示例中使用, 以使所有示例中的名称保持简短和一致.
| Alias Name | Alias Type |
|---|---|
| a | agent |
| c | channel |
| r | source |
| k | sink |
| g | sink group |
| i | interceptor |
| y | key |
| h | host |
| s | serializer |