环境
centos：7.2
JDK:1.8
Flume:1.8
一、Flume 安装
1) 下载
wget http://mirrors.tuna.tsinghua.edu.cn/apache/flume/1.8.0/apache-flume-1.8.0-bin.tar.gz
2) 解压
tar –zxvf apache-flume-1.8.0-bin.tar.gz
mv apache-flume-1.8.0-bin /usr/local/flume
3) 设置环境变量
Vim /etc/profile.d/flume.sh
Source /etc/profile
4) 配置java_home
cp flume-env.sh.template flume-env.sh
vim flume-env.sh
5) 配置详解
主要介绍poolingDirectory Source，HDFS Sink，MemoryChannel， File Channel 四种，其他请参考官方文档。
- spoolingDirectory Source
Spooling Directory Source可以获取硬盘上“spooling”目录的数据，这个Source将监视指定目录是否有新文件，如果有新文件的话，就解析这个新文件。事件的解析逻辑是可插拔的。在文件的内容所有的都读取到Channel之后，Spooling Directory Source会重名或者是删除该文件以表示文件已经读取完成。
不像Exec Source，这个Source是可靠的，且不会丢失数据。即使Flume重启或者被Kill。但是需要注意如下两点:
- 如果文件在放入spooling目录之后还在写，那么Flume会打印错误日志，并且停止处理该文件。
- 如果文件之后重复使用，Flume将打印错误日志，并且停止处理。
为了避免以上问题，我们可以使用唯一的标识符来命令文件，例如：时间戳。
属性名 | 默认 | 描述 |
channels | – |
|
type | – | 组件名：spooldir. |
spoolDir | – | 读取文件的目录。 |
fileSuffix | .COMPLETED | Spooling读取过的文件，添加的后缀。 |
deletePolicy | never | 完成后的文件是否删除。never：不删除 或 immediate：立即删除 |
fileHeader | FALSE | 是不把路径加入到Heander |
fileHeaderKey | file | 路径加入到Header的Key是什么 |
basenameHeader | FALSE | 是不把文件名加入到Heander |
basenameHeaderKey | basename | 文件名加入到Header的Key是什么 |
ignorePattern | ^$ | 采用正则表达是去过滤一些文件。只有符合正则表达式的文件才会被使用。 |
trackerDir | .flumespool | 被处理文件的元数据的存储目录，如果不是绝对路径，就被会解析到spoolDir目录下。 |
consumeOrder | oldest | 消费spooling目录文件的规则，分别有：oldest，youngest和random。在oldest 和 youngest的情况下，通过文件的最后修改时间来比较文件。如果最后修改时间相同，就根据字典的序列从小开始。在随机的情况下，就随意读取文件。如果文件列表很长，采用oldest/youngest可能会很慢，因为用oldest/youngest要扫描文件。但是如果采用random的话，就可能造成新的文件消耗的很快，老的文件一直都没有被消费。 |
maxBackoff | 4000 | 如果Channel已经满了，那么该Source连续尝试写入该Channel的最长时间（单位：毫秒）。 |
batchSize | 100 | 批量传输到Channel的粒度。 |
inputCharset | UTF-8 | 字符集 |
decodeErrorPolicy | FAIL | 在文件中有不可解析的字符时的解析策略。FAIL: 抛出一个异常，并且不能解析该文件。REPLACE: 取代不可解析的字符，通常用Unicode U+FFFD. IGNORE: 丢弃不可能解析字符序列。 |
deserializer | LINE | 自定序列化的方式，自定的话，必须实现EventDeserializer.Builder. |
deserializer.* |
|
|
bufferMaxLines | – | 已废弃。 |
bufferMaxLineLength | 5000 | (不推荐使用) 一行中最大的长度，可以使用deserializer.maxLineLength代替。 |
selector.type | replicating | replicating（复制） 或 multiplexing（复用） |
selector.* |
| 取决于selector.type的值 |
interceptors | – | 空格分割的interceptor列表。 |
interceptors.* |
|
|
2.HDFS Sink
属性名 | 默认值 | 描述 |
channel | – |
|
type | – | 组件的名称，必须为：HDFS |
hdfs.path | – | HDFS目录路径，例如：hdfs://namenode/flume/webdata/ |
hdfs.filePrefix | FlumeData | HDFS目录中，由Flume创建的文件前缀。 |
hdfs.fileSuffix | – | 追加到文件的后缀，例如：.txt |
hdfs.inUsePrefix | – | 文件正在写入时的前缀。 |
hdfs.inUseSuffix | .tmp | 文件正在写入时的后缀。 |
hdfs.rollInterval | 30 | 当前写入的文件滚动间隔，默认30秒生成一个新的文件 (0 = 不滚动) |
hdfs.rollSize | 1024 | 以文件大小触发文件滚动，单位字节(0 = 不滚动) |
hdfs.rollCount | 10 | 以写入的事件数触发文件滚动。(0 = 不滚动) |
hdfs.idleTimeout | 0 | 超时多久以后关闭无效的文件。(0 = 禁用自动关闭的空闲文件)但是还是可能因为网络等多种原因导致，正在写的文件始终没有关闭，从而产生tmp文件 |
hdfs.batchSize | 100 | 有多少Event后，写到文件才刷新到HDFS。 |
hdfs.codeC | – | 压缩编解码器，可以使用：gzip, bzip2, lzo, lzop, snappy |
hdfs.fileType | SequenceFile | 文件格式：通常使用SequenceFile（默认）, DataStream 或者 CompressedStream |
hdfs.maxOpenFiles | 5000 | HDFS中允许打开文件的数据，如果数量超过了，最老的文件将被关闭。 |
hdfs.callTimeout | 10000 | 允许HDFS操作的毫秒数，例如：open，write, flush, close。如果很多HFDS操作超时，这个配置应该增大。 |
hdfs.threadsPoolSize | 10 | 每个HDFS sink的HDFS的IO操作线程数（例如：open，write） |
hdfs.rollTimerPoolSize | 1 | 每个HDFS sink调度定时文件滚动的线程数。 |
hdfs.kerberosPrincipal | – | 安全访问HDFS Kerberos的主用户。 |
hdfs.kerberosKeytab | – | 安全访问HDFS Kerberos keytab |
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 header的时间戳) |
hdfs.closeTries | 0 | 在发起一个关闭命令后，HDFS sink必须尝试重命名文件的次数。如果设置为1，重命名失败后，HDFS sink不会再次尝试重命名该文件，这个文件处于打开状态，并且用.tmp作为扩展名。如果为0，Sink会一直尝试重命名，直至重命名成功。如果文件 失败，这个文件可能一直保持打开状态，但是这种情况下数据是完整的。文件将会在Flume下次重启时被关闭。 |
hdfs.retryInterval | 180 | 在几秒钟之间连续尝试关闭文件。每个关闭请求都会有多个RPC往返Namenode，因此设置的太低可能导致Namenode超负荷，如果设置0或者更小，如果第一次尝试失败的话，该Sink将不会尝试关闭文件。并且把文件打开，或者用“.tmp”作为扩展名。 |
serializer | TEXT | 可能的选项包括avro_event或继承了EventSerializer.Builder接口的类名。 |
serializer.* |
|
|
3.MemoryChannel
事件存储在具有可配置的最大大小的内存队列中。对于需要更高吞吐量的流程来说是理想的，并且在代理失败的情况下准备丢失阶段数据。
属性名 | 默认值 | 描述 |
type | - | memory |
capacity | 100 | MemroyChannel的容量 |
transactionCapacity | 100 | 每个事务最大的容量，也就是每个事务能够获取的最大Event数量。默认也是100 |
Keep-alive | 3 | 增加和删除一个Event的超时时间（单位：秒） |
byteCapacityBufferPercentage | 20 | 义Channle中Event所占的百分比，需要考虑在Header中的数据。 |
4.File Channel
属性名 | 默认值 | 描述 |
type | – | file |
checkpointDir | ~/.flume/file-channel/checkpoint | 存储文件检查点目录 |
useDualCheckpoints | FALSE | 备份检查点. 如果为true, backupCheckpointDir 必填 |
backupCheckpointDir | – | 文件检查点的备份目录，不能和checkpointDir相同。 |
dataDirs | ~/.flume/file-channel/data | 用于存储日志文件目录，多目录用逗号分隔。 |
transactionCapacity | 10000 | 通道支持的事务的最大大小。 |
checkpointInterval | 30000 | 检查点时间间隔（毫秒） |
maxFileSize | 2146435071 | 单个文件最大大小（字节） |
minimumRequiredSpace | 524288000 | 最小需要的*空间（以字节为单位）。为了避免数据损坏，当空闲空间低于该值时，文件通道停止接受“取/放”请求。 |
capacity | 1000000 | 信道最大容量 |
keep-alive | 3 | 等待放置操作时间（秒） |
use-log-replay-v1 | FALSE | Expert: 使用旧的回放逻辑 |
use-fast-replay | FALSE | Expert: 不使用队列回放 |
checkpointOnClose | TRUE | 控制是否在关闭通道时创建检查点。在关闭时创建检查点通过避免重放加速文件通道的后续启动. |
encryption.activeKey | – | 用于加密新数据的**名称 |
encryption.cipherProvider | – | **提供者类型：AESCTRNOPADDING |
encryption.keyProvider | – | **提供程序类型: JCEKSFILE |
encryption.keyProvider.keyStoreFile | – | **存储文件路径 |
encrpytion.keyProvider.keyStorePasswordFile | – | **存储密码文件路径 |
encryption.keyProvider.keys | – | 所有**列表 (e.g. history of the activeKey setting) |
encyption.keyProvider.keys.*.passwordFile | – | 可选**密码文件的路径 |
6) 验证
l 查看版本：
l 执行测试命令：
>cd /usr/local/flume/conf/
>cp flume-conf.properties.templateflume-conf.properties
>flume-ng agent –conf conf –conf-file flume-conf.properties –name a1
-Dflume.root.logger=INFO,console
l 出现错误：
Warning:No configuration directory set! Use --conf <dir> to override.
Warning:JAVA_HOME is not set!
Info:Including Hive libraries found via () for Hive access
+ exec/usr/local/jdk1.8/bin/java -Xmx20m -cp '/usr/local/flume/lib/*:/lib/*'-Djava.library.path= org.apache.flume.node.Application -n a1 -fexample-conf.properties
log4j:WARNNo appenders could be found for logger (org.apache.flume.node.Application).
log4j:WARNPlease initialize the log4j system properly.
log4j:WARN Seehttp://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
命令格式错误,应改为如下：
>flume-ngagent --conf /usr/local/flume/conf –-conf-file /usr/local/flume/flume-conf.properties -–namea1
-Dflume.root.logger=INFO,console