2018-11-03 18:44:44,913 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO – org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{} body: 61 20 61 20 61 61 61 20 61 20 0D                a a aaa a . }

commands:

  help                      显示本帮助信息

  agent                     启动一个agent进程

  avro-client                 启动一个用于测试avro source的客户端（能够发送avro序列化流）

  version                    显示当前flume的版本信息

global options:   全局通用选项

  –conf,-c <conf>          指定flume的系统配置文件所在目录

  –classpath,-C <cp>        添加额外的jar路径

  –dryrun,-d               不去真实启动flume agent，而是打印当前命令

  –plugins-path <dirs>       指定插件（jar）所在路径

  -Dproperty=value          传入java环境参数

  -Xproperty=value          传入所需的JVM配置参数

agent options:

  –name,-n <name>          agent的别名（在用户采集方案配置文件中）

  –conf-file,-f <file>          指定用户采集方案配置文件的路径

  –zkConnString,-z <str>      指定zookeeper的连接地址

  –zkBasePath,-p <path>      指定用户配置文件所在的zookeeper path，比如：/flume/config

  –no-reload-conf            关闭配置文件动态加载

  –help,-h                   display help text

avro-client options:

  –rpcProps,-P <file>   RPC client properties file with server connection params

  –host,-H <host>     avro序列化数据所要发往的目标主机（avro source所在机器）

  –port,-p <port>      avro序列化数据所要发往的目标主机的端口号

  –dirname <dir>      需要被序列化发走的数据所在目录（提前准备好测试数据放在一个文件中）

  –filename,-F <file>   需要被序列化发走的数据所在文件(default: std input)

  –headerFile,-R <file>  存储header key-value的文件

  –help,-h             帮助信息

  Either –rpcProps or both –host and –port must be specified.

Note that if <conf> directory is specified, then it is always included first

in the classpath.

# 定义这个agent中各组件的名字

a1.sources = r1

a1.sinks = k1

a1.channels = c1

# 描述和配置source组件：r1

a1.sources.r1.type = netcat

a1.sources.r1.bind = 0.0.0.0

a1.sources.r1.port = 9999

# source 和 channel关联

a1.sources.r1.channels = c1  

# 描述和配置sink组件：k1

a1.sinks.k1.type = logger

# sink也要关联channel

a1.sinks.k1.channel = c1

# 描述和配置channel组件，此处使用是内存缓存的方式

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

[root@hdp-01 apache-flume-1.6.0-bin]# bin/flume-ng agent -n a1 -c conf -f myconf/netcat-logger.conf    -Dflume.root.logger=INFO,console

nohup bin/flume-ng agent -n a1 -c conf -f myconf/netcat-logger.conf 1>/dev/null 2>&1 &

a1.sources = s1

a1.sources.s1.type = netcat

a1.sources.s1.bind = 0.0.0.0

a1.sources.s1.port = 44444

a1.sources.s1.channels = c1

channels

–

本source要发往的channel

type

–

本source的类别名称：exec

command

–

本source所要运行的linux命令,比如： tail -F /path/file

shell

–

指定运行上述命令所用shell

restartThrottle

10000

命令die了以后，重启的时间间隔

restart

false

命令die了以后，是否要重启

logStdErr

false

是否收集命令的错误输出stderr

batchSize

20

提交的event批次大小

batchTimeout

3000

发往下游没完成前，等待的时间

selector.type

replicating

指定channel选择器：replicating or multiplexing

selector.*

选择器的具体参数

interceptors

–

指定拦截器

interceptors.*

 指定的拦截器的具体参数

a1.sources = s1

a1.sources.s1.channels = c1

a1.sources.s1.type = exec

a1.sources.s1.command = tail -F /root/weblog/access.log

a1.sources.s1.batchSize = 100

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

for i in {1..10000}; do echo ${i}————————— >> access.log ; sleep 0.5; done

Property Name

Default

Description

channels

–

type

–

The component type name, needs to be spooldir.

spoolDir

–

The directory from which to read files from.

fileSuffix

.COMPLETED

采集完成的文件，添加什么后缀名

deletePolicy

never

是否删除采完的文件: never or immediate

fileHeader

false

是否将所采集文件的绝对路径添加到header中

fileHeaderKey

file

上述header的key名称

basenameHeader

false

是否将文件名添加到header

basenameHeaderKey

basename

上述header的key名称

includePattern

^.*$

指定需要采集的文件名的正则表达式

ignorePattern

^$

指定要排除的文件名的正则表达式

如果一个文件名即符合includePattern又匹配ignorePattern，则该文件不采

trackerDir

.flumespool

记录元数据的目录所在路径，可以用绝对路径也可以用相对路径（相对于采集目录）

trackingPolicy

rename

采集进度跟踪策略，有两种： “rename”和 “tracker_dir”. 本参数只在deletePolicy=never时才生效

 “rename”– 采完的文件根据filesuffix重命名

 “tracker_dir” – 采完的文件会在trackerDir目录中生成一个同名的空文件

consumeOrder

oldest

采集顺序： oldest, youngest and random.

oldest和youngest情况下，可能会带来一定效率的损失；（需要对文件夹中所有文件进行一次扫描以寻找最old或最young的）

pollDelay

500

Delay (in milliseconds) used when polling for new files.

recursiveDirectorySearch

false

Whether to monitor sub directories for new files to read.

maxBackoff

4000

The maximum time (in millis) to wait between consecutive attempts to write to the channel(s) if the channel is full. The source will start at a low backoff and increase it exponentially each time the channel throws a ChannelException, upto the value specified by this parameter.

batchSize

100

一次传输到channel的event条数（一批）

inputCharset

UTF-8

Character set used by deserializers that treat the input file as text.

decodeErrorPolicy

FAIL

What to do when we see a non-decodable character in the input file. FAIL: Throw an exception and fail to parse the file. REPLACE: Replace the unparseable character with the “replacement character” char, typically Unicode U+FFFD. IGNORE: Drop the unparseable character sequence.

deserializer

LINE

Specify the deserializer used to parse the file into events. Defaults to parsing each line as an event. The class specified must implementEventDeserializer.Builder.

deserializer.*

Varies per event deserializer.

bufferMaxLines

–

(Obselete) This option is now ignored.

bufferMaxLineLength

5000

(Deprecated) Maximum length of a line in the commit buffer. Use deserializer.maxLineLength instead.

selector.type

replicating

replicating or multiplexing

selector.*

Depends on the selector.type value

interceptors

–

Space-separated list of interceptors

interceptors.*

a1.sources = s1

a1.sources.s1.channels = c1

a1.sources.s1.type = spooldir

a1.sources.s1.spoolDir = /root/weblog

a1.sources.s1.batchSize = 200

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

bin/flume-ng agent -n a1 -c conf -f myconf/spooldir-mem-logger.conf -Dflume.root.logger=DEBUG,console

Property Name

Default

Description

channels

–

type

–

本source的别名： avro

bind

–

要绑定的地址

port

–

要绑定的端口号

threads

–

服务的最大线程数

selector.type

selector.*

interceptors

–

Space-separated list of interceptors

interceptors.*

compression-type

none

压缩类型：跟发过来的数据是否压缩要匹配：none | deflate

ssl

false

Set this to true to enable SSL encryption. If SSL is enabled, you must also specify a “keystore” and a “keystore-password”, either through component level parameters (see below) or as global SSL parameters (see SSL/TLS support section).

keystore

–

This is the path to a Java keystore file. If not specified here, then the global keystore will be used (if defined, otherwise configuration error).

keystore-password

–

The password for the Java keystore. If not specified here, then the global keystore password will be used (if defined, otherwise configuration error).

keystore-type

JKS

The type of the Java keystore. This can be “JKS” or “PKCS12”. If not specified here, then the global keystore type will be used (if defined, otherwise the default is JKS).

exclude-protocols

SSLv3

Space-separated list of SSL/TLS protocols to exclude. SSLv3 will always be excluded in addition to the protocols specified.

include-protocols

–

Space-separated list of SSL/TLS protocols to include. The enabled protocols will be the included protocols without the excluded protocols. If included-protocols is empty, it includes every supported protocols.

exclude-cipher-suites

–

Space-separated list of cipher suites to exclude.

include-cipher-suites

–

Space-separated list of cipher suites to include. The enabled cipher suites will be the included cipher suites without the excluded cipher suites. If included-cipher-suites is empty, it includes every supported cipher suites.

ipFilter

false

Set this to true to enable ipFiltering for netty

ipFilterRules

–

Define N netty ipFilter pattern rules with this config.

a1.sources = r1

a1.sources.r1.type = avro

a1.sources.r1.channels = c1

a1.sources.r1.bind = 0.0.0.0

a1.sources.r1.port = 4141

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

bin/flume-ng agent -c ./conf -f ./myconf/avro-mem-logger.conf -n a1 -Dflume.root.logger=DEBUG,consol

bin/flume-ng avro-client –host c703 –port 4141

a1.sources = r1

a1.channels = c1

a1.sinks = k1

a1.sources.r1.channels = c1

a1.sources.r1.type = exec

a1.sources.r1.command = tail -F /tmp/logs/access.log

a1.sources.r1.batchSize = 100

a1.channels.c1.type = memory

a1.channels.c1.capacity = 10000

a1.channels.c1.trasactionCapacity = 100

a1.sinks.k1.channel = c1

a1.sinks.k1.type = avro

a1.sinks.k1.hostname = h3

a1.sinks.k1.port = 4455

a1.sources = r1

a1.channels = c1

a1.sinks = k1

a1.sources.r1.channels = c1

a1.sources.r1.type = avro

a1.sources.r1.bind = 0.0.0.0

a1.sources.r1.port = 4455

a1.channels.c1.type = memory

a1.channels.c1.capacity = 10000

a1.channels.c1.trasactionCapacity = 100

a1.sinks.k1.channel = c1

a1.sinks.k1.type = logger

while true

do

echo "hello "  >> /tmp/logs/access.log

sleep 0.1

done

Property Name

Default

Description

channels

–

 数据发往的channel

type

–

本source的名称：

org.apache.flume.source.kafka.KafkaSource

kafka.bootstrap.servers

–

Kafka broker服务器列表，逗号分隔

kafka.consumer.group.id

flume

Kafka消费者组id

kafka.topics

–

Kafka消息主题列表，逗号隔开

kafka.topics.regex

–

用正则表达式来指定一批topic；本参数的优先级高于kafka.topics

batchSize

1000

写入channel的event 批，最大消息条数

batchDurationMillis

1000

批次写入channel的最大时长

backoffSleepIncrement

1000

Kafka Topic 显示为空时触发的初始和增量等待时间。

maxBackoffSleep

5000

Kafka Topic 显示为空时触发的最长等待时间

useFlumeEventFormat

false

默认情况下，event 将从Kafka Topic 直接作为字节直接进入event 主体。设置为true以读取event 作为Flume Avro二进制格式。与Kafka Sink上的相同属性或Kafka Channel上的parseAsFlumeEvent属性一起使用时，这将保留在生成端发送的任何Flume标头。

setTopicHeader

true

是否要往header中加入一个kv：topic信息

topicHeader

topic

应上面开关的需求，加入kv：topic =>topic名称

kafka.consumer.security.protocol

PLAINTEXT

Set to SASL_PLAINTEXT, SASL_SSL or SSL if writing to Kafka using some level of security. See below for additional info on secure setup.

more consumer security props

If using SASL_PLAINTEXT, SASL_SSL or SSL refer to Kafka security for additional properties that need to be set on consumer.

Other Kafka Consumer Properties

–

本source，允许直接配置任意的kafka消费者参数，格式如下：

For example: kafka.consumer.auto.offset.reset

（就是在消费者参数前加统一前缀： kafka.consumer.）

a1.sources = s1

a1.sources.s1.type = org.apache.flume.source.kafka.KafkaSource

a1.sources.s1.channels = c1

a1.sources.s1.batchSize = 100

a1.sources.s1.batchDurationMillis = 2000

a1.sources.s1.kafka.bootstrap.servers = c701:9092,c702:9092,c703:9092

a1.sources.s1.kafka.topics = TAOGE

a1.sources.s1.kafka.consumer.group.id = g1

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

Property Name

Default

Description

channels

–

所要写往的channel

type

–

本source的别名： TAILDIR.

filegroups

–

空格分割的组名，每一组代表着一批文件

g1 g2

filegroups.<filegroupName>

–

每个文件组的绝路路径，文件名可用正则表达式

positionFile

~/.flume/taildir_position.json

记录偏移量位置的文件所在路径

headers.<filegroupName>.<headerKey>

–

Header value which is the set with header key. Multiple headers can be specified for one file group.

byteOffsetHeader

false

Whether to add the byte offset of a tailed line to a header called ‘byteoffset’.

skipToEnd

false

Whether to skip the position to EOF in the case of files not written on the position file.

idleTimeout

120000

关闭非活动文件的时延。如果被关闭的这个文件又在某个时间有了新增行,会被此source检测到，并重新打开

writePosInterval

3000

3s 记录一次偏移量到positionfile

batchSize

100

提交event到channel的批次最大条数

maxBatchCount

Long.MAX_VALUE

控制在一个文件上连续读取的最大批次个数（如果某个文件正在被高速写入，那就应该让这个参数调为最大值，以让source可以集中精力专采这个文件）

backoffSleepIncrement

1000

The increment for time delay before reattempting to poll for new data, when the last attempt did not find any new data.

maxBackoffSleep

5000

The max time delay between each reattempt to poll for new data, when the last attempt did not find any new data.

cachePatternMatching

true

Listing directories and applying the filename regex pattern may be time consuming for directories containing thousands of files. Caching the list of matching files can improve performance. The order in which files are consumed will also be cached. Requires that the file system keeps track of modification times with at least a 1-second granularity.

fileHeader

false

Whether to add a header storing the absolute path filename.

fileHeaderKey

file

Header key to use when appending absolute path filename to event header.

a1.sources = r1

a1.sources.r1.type = TAILDIR

a1.sources.r1.channels = c1

a1.sources.r1.positionFile = /root/flumedata/taildir_position.json

a1.sources.r1.filegroups = f1

a1.sources.r1.filegroups.f1 = /root/weblog/access.log

a1.sources.r1.fileHeader = true

a1.sources.ri.maxBatchCount = 1000

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

bin/flume-ng agent -n a1 -c conf/ -f myconf/taildir-mem-logger.conf -Dflume.root.logger=DEBUG,console

Property Name

Default

Description

type

–

本拦截器的名称：timestamp

headerName

timestamp

要插入header的key名

preserveExisting

false

如果header中已存在同名key，是否要覆盖

a1.sources = s1

a1.sources.s1.channels = c1

a1.sources.s1.type = exec

a1.sources.s1.command = tail -F /root/weblog/access.log

a1.sources.s1.batchSize = 100

a1.sources.s1.interceptors = i1

a1.sources.s1.interceptors.i1.type = timestamp

a1.sources.s1.interceptors.i1.preserveExisting = false

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

2019-06-09 10:24:21,884 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO – org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{timestamp=1560047061012} body: 31 30 32 34 34 2E 2E 2E 2E 2E 2E 2E 2E 2E 2E 2E 10244……….. }

Property Name

Default

Description

type

–

别名： static

preserveExisting

true

是否覆盖同名kv

key

key

你要插入的key名

value

value

你要插入的value

a1.sources = s1

a1.sources.s1.channels = c1

a1.sources.s1.type = exec

a1.sources.s1.command = tail -F /root/weblog/access.log

a1.sources.s1.batchSize = 100

a1.sources.s1.interceptors = i1 i2 i3

a1.sources.s1.interceptors.i1.type = timestamp

a1.sources.s1.interceptors.i1.preserveExisting = false

a1.sources.s1.interceptors.i2.type = host

a1.sources.s1.interceptors.i2.preserveExisting = false

a1.sources.s1.interceptors.i2.useIP = true

a1.sources.r1.interceptors.i3.type = static

a1.sources.r1.interceptors.i3.key = hero

a1.sources.r1.interceptors.i3.value = TAOGE

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

Property Name

Default

Description

type

–

本拦截器的别名： host

preserveExisting

false

是否覆盖已存在的hader key-value

useIP

true

插入ip还是主机名

hostHeader

host

要插入header的key名

a1.sources = s1

a1.sources.s1.channels = c1

a1.sources.s1.type = exec

a1.sources.s1.command = tail -F /root/weblog/access.log

a1.sources.s1.batchSize = 100

a1.sources.s1.interceptors = i1 i2

a1.sources.s1.interceptors.i1.type = timestamp

a1.sources.s1.interceptors.i1.preserveExisting = false

a1.sources.s1.interceptors.i2.type = host

a1.sources.s1.interceptors.i2.preserveExisting = false

a1.sources.s1.interceptors.i2.useIP = true

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

Property Name

Default

Description

type

–

全名：org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder

headerName

id

Key名称

preserveExisting

true

是否覆盖同名key

prefix

""

Uuid前的前缀

a1.sources = s1

a1.sources.s1.channels = c1

a1.sources.s1.type = exec

a1.sources.s1.command = tail -F /root/weblog/access.log

a1.sources.s1.batchSize = 100

a1.sources.s1.interceptors = i1 i2 i3 i4

a1.sources.s1.interceptors.i1.type = timestamp

a1.sources.s1.interceptors.i1.preserveExisting = false

a1.sources.s1.interceptors.i2.type = host

a1.sources.s1.interceptors.i2.preserveExisting = false

a1.sources.s1.interceptors.i2.useIP = true

a1.sources.s1.interceptors.i3.type = static

a1.sources.s1.interceptors.i3.key = hero

a1.sources.s1.interceptors.i3.value = TAOGE

a1.sources.s1.interceptors.i4.type = org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder

a1.sources.s1.interceptors.i4.headName = duanzong

a1.sources.s1.interceptors.i4.prefix =  666_

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

Property Name

Default

Description

type

–

别名： memory

capacity

100

能存储的最大事件event数

transactionCapacity

100

最大事务控制容量

keep-alive

3

添加或移除event的超时时间

byteCapacityBufferPercentage

20

除了body以外的字节所能占用的容量百分比

byteCapacity

see description

channel中最大的总byte数（只计算body）

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

// lock to guard queue, mainly needed to keep it locked down during resizes// it should never be held through a blocking operation

private Object queueLock = new Object();

//queue为Memory Channel中存放Event的地方，这里用了LinkedBlockingDeque来实现

@GuardedBy(value = "queueLock")p

rivate LinkedBlockingDeque<Event> queue;

//下面的两个信号量用来做同步操作，queueRemaining表示queue中的剩余空间，queueStored表示queue中的使用空间

// invariant that tracks the amount of space remaining in the queue(with all uncommitted takeLists deducted)

// we maintain the remaining permits = queue.remaining – takeList.size()// this allows local threads waiting for space in the queue to commit without denying access to the

// shared lock to threads that would make more space on the queue

private Semaphore queueRemaining;

// used to make "reservations" to grab data from the queue.

// by using this we can block for a while to get data without locking all other threads out

// like we would if we tried to use a blocking call on queue

private Semaphore queueStored;

//下面几个变量为配置文件中Memory Channel的配置项

// 一个事务中Event的最大数目

private volatile Integer transCapacity;

// 向queue中添加、移除Event的等待时间

private volatile int keepAlive;

// queue中，所有Event所能占用的最大空间

private volatile int byteCapacity;

private volatile int lastByteCapacity;

// queue中，所有Event的header所能占用的最大空间占byteCapacity的比例

private volatile int byteCapacityBufferPercentage;

// 用于标示byteCapacity中剩余空间的信号量

private Semaphore bytesRemaining;

// 用于记录Memory Channel的一些指标，后面可以通过配置监控来观察Flume的运行情况

private ChannelCounter channelCounter;

Property Name Default

Description

type

–

别名： file.

checkpointDir

~/.flume/file-channel/checkpoint

Checkpoint信息保存目录

useDualCheckpoints

false

Checkpoint是否双重checkpoint机制

backupCheckpointDir

–

备份checkpoint的保存目录

dataDirs

~/.flume/file-channel/data

Event数据缓存目录

transactionCapacity

10000

事务管理容量

checkpointInterval

30000

记录checkpoint信息的时间间隔

maxFileSize

2146435071

控制一个数据文件的大小规格

minimumRequiredSpace

524288000

所需的最低磁盘空间，低于则停止接收新数据

capacity

1000000

最大event缓存数

keep-alive

3

等待添加数据的最大时间

a1.sources = r1

a1.sources.r1.type = TAILDIR

a1.sources.r1.channels = c1

a1.sources.r1.positionFile = /root/taildir_chkp/taildir_position.json

a1.sources.r1.filegroups = f1

a1.sources.r1.filegroups.f1 = /root/weblog/access.log

a1.sources.r1.fileHeader = true

a1.sources.ri.maxBatchCount = 1000

a1.channels = c1

a1.channels.c1.type = file

a1.channels.c1.capacity = 1000000

a1.channels.c1.transactionCapacity = 100

a1.channels.c1.checkpointDir = /root/flume_chkp

a1.channels.c1.dataDirs = /root/flume_data

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

Property Name

Default

Description

type

–

名字： org.apache.flume.channel.kafka.KafkaChannel

kafka.bootstrap.servers

–

Kafka服务器地址

kafka.topic

flume-channel

所使用的topic

kafka.consumer.group.id

flume

消费者组id

parseAsFlumeEvent

true

跟上、下游匹配，是否需要将数据解析为Flume的Event格式

pollTimeout

500

从kafka取数据的超时时间

defaultPartitionId

–

默认指派的partitionid

partitionIdHeader

–

将一个event指派到某个分区时所使用的header 的key

kafka.consumer.auto.offset.reset

latest

初始化读取偏移量的策略

a1.sources = r1

# 配置两个channel,为了便于观察

# c1 是kafkachannel ,c2是一个内存channel

a1.channels = c1 c2

a1.sinks = k1

a1.sources.r1.type = exec

a1.sources.r1.command = tail -F /root/logs/a.log

a1.sources.r1.channels = c1 c2

# kafka-channel具体配置，该channel没有sink

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel

a1.channels.c1.kafka.bootstrap.servers = h1:9092,h2:9092,h3:9092

a1.channels.c1.parseAsFlumeEvent = false

# 内存channel 配置,并对接一个logger sink来观察

a1.channels.c2.type = memory

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c2

a1.sources = r1

a1.sinks = k1

a1.channels = c1

a1.sources.r1.type = exec

a1.sources.r1.command = tail -F /root/logs/a.log

a1.sources.r1.channels = c1

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel

a1.channels.c1.kafka.bootstrap.servers = h1:9092,h2:9092,h3:9092

a1.channels.c1.parseAsFlumeEvent = false

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

Name

Default

Description

channel

–

从哪个channel取数据

type

–

别名： hdfs

hdfs.path

–

目标hdfs存储路径（URI）

hdfs.filePrefix

FlumeData

指定生成的文件名前缀

hdfs.fileSuffix

–

后缀

hdfs.inUsePrefix

–

正在写入的文件的前缀标识

hdfs.inUseSuffix

.tmp

正在写入的文件的后缀标识

hdfs.rollInterval

30

切换文件的条件：间隔时间；为0则不生效

hdfs.rollSize

134217728

切换文件的条件：文件大小；为0则不生效

hdfs.rollCount

10

切换文件的条件：event条数；为0则不生效

hdfs.idleTimeout

0

不活跃文件的关闭超时时长；0则不自动关闭

hdfs.batchSize

100

从channel中取一批数据的最大大小；

hdfs.codeC

–

压缩编码: gzip, bzip2, lzo, lzop, snappy

hdfs.fileType

SequenceFile

目标文件格式: SequenceFile, DataStream or CompressedStream 

注意：DataStream 不能支持压缩

CompressedStream 必须设置压缩编码

SequenceFile 可压缩可不压缩

hdfs.maxOpenFiles

5000

允许同时最多打开的文件数；如果超出，则会关闭最早打开的

hdfs.minBlockReplicas

–

目标文件的block副本数

hdfs.writeFormat

Writable

指定sequence file中的对象类型；支持Text和Writable

同时请使用Text，否则后续数据处理平台可能无法解析

hdfs.threadsPoolSize

10

操作HDFS时的线程池大小

hdfs.rollTimerPoolSize

1

检查文件是否需要被roll的线程数

hdfs.kerberosPrincipal

–

Kerberos user principal for accessing secure HDFS

hdfs.kerberosKeytab

–

Kerberos keytab for accessing secure HDFS

hdfs.proxyUser

hdfs.round

false

目录通配符切换是是否需要切掉尾数

hdfs.roundValue

10

时间尾数切掉多少

hdfs.roundUnit

minute

时间尾数切掉大小的单位– second, minute or hour.

hdfs.timeZone

Local Time

时间通配符所使用的时区

hdfs.useLocalTimeStamp

false

所用的时间是否要从agent sink本地获取

hdfs.closeTries

0

重命名已完成文件的重试次数；0则一直尝试重命名

hdfs.retryInterval

180

关闭一个文件的重试时间间隔

serializer

TEXT

将channel中的event body解析成什么格式：Text| avro_event ； 也可以使用自定义的序列化器

serializer.*

小提示：什么叫做URI

## 定义

a1.sources = r1

a1.sinks = k1

a1.channels = c1

## source

a1.sources.r1.type = exec

a1.sources.r1.command = tail -F /root/logs/a.log

a1.sources.r1.channels = c1

a1.sources.r1.interceptors = i1

a1.sources.r1.interceptors.i1.type = timestamp

## channel

a1.channels.c1.type = memory

a1.channels.c1.capacity = 100000000

## sink

a1.sinks.k1.channel = c1

a1.sinks.k1.type = hdfs

a1.sinks.k1.hdfs.path = hdfs://h1:8020/doitedu/%Y-%m-%d/%H-%M

a1.sinks.k1.hdfs.round = true

a1.sinks.k1.hdfs.roundValue = 10

a1.sinks.k1.hdfs.roundUnit = minute

a1.sinks.k1.hdfs.filePrefix = doit_

a1.sinks.k1.hdfs.fileSuffix = .log.gz

a1.sinks.k1.hdfs.rollInterval = 0

a1.sinks.k1.hdfs.rollSize = 102400

a1.sinks.k1.hdfs.rollCount = 0

a1.sinks.k1.hdfs.fileType = CompressedStream

a1.sinks.k1.hdfs.codeC = gzip

a1.sinks.k1.hdfs.writeFormat = Text

Property Name

Default

Description

type

–

名称： org.apache.flume.sink.kafka.KafkaSink

kafka.bootstrap.servers

–

Kafka服务器列表

kafka.topic

default-flume-topic

Kafka的topic

flumeBatchSize

100

从channel中取event的批次大小

kafka.producer.acks

1

Kafka生产者消息推送应答级别：

1 : Leader接收到即回应

0 ：不等回应 

–1：副本同步完成，再回应

useFlumeEventFormat

false

是否使用avro序列化

defaultPartitionId

–

指定所有event默认发往的分区id；不指定则按kafka生产者的分区器，均匀分发

partitionIdHeader

–

通过在header中指定分区id，来约束这个event发往的分区

allowTopicOverride

true

允许在event的header中指定要写入的topic

topicHeader

topic

如果上一条开关开启，则在header中放入的key的名称

Other Kafka Producer Properties

–

可以用kafka.producer.xxx来配置任何kafka producer的参数

a1.sources = r1

a1.sources.r1.type = TAILDIR

a1.sources.r1.channels = c1

a1.sources.r1.positionFile = /root/taildir_chkp/taildir_position.json

a1.sources.r1.filegroups = f1

a1.sources.r1.filegroups.f1 = /root/weblog/access.log

a1.sources.r1.fileHeader = true

a1.sources.r1.maxBatchCount = 1000

a1.sources.r1.interceptors = i1

a1.sources.r1.interceptors.i1.type = timestamp

a1.channels = c1

a1.channels.c1.type = file

a1.channels.c1.capacity = 1000000

a1.channels.c1.transactionCapacity = 100

a1.channels.c1.checkpointDir = /root/flume_chkp

a1.channels.c1.dataDirs = /root/flume_data

a1.sinks = k1

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 = c701:9092,c702:9092,c703: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

Property Name

Default

Description

channel

–

type

–

The component type name, needs to be avro.

hostname

–

目标avro source的主机

port

–

目标avro source的绑定端口

batch-size

100

number of event to batch together for send.

connect-timeout

20000

连接超时时间

request-timeout

20000

请求超时时间

reset-connection-interval

none

Amount of time (s) before the connection to the next hop is reset. This will force the Avro Sink to reconnect to the next hop. This will allow the sink to connect to hosts behind a hardware load-balancer when news hosts are added without having to restart the agent.

compression-type

none

This can be “none” or “deflate”. The compression-type must match the compression-type of matching AvroSource

compression-level

6

The level of compression to compress event. 0 = no compression and 1-9 is compression. The higher the number the more compression

ssl

false

Set to true to enable SSL for this AvroSink. When configuring SSL, you can optionally set a “truststore”, “truststore-password”, “truststore-type”, and specify whether to “trust-all-certs”.

trust-all-certs

false

If this is set to true, SSL server certificates for remote servers (Avro Sources) will not be checked. This should NOT be used in production because it makes it easier for an attacker to execute a man-in-the-middle attack and “listen in” on the encrypted connection.

truststore

–

The path to a custom Java truststore file. Flume uses the certificate authority information in this file to determine whether the remote Avro Source’s SSL authentication credentials should be trusted. If not specified, then the global keystore will be used. If the global keystore not specified either, then the default Java JSSE certificate authority files (typically “jssecacerts” or “cacerts” in the Oracle JRE) will be used.

truststore-password

–

The password for the truststore. If not specified, then the global keystore password will be used (if defined).

truststore-type

JKS

The type of the Java truststore. This can be “JKS” or other supported Java truststore type. If not specified, then the global keystore type will be used (if defined, otherwise the defautl is JKS).

exclude-protocols

SSLv3

Space-separated list of SSL/TLS protocols to exclude. SSLv3 will always be excluded in addition to the protocols specified.

maxIoWorkers

2 * the number of available processors in the machine

The maximum number of I/O worker threads. This is configured on the NettyAvroRpcClient NioClientSocketChannelFactory.

## c703 ##

a1.sources = s1

a1.sources.s1.type = exec

a1.sources.s1.command = tail -F /root/weblog/access.log

a1.sources.s1.channels = c1

a1.channels = c1

a1.channels.c1.type = memory

a1.sinks = k1

a1.sinks.k1.type = avro

a1.sinks.k1.channel = c1

a1.sinks.k1.hostname = c701

a1.sinks.k1.port = 4545

## c701 ##

a1.sources = s1

a1.sources.s1.type = avro

a1.sources.s1.hostname = 0.0.0.0

a1.sources.s1.port = 4545

a1.sources.s1.channel = c1

a1.channels = c1

a1.channels.c1.type = memory

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

sourceRunner 拿到 source实例对象

然后调 source的start方法

source读数据，然后将数据转成event

然后将event传给channelprocessor.processEvent(event)

processEvent中第一个动作就是调用拦截器拦截这个event，然后再往channel中写入

package cn.doitedu.flume.custom;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.SystemClock;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.apache.flume.source.ExecSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 能够记录读取位置偏移量的自定义source
 */
public class HoldOffesetSource extends AbstractSource implements EventDrivenSource, Configurable {
    private static final Logger logger = LoggerFactory.getLogger(HoldOffesetSource.class);

    private String positionfilepath;
    private String logfile;
    private int batchsize;

    private ExecutorService exec;

    /**
     * 框架调用本方法，开始采集数据
     * 自定义代码去读取数据，转为event
     * 用getChannelProcessor（）方法（定义在父类中）去获取框架的channelprocessor（channel处理器）
     * 调用这个channelprocessor将event提交给channel
     */
    @Override
    public synchronized void start() {

        super.start();
        // 用于向channel提交数据的一个处理器
        ChannelProcessor channelProcessor = getChannelProcessor();

        // 获取历史偏移量
        long offset = 0;
        try {
            File positionfile = new File(this.positionfilepath);
            String s = FileUtils.readFileToString(positionfile);
            offset = Long.parseLong(s);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 构造一个线程池
        exec = Executors.newSingleThreadExecutor();
        // 向线程池提交数据采集任务
        exec.execute(new HoldOffsetRunnable(offset, logfile, channelProcessor, batchsize, positionfilepath));

    }

    /**
     * 停止前要调用的方法
     * 可以在这里做一些资源关闭清理工作
     */
    @Override
    public synchronized void stop() {
        super.stop();

       try{
           exec.shutdown();
       }catch (Exception e){
           exec.shutdown();
       }
    }

    /**
     * 获取配置文件中的参数，来配置本source实例
     * <p>
     * 要哪些参数：
     * 偏移量记录文件所在路径
     * 要采集的文件所在路径
     *
     * @param context
     */
    public void configure(Context context) {

        // 这是我们source用来记录偏移量的文件路径
        this.positionfilepath = context.getString("positionfile", "./");

        // 这是我们source要采集的日志文件的路径
        this.logfile = context.getString("logfile");

        // 这是用户配置的采集事务批次最大值
        this.batchsize = context.getInteger("batchsize", 100);

        // 如果日志文件路径没有指定，则抛异常
        if (StringUtils.isBlank(logfile)) throw new RuntimeException("请配置需要采集的文件路径");

    }

    /**
     * 采集文件的具体工作线程任务类
     */
    private static class HoldOffsetRunnable implements Runnable {

        long offset;
        String logfilepath;
        String positionfilepath;
        ChannelProcessor channelProcessor;  // channel提交器 （里面会调拦截器，会开启写入channel的事务）
        int batchsize; // 批次大小
        List<Event> events = new ArrayList<Event>();  // 用来保存一批事件
        SystemClock systemClock = new SystemClock();

        public HoldOffsetRunnable(long offset, String logfilepath, ChannelProcessor channelProcessor, int batchsize, String positionfilepath) {
            this.offset = offset;
            this.logfilepath = logfilepath;
            this.channelProcessor = channelProcessor;
            this.batchsize = batchsize;
            this.positionfilepath = positionfilepath;
        }

        public void run() {

            try {
                // 先定位到指定的offset
                RandomAccessFile raf = new RandomAccessFile(logfilepath, "r");
                raf.seek(offset);

                // 循环读数据
                String line = null;

                // 记录上一批提交的时间
                long lastBatchTime = System.currentTimeMillis();
                while (true) {
                    line = raf.readLine();
                    if(line == null ){
                        Thread.sleep(2000);
                        continue;
                    }

                    // 将数据转成event
                    Event event = EventBuilder.withBody(line.getBytes());
                    // 装入list batch
                    synchronized (HoldOffesetSource.class) {
                        events.add(event);
                    }

                    // 判断批次大小是否满 或者 时间到了没有
                    if (events.size() >= batchsize || timeout(lastBatchTime)) {
                        // 满足，则提交
                        channelProcessor.processEventBatch(events);

                        // 记录提交时间
                        lastBatchTime = systemClock.currentTimeMillis();

                        // 记录偏移量
                        long offset = raf.getFilePointer();
                        FileUtils.writeStringToFile(new File(positionfilepath), offset + "");

                        // 清空本批event
                        events.clear();

                    }

                    // 不满足，继续读
                }
            } catch (FileNotFoundException e) {
                logger.error("要采集的文件不存在");
            } catch (IOException e) {
                logger.error("我也不知道怎么搞的，不好意思，我罢工了");
            } catch (InterruptedException e) {
                logger.error("线程休眠出问题了");
            }
        }

        // 判断是否批次间隔超时
        private boolean timeout(long lastBatchTime) {
            return systemClock.currentTimeMillis() – lastBatchTime > 2000;
        }
    }
}

a1.sources = s1

a1.channels = c1

a1.sinks = k1

# source 配置

a1.sources.s1.type = cn.doitedu.flume.custom.HoldOffesetSource

a1.sources.s1.position_file_path= /root/myposition

a1.sources.s1.data_file_path = /root/weblog/access.log

a1.sources.s1.batchSize = 100

a1.sources.s1.batchTime = 5000

a1.sources.s1.channels = c1

a1.channels.c1.type = memory

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

package cn.doitedu.flume.custom;

import org.apache.commons.codec.binary.Base64;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;

public class EncryptInterceptor  implements Interceptor {
    // 要加密的字段索引s
    String indices;
    // 索引之间的分隔符
    String idxSplitBy;
    // 数据体字段之间的分隔符
    String dataSplitBy;

    /**
     * 构造方法
     * @param indices
     * @param idxSplitBy
     * @param dataSplitBy
     */
    public EncryptInterceptor(String indices, String idxSplitBy, String dataSplitBy) {
        // 0,3
        this.indices = indices;
        this.idxSplitBy = idxSplitBy;
        this.dataSplitBy = dataSplitBy;
    }

    // 这个方法会被框架调用一次，用来做一些初始化工作
    public void initialize() {

    }

    // 拦截方法—对一个event进行处理
    public Event intercept(Event event) {

        byte[] body = event.getBody();
        String dataStr = new String(body);

        // 数据的字段数组
        String[] dataFieldsArr = dataStr.split(dataSplitBy);

        // 需要加密的索引的数组
        String[] idxArr = indices.split(idxSplitBy);

        for (String s : idxArr) {

            int index = Integer.parseInt(s);
            // 取出要加密的字段的内容
            String field = dataFieldsArr[index];
            // MD5加密这个字段
            String encryptedField = DigestUtils.md5Hex(field);
            // BASE64编码
            byte[] bytes = Base64.decodeBase64(encryptedField);
            // 替换掉原来的未加密内容
            dataFieldsArr[index] = new String(bytes);
        }

        // 将加密过的字段重新拼接成一条数据，并使用原来的分隔符
        StringBuilder sb = new StringBuilder();
        for (String field : dataFieldsArr) {
            sb.append(field).append(dataSplitBy);
        }

        sb.deleteCharAt(sb.lastIndexOf(dataSplitBy));

        // 返回加密后的字段所封装的event对象
        return EventBuilder.withBody(sb.toString().getBytes());
    }

    // 拦截方法—对一批event进行处理
    public List<Event> intercept(List<Event> events) {

        ArrayList<Event> lst = new ArrayList<Event>();

        for (Event event : events) {
            Event eventEncrpt = intercept(event);
            lst.add(eventEncrpt);
        }

        return lst;
    }

    // agent退出前，会调一次该方法，进行需要的清理、关闭操作
    public void close() {

    }

    /**
     * 拦截器的构造器
     */
    public staticc lass EncryptInterceptorBuilder implements Interceptor.Builder{
        // 要加密的字段索引s
        String indices;
        // 索引之间的分隔符
        String idxSplitBy;
        // 数据体字段之间的分隔符
        String dataSplitBy;

        // 构造一个拦截器实例
        public Interceptor build() {

            return new EncryptInterceptor(indices,idxSplitBy,dataSplitBy);
        }

        // 获取配置文件中的拦截器参数
        public void configure(Context context) {
            // 要加密的字段索引s
            this.indices = context.getString(Constants.INDICES);
            // 索引之间的分隔符
            this.idxSplitBy = context.getString(Constants.IDX_SPLIT_BY);
            // 数据体字段之间的分隔符
            this.dataSplitBy = context.getString(Constants.DATA_SPLIT_BY);

        }
    }

    public static class Constants {
        public static final String INDICES = "indices";
        public static final String IDX_SPLIT_BY = "idxSplitBy";
        public static final String DATA_SPLIT_BY = "dataSplitBy";
    }
}

a1.sources = s1

a1.sources.s1.channels = c1

a1.sources.s1.type = exec

a1.sources.s1.command = tail -F /root/weblog/access.log

a1.sources.s1.batchSize = 100

a1.sources.s1.interceptors = i1

a1.sources.s1.interceptors.i1.type = cn.doitedu.flume.custom.EncryptInterceptor$EncryptInterceptorBuilder

a1.sources.s1.interceptors.i1.indices = 0:4

a1.sources.s1.interceptors.i1.idxSplitBy = :

a1.sources.s1.interceptors.i1.dataSplitBy = ,

a1.channels = c1

a1.channels.c1.type = memory

a1.channels.c1.capacity = 200

a1.channels.c1.transactionCapacity = 100

a1.sinks = k1

a1.sinks.k1.type = logger

a1.sinks.k1.channel = c1

# 写到kafka就用这一段sink配置

# a1.sinks = k1

# a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink

# a1.sinks.k1.kafka.bootstrap.servers = h1:9092,h2:9092,h3:9092

# a1.sinks.k1.channel = c1

a1.sources = r1 r2 r3

## source 配置

a1.sources.r1.type = TAILDIR

a1.sources.r1.channels = c1

a1.sources.r1.positionFile = /root/chekp1/taildir_position.json

a1.sources.r1.filegroups = f1

a1.sources.r1.filegroups.f1 = /root/weblog/log1/access.log

a1.sources.r1.maxBatchCount = 1000

a1.sources.r1.interceptors = i1 i2 i3

a1.sources.r1.interceptors.i1.type = timestamp

a1.sources.r1.interceptors.i2.type = cn.doitedu.flume.custom.EncryptInterceptor$EncryptInterceptorBuilder

a1.sources.r1.interceptors.i2.indices = 0:4

a1.sources.r1.interceptors.i2.idxSplitBy = :

a1.sources.r1.interceptors.i2.dataSplitBy = ,

a1.sources.r1.interceptors.i3.type = static

a1.sources.r1.interceptors.i3.key = logtype

a1.sources.r1.interceptors.i3.value = access

a1.sources.r2.type = TAILDIR

a1.sources.r2.channels = c1

a1.sources.r2.positionFile = /root/chekp2/taildir_position.json

a1.sources.r2.filegroups = f1

a1.sources.r2.filegroups.f1 = /root/weblog/log2/nginx.log

a1.sources.r2.maxBatchCount = 1000

a1.sources.r2.interceptors = i1 i2 i3

a1.sources.r2.interceptors.i1.type = timestamp

a1.sources.r2.interceptors.i2.type = cn.doitedu.flume.custom.EncryptInterceptor$EncryptInterceptorBuilder

a1.sources.r2.interceptors.i2.indices = 0:4

a1.sources.r2.interceptors.i2.idxSplitBy = :

a1.sources.r2.interceptors.i2.dataSplitBy = ,

a1.sources.r2.interceptors.i3.type = static

a1.sources.r2.interceptors.i3.key = logtype

a1.sources.r2.interceptors.i3.value = nginx

a1.sources.r3.type = TAILDIR

a1.sources.r3.channels = c1

a1.sources.r3.positionFile = /root/chekp3/taildir_position.json

a1.sources.r3.filegroups = f1

a1.sources.r3.filegroups.f1 = /root/weblog/log3/weblog.log

a1.sources.r3.maxBatchCount = 1000

a1.sources.r3.interceptors = i1 i2 i3

a1.sources.r3.interceptors.i1.type = timestamp

a1.sources.r3.interceptors.i2.type = cn.doitedu.flume.custom.EncryptInterceptor$EncryptInterceptorBuilder

a1.sources.r3.interceptors.i2.indices = 0:4

a1.sources.r3.interceptors.i2.idxSplitBy = :

a1.sources.r3.interceptors.i2.dataSplitBy = ,

a1.sources.r3.interceptors.i3.type = static

a1.sources.r3.interceptors.i3.key = logtype

a1.sources.r3.interceptors.i3.value = weblog

## channel 配置

a1.channels = c1

a1.channels.c1.type = file

a1.channels.c1.capacity = 1000000

a1.channels.c1.transactionCapacity = 100

a1.channels.c1.checkpointDir = /root/channel_chkp

a1.channels.c1.dataDirs = /root/channel_data

## sink 配置

a1.sinks = k1 k2

a1.sinks.k1.type = avro

a1.sinks.k1.channel = c1

a1.sinks.k1.hostname = c704

a1.sinks.k1.port = 4545

a1.sinks.k2.type = avro

a1.sinks.k2.channel = c1

a1.sinks.k2.hostname = c705

a1.sinks.k2.port = 4545

## sink processor – fail over 失败配置

a1.sinkgroups = g1

a1.sinkgroups.g1.sinks = k1 k2

a1.sinkgroups.g1.processor.type = failover

## 对两个sink分配不同的优先级

a1.sinkgroups.g1.processor.priority.k1 = 200

a1.sinkgroups.g1.processor.priority.k2 = 100

## 主sink失败后，停用惩罚时间

a1.sinkgroups.g1.processor.maxpenalty = 5000

a1.sources = s1

a1.channels = c1

a1.sinks = k1

## source 配置

a1.sources.s1.type = avro

a1.sources.s1.bind= 0.0.0.0

a1.sources.s1.port = 4545

a1.sources.s1.channels = c1

a1.sources.s1.interceptors = i1

a1.sources.s1.interceptors.i1.type = timestamp

## channel 配置

a1.channels.c1.type = file

a1.channels.c1.capacity = 1000000

a1.channels.c1.transactionCapacity = 100

a1.channels.c1.checkpointDir = /root/lev2_channel_chkp

a1.channels.c1.dataDirs = /root/lev2_channel_data

## sink配置

a1.sinks.k1.type = hdfs

a1.sinks.k1.channel = c1

a1.sinks.k1.hdfs.path = hdfs://c701:8020/doitedu/%{logtype}/%Y-%m-%d/%H/

a1.sinks.k1.hdfs.filePrefix = doitedu-

a1.sinks.k1.hdfs.fileSuffix = .log.gz

a1.sinks.k1.hdfs.rollInterval = 0

a1.sinks.k1.hdfs.rollSize = 134217728

a1.sinks.k1.hdfs.rollCount = 0

## 配置压缩

a1.sinks.k1.hdfs.fileType = CompressedStream

a1.sinks.k1.hdfs.codeC = gzip

## 数据格式

a1.sinks.k1.hdfs.serializer = TEXT