flume sink 简介及官方用例

1、HDFS Sink

此sink将事件写入 Hadoop 分布式文件系统（HDFS）中。它目前支持创建文本和序列文件。它支持两种文件类型的压缩。可以根据经过的时间或数据大小或事件数定期滚动文件（关闭当前文件并创建一个新文件）。它还按事件起源的时间戳或计算机等属性对数据进行存储/分区。HDFS 目录路径可能包含格式转义序列，这些转义序列将由 HDFS 接收器替换，以生成用于存储事件的目录/文件名。使用此 sink 需要安装 hadoop，以便 Flume 可以使用 Hadoop jar 与 HDFS 集群进行通信。请注意，需要支持 sync（）调用的 Hadoop 版本。

注意对于所有与时间相关的转义序列，事件的标头中必须存在键为“timestamp”的标头（除非 hdfs.useLocalTimeStamp 设置为 true）。自动添加此功能的一种方法是使用 TimestampInterceptor。

Name	Default	Description
channel	–
type	–	The component type name, needs to be hdfs
hdfs.path	–	HDFS directory path (eg hdfs://namenode/flume/webdata/)

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，则 hdfs 路径将变为 /flume/events/2012-06-12/1150/00。

2、Hive Sink

此接收器将包含分隔文本或 JSON 数据的事件直接流式传输到 Hive 表或分区中。事件是使用 Hive 事务写入的。一旦将一组事件提交到 Hive，Hive 查询就会立即看到这些事件。Flume 将流式传输到的分区可以预先创建，也可以选择性地在缺少分区时创建它们。传入事件数据中的字段将映射到 Hive 表中的相应列。

Name	Default	Description
channel	–
type	–	The component type name, needs to be hive
hive.metastore	–	Hive metastore URI (eg thrift://a.b.com:9083 )
hive.database	–	Hive database name
hive.table	–	Hive table name
hive.partition	–	Comma separate list of partition values identifying the partition to write to. May contain escape sequences. E.g: If the table is partitioned by (continent: string, country :string, time : string) then ‘Asia,India,2014-02-26-01-21’ will indicate continent=Asia,country=India,time=2014-02-26-01-21
serializer		Serializer is responsible for parsing out field from the event and mapping them to columns in the hive table. Choice of serializer depends upon the format of the data in the event. Supported serializers: DELIMITED and JSON

Example Hive table :

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 = "\t"

a1.sinks.k1.serializer.serdeSeparator = '\t'

a1.sinks.k1.serializer.fieldnames =id,,msg

上述配置会将时间戳四舍五入到最后 10 分钟。例如，如果事件的时间戳标头设置为 2012 年 6 月 12 日上午 11：54：34，并且“country”标头设置为“india”，则该事件的计算结果将达到分区（continent='asia'，country='india'，time='2012-06-12-11-50'）。序列化程序配置为接受包含三个字段的制表符分隔输入，并跳过第二个字段。

3、Logger Sink

在 INFO 级别记录事件。通常可用于测试/调试目的。

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be logger
maxBytesToLog	16	Maximum number of bytes of the Event body to log

Example for agent named a1:

a1.channels = c1

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

4、Avro Sink

发送到此sink的 Flume 事件将转换为 Avro 事件，并发送到配置的主机名/端口对。事件以配置的批处理大小的批处理方式从配置的通道中获取。

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be avro.
hostname	–	The hostname or IP address to bind to.
port	–	The port # to listen on.

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

5、Thrift Sink

发送到此接收器的 Flume 事件将转换为 Thrift 事件，并发送到配置的主机名/端口对。事件从配置的通道中按配置的批处理大小批量获取

通过启用 kerberos 身份验证，可以将 Thrift sink 配置为在安全模式下启动。要与以安全模式启动的 Thrift 源进行通信，Thrift 接收器也应在安全模式下运行。client-principal 和 client-keytab 是 Thrift 接收器用于向 kerberos KDC 进行身份验证的属性。server-principal 表示此接收器配置为在安全模式下连接到的 Thrift 源的主体。

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be thrift.
hostname	–	The hostname or IP address to bind to.
port	–	The port # to listen on.

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

6、File Roll Sink

将事件存储在本地文件系统上

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be file_roll.
sink.directory	–	The directory where files will be stored

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

7、IRC Sink

IRC 接收器从附加通道获取消息，并将这些消息中继到配置的 IRC 目标。

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be irc
hostname	–	The hostname or IP address to connect to
port	6667	The port number of remote host to connect
nick	–	Nick name
user	–	User name
password	–	User password
chan	–	channel

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

8、HBaseSinks¶.

此接收器将数据写入 HBase。Hbase 配置是从类路径中遇到的第一个hbase-site.xml中选取的。实现 HbaseEventSerializer 的类（由配置指定）用于将事件转换为 HBase 全量和/或增量。然后将这些全量和增量写入 HBase。此接收器提供与 HBase 相同的一致性保证，HBase 目前是逐行原子性。如果 Hbase 无法写入某些事件，接收器将回滚该事务中的所有事件。

HBaseSink 支持写入数据以保护 HBase。若要写入安全 HBase，代理正在运行的用户必须对接收器配置为写入的表具有写入权限。可以在配置中指定用于对 KDC 进行身份验证的主体和密钥表。Flume 代理类路径中的hbase-site.xml必须将身份验证设置为 kerberos（有关如何执行此操作的详细信息，请参阅 HBase 文档）

为方便起见，Flume 提供了两个序列化器。SimpleHbaseEventSerializer （org.apache.flume.sink.hbase.SimpleHbaseEventSerializer）将事件正文按原样写入 HBase，并选择性地在 Hbase 中递增列。RegexHbaseEventSerializer （org.apache.flume.sink.hbase.RegexHbaseEventSerializer）根据给定的正则表达式中断事件正文，并将每个部分写入不同的列中。类型为 FQCN：org.apache.flume.sink.hbase.HBaseSink。

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be hbase
table	–	The name of the table in Hbase to write to.
columnFamily	–	The column family in Hbase to write to.

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

9、HBase2Sink

HBase2Sink 相当于 HBase 版本 2 的 HBaseSink。提供的功能和配置参数与 HBaseSink 相同（sink 类型中的 hbase2 标记和包/类名称除外）。类型为FQCN: org.apache.flume.sink.hbase2.HBase2Sink

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be hbase2
table	–	The name of the table in HBase to write to.
columnFamily	–	The column family in HBase to write to.

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

10、AsyncHBaseSink

此接收器使用异步模型将数据写入 HBase。实现 AsyncHbaseEventSerializer 的类（由配置指定）用于将事件转换为 HBase 全量和/或增量。然后将这些看全量和增量写入 HBase。此接收器使用 Asynchbase API 写入 HBase。此接收器提供与 HBase 相同的一致性保证，HBase 目前是逐行原子性。如果 Hbase 无法写入某些事件，接收器将回滚该事务中的所有事件。AsyncHBaseSink 只能与 HBase 1.x 一起使用。AsyncHBaseSink 使用的异步客户端库不适用于 HBase 2。类型为 FQCN：org.apache.flume.sink.hbase.AsyncHBaseSink。

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be asynchbase
table	–	The name of the table in Hbase to write to.
columnFamily	–	The column family in Hbase to write to.

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

11、MorphlineSolrSink

该sink从 Flume 事件中提取数据，对其进行转换，并将其近乎实时地加载到 Apache Solr 服务器中，而 Apache Solr 服务器又向最终用户或搜索应用程序提供查询。

此接收器非常适合将原始数据流式传输到 HDFS（通过 HdfsSink）并同时提取、转换和加载相同数据到 Solr（通过 MorphlineSolrSink）的用例。具体而言，此sink可以处理来自不同数据源的任意异构原始数据，并将其转换为对搜索应用程序有用的数据模型。

ETL 功能可使用 morphline 配置文件进行自定义，该文件定义了一系列转换命令，这些命令将事件记录从一个命令传递到另一个命令。

Morphlines 可以看作是 Unix 管道的演变，其中数据模型被推广为处理通用记录流，包括任意二进制有效负载。morphline 命令有点像 Flume Interceptor。Morphlines 可以嵌入到 Hadoop 组件（如 Flume）中

提供用于解析和转换一组标准数据格式（如日志文件、Avro、CSV、文本、HTML、XML、PDF、Word、Excel 等）的命令，并且可以添加用于其他数据格式的其他自定义命令和解析器作为 morphline 插件。可以对任何类型的数据格式进行索引，并且可以生成任何类型的Solr模式的任何Solr文档，并且可以注册和执行任何自定义ETL逻辑。

Morphlines 操作连续的记录流。数据模型可以描述如下：记录是一组命名字段，其中每个字段都有一个或多个值的有序列表。值可以是任何 Java 对象。也就是说，记录本质上是一个哈希表，其中每个哈希表条目都包含一个 String 键和一个 Java 对象列表作为值。（该实现使用 Guava 的 ArrayListMultimap，即 ListMultimap）。请注意，一个字段可以有多个值，并且任何两个记录都不需要使用通用字段名称。

此sink将 Flume 事件的主体填充到 morphline 记录的 _attachment_body 字段中，并将 Flume 事件的标头复制到同名的记录字段中。然后，命令可以对此数据执行操作。

支持路由到 SolrCloud 集群以提高可扩展性。索引负载可以分布在大量 MorphlineSolrSink 上，以提高可伸缩性。索引负载可以在多个MorphlineSolrSinks之间复制，以实现高可用性，例如使用Flume功能，如负载平衡Sink Processor。MorphlineInterceptor 还可以帮助实现到多个 Solr 集合的动态路由（例如，用于多租户）。

您的环境所需的 morphline 和 solr jar 必须放在 Apache Flume 安装的 lib 目录中。

The type is the FQCN: org.apache.flume.sink.solr.morphline.MorphlineSolrSink

Property Name	Default	Description
channel	–
type	–	The component type name, needs to be org.apache.flume.sink.solr.morphline.MorphlineSolrSink
morphlineFile	–	The relative or absolute path on the local file system to the morphline configuration file. Example: /etc/flume-ng/conf/morphline.conf

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

12、Kafka Sink

可以将数据发布到 Kafka topic。其中一个目标是将 Flume 与 Kafka 集成，以便基于拉取的处理系统可以处理来自各种 Flume 源的数据。

目前支持 Kafka 服务器版本 0.10.1.0 或更高版本。测试完成到 2.0.1，这是发布时最高的可用版本。

Property Name	Default	Description
type	–	Must be set to org.apache.flume.sink.kafka.KafkaSink
kafka.bootstrap.servers	–	List of brokers Kafka-Sink will connect to, to get the list of topic partitions This can be a partial list of brokers, but we recommend at least two for HA. The format is comma separated list of hostname:port

kafka Sink 使用 FlumeEvent 标头中的主题和键属性将事件发送到 Kafka。如果标头中存在主题，则事件将发送到该特定主题，覆盖为 Sink 配置的主题。如果标头中存在 key，则 Kafka 将使用该 key 在主题分区之间对数据进行分区。具有相同键的事件将发送到同一分区。如果键为 null，则事件将发送到随机分区。

下面给出了 Kafka 接收器的示例配置。以前缀 kafka.producer 开头的属性，即 Kafka 生产者。创建 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

Flume 和 Kafka 之间的通信通道支持安全认证和数据加密。对于安全身份验证，可以从 Kafka 版本 0.9.0 开始使用 SASL/GSSAPI （Kerberos V5）或 SSL（即使参数名为 SSL，实际协议是 TLS 实现）。

截至目前，数据加密仅由 SSL/TLS 提供。

将 kafka.producer.security.protocol 设置为以下任一值意味着：