flume基础（二）

秒速五厘米 2022-05-21 02:47 121阅读 0赞

Flume 的 Source、Channel和Sink的内置类型如下：

（可参考官网修改配置  http://flume.apache.org/FlumeUserGuide.html）

Flume Source

<table> 
 <tbody> 
 <tr> 
 <td> Source类型 </td> 
 <td> 说明 </td> 
 </tr> 
 <tr> 
 <td> Avro Source </td> 
 <td> 支持Avro协议（实际上是Avro RPC），内置支持 </td> 
 </tr> 
 <tr> 
 <td> Thrift Source </td> 
 <td> 支持Thrift协议，内置支持 </td> 
 </tr> 
 <tr> 
 <td> Exec Source </td> 
 <td> 基于Unix的command在标准输出上生产数据 </td> 
 </tr> 
 <tr> 
 <td> JMS Source </td> 
 <td> 从JMS系统（消息、主题）中读取数据，ActiveMQ已经测试过 </td> 
 </tr> 
 <tr> 
 <td> Spooling Directory Source </td> 
 <td> 监控指定目录内数据变更 </td> 
 </tr> 
 <tr> 
 <td> Taildir Source </td> 
 <td> &nbsp; </td> 
 </tr> 
 <tr> 
 <td> Twitter 1% firehose Source </td> 
 <td> 通过API持续下载Twitter数据，试验性质 </td> 
 </tr> 
 <tr> 
 <td> Kalfka Source </td> 
 <td> 支持Kafka </td> 
 </tr> 
 <tr> 
 <td> Netcat Source </td> 
 <td> 监控某个端口，将流经端口的每一个文本行数据作为Event输入 </td> 
 </tr> 
 <tr> 
 <td> Sequence Generator Source </td> 
 <td> 序列生成器数据源，生产序列数据 </td> 
 </tr> 
 <tr> 
 <td> Syslog Sources </td> 
 <td> 读取syslog数据，产生Event，支持UDP和TCP两种协议 </td> 
 </tr> 
 <tr> 
 <td> HTTP Source </td> 
 <td> 基于HTTP POST或GET方式的数据源，支持JSON、BLOB表示形式 </td> 
 </tr> 
 <tr> 
 <td> Stress Source </td> 
 <td> &nbsp; </td> 
 </tr> 
 <tr> 
 <td> Legacy Sources </td> 
 <td> 兼容老的Flume OG中Source（0.9.x版本） </td> 
 </tr> 
 <tr> 
 <td> Custom Source </td> 
 <td> 自定义Source实现 </td> 
 </tr> 
 <tr> 
 <td> Scribe Source </td> 
 <td> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

注：在1.6.0版本才加入了Kafka Source。

source是被动接收数据的。

Flume Channel

<table> 
 <tbody> 
 <tr> 
 <td> Channel类型 </td> 
 <td> 说明 </td> 
 </tr> 
 <tr> 
 <td> Memory Channel </td> 
 <td> Event数据存储在内存中 </td> 
 </tr> 
 <tr> 
 <td> JDBC Channel </td> 
 <td> Event数据存储在持久化存储中，当前Flume Channel内置支持Derby </td> 
 </tr> 
 <tr> 
 <td> Kafka Channel </td> 
 <td> Event数据存储在Kafka cluster中 </td> 
 </tr> 
 <tr> 
 <td> File Channel </td> 
 <td> Event数据存储在磁盘文件中 </td> 
 </tr> 
 <tr> 
 <td> Spillable Memory Channel </td> 
 <td> Event数据存储在内存中和磁盘上，当内存队列满了，会持久化到磁盘文件（当前试验性的，不建议生产环境使用） </td> 
 </tr> 
 <tr> 
 <td> Pseudo Transaction Channel </td> 
 <td> 测试用途 </td> 
 </tr> 
 <tr> 
 <td> Custom Channel </td> 
 <td> 自定义Channel实现 </td> 
 </tr> 
 </tbody> 
</table>

注：flume的处理文件的速度大概在十几M每秒。

Flume Sink

<table> 
 <tbody> 
 <tr> 
 <td> Sink类型 </td> 
 <td> 说明 </td> 
 </tr> 
 <tr> 
 <td> HDFS Sink </td> 
 <td> 数据写入HDFS </td> 
 </tr> 
 <tr> 
 <td> Hive Sink </td> 
 <td> &nbsp; </td> 
 </tr> 
 <tr> 
 <td> Logger Sink </td> 
 <td> 数据写入日志文件 </td> 
 </tr> 
 <tr> 
 <td> Avro Sink </td> 
 <td> 数据被转换成Avro Event，然后发送到配置的RPC端口上 </td> 
 </tr> 
 <tr> 
 <td> Thrift Sink </td> 
 <td> 数据被转换成Thrift Event，然后发送到配置的RPC端口上 </td> 
 </tr> 
 <tr> 
 <td> IRC Sink </td> 
 <td> 数据在IRC上进行回放 </td> 
 </tr> 
 <tr> 
 <td> File Roll Sink </td> 
 <td> 存储数据到本地文件系统 </td> 
 </tr> 
 <tr> 
 <td> Null Sink </td> 
 <td> 丢弃到所有数据 </td> 
 </tr> 
 <tr> 
 <td> HBase Sink </td> 
 <td> 数据写入HBase数据库 </td> 
 </tr> 
 <tr> 
 <td> Morphline Solr Sink </td> 
 <td> 数据发送到Solr搜索服务器（集群） </td> 
 </tr> 
 <tr> 
 <td> ElasticSearch Sink </td> 
 <td> 数据发送到Elastic Search搜索服务器（集群） </td> 
 </tr> 
 <tr> 
 <td> Kite Dataset Sink </td> 
 <td> 写数据到Kite Dataset，试验性质的 </td> 
 </tr> 
 <tr> 
 <td> Kafka Sink </td> 
 <td> 数据写入Kafka topic </td> 
 </tr> 
 <tr> 
 <td> Custom Sink </td> 
 <td> 自定义Sink实现 </td> 
 </tr> 
 </tbody> 
</table>

另外还有ChannelSelector、Sink Processor、EventSerializer、Interceptor等组件，可以参考官网提供的用户手册。

存储到hdfs，要注意存储速度，几十M/s。条测性能的可以扩大机器数。

Flume Interceptors

<table> 
 <tbody> 
 <tr> 
 <td> Interceptor类型 </td> 
 <td> 说明 </td> 
 </tr> 
 <tr> 
 <td> Timestamp Interceptor </td> 
 <td> 在event的header中添加一个key叫：timestamp,value为当前的时间戳。 </td> 
 </tr> 
 <tr> 
 <td> Host Interceptor </td> 
 <td> 在event的header中添加一个key叫：host,value为当前机器的hostname或者ip。 </td> 
 </tr> 
 <tr> 
 <td> Static Interceptor </td> 
 <td> 可以在event的header中添加自定义的key和value。 </td> 
 </tr> 
 <tr> 
 <td> UUID Interceptor </td> 
 <td> &nbsp;用于在每个events header中生成一个UUID字符串，生成的UUID可以在sink中读取并使用 </td> 
 </tr> 
 <tr> 
 <td> Morphline Interceptor </td> 
 <td> &nbsp;http://kitesdk.org/docs/current/morphlines/morphlines-reference-guide.html 基本相当于一个ETL 工具 </td> 
 </tr> 
 <tr> 
 <td> Search and Replace Interceptor </td> 
 <td> &nbsp; </td> 
 </tr> 
 <tr> 
 <td> Regex Filtering Interceptor </td> 
 <td> 通过正则来清洗或包含匹配的events。 </td> 
 </tr> 
 <tr> 
 <td> Regex Extractor Interceptor </td> 
 <td> 通过正则表达式来在header中添加指定的key,value则为正则匹配的部分。 </td> 
 </tr> 
 </tbody> 
</table>

Flume Channel Selectors

If the type isnot specified, then defaults to “replicating”.