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 ????????????????????啟動(dòng)一個(gè)agent進(jìn)程

??avro-client ????????????????啟動(dòng)一個(gè)用于測試avro?source的客戶端（能夠發(fā)送avro序列化流）

??version ???????????????????顯示當(dāng)前flume的版本信息

global options: ??全局通用選項(xiàng)

??--conf,-c <conf> ?????????指定flume的系統(tǒng)配置文件所在目錄

??--classpath,-C <cp> ???????添加額外的jar路徑

??--dryrun,-d ??????????????不去真實(shí)啟動(dòng)flume?agent，而是打印當(dāng)前命令

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

??-Dproperty=value ?????????傳入java環(huán)境參數(shù)

??-Xproperty=value ?????????傳入所需的JVM配置參數(shù)

agent options:

??--name,-n <name> ?????????agent的別名（在用戶采集方案配置文件中）

??--conf-file,-f <file> ?????????指定用戶采集方案配置文件的路徑

??--zkConnString,-z <str> ?????指定zookeeper的連接地址

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

??--no-reload-conf ???????????關(guān)閉配置文件動(dòng)態(tài)加載

??--help,-h ??????????????????display help text

avro-client options:

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

??--host,-H <host> ????avro序列化數(shù)據(jù)所要發(fā)往的目標(biāo)主機(jī)（avro?source所在機(jī)器）

??--port,-p <port> ?????avro序列化數(shù)據(jù)所要發(fā)往的目標(biāo)主機(jī)的端口號(hào)

??--dirname <dir> ?????需要被序列化發(fā)走的數(shù)據(jù)所在目錄（提前準(zhǔn)備好測試數(shù)據(jù)放在一個(gè)文件中）

??--filename,-F <file> ??需要被序列化發(fā)走的數(shù)據(jù)所在文件(default: std input)

??--headerFile,-R <file> ?存儲(chǔ)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.

# 定義這個(gè)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關(guān)聯(lián)

a1.sources.r1.channels = c1??

# 描述和配置sink組件：k1

a1.sinks.k1.type = logger

#?sink也要關(guān)聯(lián)channel

a1.sinks.k1.channel = c1

# 描述和配置channel組件，此處使用是內(nèi)存緩存的方式

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要發(fā)往的channel

type

–

本source的類別名稱：exec

command

–

本source所要運(yùn)行的linux命令,比如： tail?-F /path/file

shell

–

指定運(yùn)行上述命令所用shell

restartThrottle

10000

命令die了以后，重啟的時(shí)間間隔

restart

false

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

logStdErr

false

是否收集命令的錯(cuò)誤輸出stderr

batchSize

20

提交的event批次大小

batchTimeout

3000

發(fā)往下游沒完成前，等待的時(shí)間

selector.type

replicating

指定channel選擇器：replicating or multiplexing

selector.*

選擇器的具體參數(shù)

interceptors

–

指定攔截器

interceptors.*

?指定的攔截器的具體參數(shù)

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

是否將所采集文件的絕對(duì)路徑添加到header中

fileHeaderKey

file

上述header的key名稱

basenameHeader

false

是否將文件名添加到header

basenameHeaderKey

basename

上述header的key名稱

includePattern

^.*$

指定需要采集的文件名的正則表達(dá)式

ignorePattern

^$

指定要排除的文件名的正則表達(dá)式

如果一個(gè)文件名即符合includePattern又匹配ignorePattern，則該文件不采

trackerDir

.flumespool

記錄元數(shù)據(jù)的目錄所在路徑，可以用絕對(duì)路徑也可以用相對(duì)路徑（相對(duì)于采集目錄）

trackingPolicy

rename

采集進(jìn)度跟蹤策略，有兩種：?“rename”和?“tracker_dir”. 本參數(shù)只在deletePolicy=never時(shí)才生效

?“rename”- 采完的文件根據(jù)filesuffix重命名

?“tracker_dir” - 采完的文件會(huì)在trackerDir目錄中生成一個(gè)同名的空文件

consumeOrder

oldest

采集順序：?oldest,?youngest?and?random.

oldest和youngest情況下，可能會(huì)帶來一定效率的損失；（需要對(duì)文件夾中所有文件進(jìn)行一次掃描以尋找最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

一次傳輸?shù)絚hannel的event條數(shù)（一批）

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

–

要綁定的端口號(hào)

threads

–

服務(wù)的最大線程數(shù)

selector.type

selector.*

interceptors

–

Space-separated list of interceptors

interceptors.*

compression-type

none

壓縮類型：跟發(fā)過來的數(shù)據(jù)是否壓縮要匹配：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

–

?數(shù)據(jù)發(fā)往的channel

type

–

本source的名稱：

org.apache.flume.source.kafka.KafkaSource

kafka.bootstrap.servers

–

Kafka?broker服務(wù)器列表，逗號(hào)分隔

kafka.consumer.group.id

flume

Kafka消費(fèi)者組id

kafka.topics

–

Kafka消息主題列表，逗號(hào)隔開

kafka.topics.regex

–

用正則表達(dá)式來指定一批topic；本參數(shù)的優(yōu)先級(jí)高于kafka.topics

batchSize

1000

寫入channel的event?批，最大消息條數(shù)

batchDurationMillis

1000

批次寫入channel的最大時(shí)長

backoffSleepIncrement

1000

Kafka?Topic?顯示為空時(shí)觸發(fā)的初始和增量等待時(shí)間。

maxBackoffSleep

5000

Kafka?Topic?顯示為空時(shí)觸發(fā)的最長等待時(shí)間

useFlumeEventFormat

false

默認(rèn)情況下，event 將從Kafka Topic 直接作為字節(jié)直接進(jìn)入event 主體。設(shè)置為true以讀取event 作為Flume Avro二進(jìn)制格式。與Kafka Sink上的相同屬性或Kafka Channel上的parseAsFlumeEvent屬性一起使用時(shí)，這將保留在生成端發(fā)送的任何Flume標(biāo)頭。

setTopicHeader

true

是否要往header中加入一個(gè)kv：topic信息

topicHeader

topic

應(yīng)上面開關(guān)的需求，加入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消費(fèi)者參數(shù)，格式如下：

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

（就是在消費(fèi)者參數(shù)前加統(tǒng)一前綴：?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>

–

每個(gè)文件組的絕路路徑，文件名可用正則表達(dá)式

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

關(guān)閉非活動(dòng)文件的時(shí)延。如果被關(guān)閉的這個(gè)文件又在某個(gè)時(shí)間有了新增行,會(huì)被此source檢測到，并重新打開

writePosInterval

3000

3s 記錄一次偏移量到positionfile

batchSize

100

提交event到channel的批次最大條數(shù)

maxBatchCount

Long.MAX_VALUE

控制在一個(gè)文件上連續(xù)讀取的最大批次個(gè)數(shù)（如果某個(gè)文件正在被高速寫入，那就應(yīng)該讓這個(gè)參數(shù)調(diào)為最大值，以讓source可以集中精力專采這個(gè)文件）

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還是主機(jī)名

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

能存儲(chǔ)的最大事件event數(shù)

transactionCapacity

100

最大事務(wù)控制容量

keep-alive

3

添加或移除event的超時(shí)時(shí)間

byteCapacityBufferPercentage

20

除了body以外的字節(jié)所能占用的容量百分比

byteCapacity

see description

channel中最大的總byte數(shù)（只計(jì)算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來實(shí)現(xiàn)

@GuardedBy(value = "queueLock")p

rivate?LinkedBlockingDeque<Event> queue;

//下面的兩個(gè)信號(hào)量用來做同步操作，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;

//下面幾個(gè)變量為配置文件中Memory Channel的配置項(xiàng)

// 一個(gè)事務(wù)中Event的最大數(shù)目

private?volatile?Integer transCapacity;

// 向queue中添加、移除Event的等待時(shí)間

private?volatile?int?keepAlive;

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

private?volatile?int?byteCapacity;

private?volatile?int?lastByteCapacity;

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

private?volatile?int?byteCapacityBufferPercentage;

// 用于標(biāo)示byteCapacity中剩余空間的信號(hào)量

private?Semaphore bytesRemaining;

// 用于記錄Memory Channel的一些指標(biāo)，后面可以通過配置監(jiān)控來觀察Flume的運(yùn)行情況

private?ChannelCounter channelCounter;

Property Name Default

Description

?

type

–

別名：?file.

checkpointDir

~/.flume/file-channel/checkpoint

Checkpoint信息保存目錄

useDualCheckpoints

false

Checkpoint是否雙重checkpoint機(jī)制

backupCheckpointDir

–

備份checkpoint的保存目錄

dataDirs

~/.flume/file-channel/data

Event數(shù)據(jù)緩存目錄

transactionCapacity

10000

事務(wù)管理容量

checkpointInterval

30000

記錄checkpoint信息的時(shí)間間隔

maxFileSize

2146435071

控制一個(gè)數(shù)據(jù)文件的大小規(guī)格

minimumRequiredSpace

524288000

所需的最低磁盤空間，低于則停止接收新數(shù)據(jù)

capacity

1000000

最大event緩存數(shù)

keep-alive

3

等待添加數(shù)據(jù)的最大時(shí)間

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服務(wù)器地址

kafka.topic

flume-channel

所使用的topic

kafka.consumer.group.id

flume

消費(fèi)者組id

parseAsFlumeEvent

true

跟上、下游匹配，是否需要將數(shù)據(jù)解析為Flume的Event格式

pollTimeout

500

從kafka取數(shù)據(jù)的超時(shí)時(shí)間

defaultPartitionId

–

默認(rèn)指派的partitionid

partitionIdHeader

–

將一個(gè)event指派到某個(gè)分區(qū)時(shí)所使用的header 的key

kafka.consumer.auto.offset.reset

latest

初始化讀取偏移量的策略

a1.sources = r1

# 配置兩個(gè)channel,為了便于觀察

# c1 是kafkachannel ,c2是一個(gè)內(nèi)存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

# 內(nèi)存channel 配置,并對(duì)接一個(gè)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

–

從哪個(gè)channel取數(shù)據(jù)

type

–

別名：?hdfs

hdfs.path

–

目標(biāo)hdfs存儲(chǔ)路徑（URI）

hdfs.filePrefix

FlumeData

指定生成的文件名前綴

hdfs.fileSuffix

–

后綴

hdfs.inUsePrefix

–

正在寫入的文件的前綴標(biāo)識(shí)

hdfs.inUseSuffix

.tmp

正在寫入的文件的后綴標(biāo)識(shí)

hdfs.rollInterval

30

切換文件的條件：間隔時(shí)間；為0則不生效

hdfs.rollSize

134217728

切換文件的條件：文件大小；為0則不生效

hdfs.rollCount

10

切換文件的條件：event條數(shù)；為0則不生效

hdfs.idleTimeout

0

不活躍文件的關(guān)閉超時(shí)時(shí)長；0則不自動(dòng)關(guān)閉

hdfs.batchSize

100

從channel中取一批數(shù)據(jù)的最大大小；

hdfs.codeC

–

壓縮編碼: gzip, bzip2, lzo, lzop, snappy

hdfs.fileType

SequenceFile

目標(biāo)文件格式: SequenceFile,?DataStream?or?CompressedStream?

注意：DataStream?不能支持壓縮

CompressedStream?必須設(shè)置壓縮編碼

SequenceFile 可壓縮可不壓縮

hdfs.maxOpenFiles

5000

允許同時(shí)最多打開的文件數(shù)；如果超出，則會(huì)關(guān)閉最早打開的

hdfs.minBlockReplicas

–

目標(biāo)文件的block副本數(shù)

hdfs.writeFormat

Writable

指定sequence?file中的對(duì)象類型；支持Text和Writable

同時(shí)請(qǐng)使用Text，否則后續(xù)數(shù)據(jù)處理平臺(tái)可能無法解析

hdfs.threadsPoolSize

10

操作HDFS時(shí)的線程池大小

hdfs.rollTimerPoolSize

1

檢查文件是否需要被roll的線程數(shù)

hdfs.kerberosPrincipal

–

Kerberos user principal for accessing secure HDFS

hdfs.kerberosKeytab

–

Kerberos keytab for accessing secure HDFS

hdfs.proxyUser

hdfs.round

false

目錄通配符切換是是否需要切掉尾數(shù)

hdfs.roundValue

10

時(shí)間尾數(shù)切掉多少

hdfs.roundUnit

minute

時(shí)間尾數(shù)切掉大小的單位-?second,?minute?or?hour.

hdfs.timeZone

Local Time

時(shí)間通配符所使用的時(shí)區(qū)

hdfs.useLocalTimeStamp

false

所用的時(shí)間是否要從agent?sink本地獲取

hdfs.closeTries

0

重命名已完成文件的重試次數(shù)；0則一直嘗試重命名

hdfs.retryInterval

180

關(guān)閉一個(gè)文件的重試時(shí)間間隔

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服務(wù)器列表

kafka.topic

default-flume-topic

Kafka的topic

flumeBatchSize

100

從channel中取event的批次大小

kafka.producer.acks

1

Kafka生產(chǎn)者消息推送應(yīng)答級(jí)別：

1?: Leader接收到即回應(yīng)

0 ：不等回應(yīng)?

-1：副本同步完成，再回應(yīng)

useFlumeEventFormat

false

是否使用avro序列化

defaultPartitionId

–

指定所有event默認(rèn)發(fā)往的分區(qū)id；不指定則按kafka生產(chǎn)者的分區(qū)器，均勻分發(fā)

partitionIdHeader

–

通過在header中指定分區(qū)id，來約束這個(gè)event發(fā)往的分區(qū)

allowTopicOverride

true

允許在event的header中指定要寫入的topic

topicHeader

topic

如果上一條開關(guān)開啟，則在header中放入的key的名稱

Other Kafka Producer Properties

–

可以用kafka.producer.xxx來配置任何kafka?producer的參數(shù)

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

–

目標(biāo)avro source的主機(jī)

port

–

目標(biāo)avro source的綁定端口

batch-size

100

number of event to batch together for send.

connect-timeout

20000

連接超時(shí)時(shí)間

request-timeout

20000

請(qǐng)求超時(shí)時(shí)間

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 “l(fā)isten 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實(shí)例對(duì)象

然后調(diào)?source的start方法

source讀數(shù)據(jù)，然后將數(shù)據(jù)轉(zhuǎn)成event

然后將event傳給channelprocessor.processEvent(event)

processEvent中第一個(gè)動(dòng)作就是調(diào)用攔截器攔截這個(gè)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;

????/**
?????* 框架調(diào)用本方法，開始采集數(shù)據(jù)
?????* 自定義代碼去讀取數(shù)據(jù)，轉(zhuǎn)為event
?????* 用getChannelProcessor（）方法（定義在父類中）去獲取框架的channelprocessor（channel處理器）
?????* 調(diào)用這個(gè)channelprocessor將event提交給channel
?????*/
????@Override
????public synchronized void start() {

????????super.start();
????????// 用于向channel提交數(shù)據(jù)的一個(gè)處理器
????????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();
????????}

????????// 構(gòu)造一個(gè)線程池
????????exec = Executors.newSingleThreadExecutor();
????????// 向線程池提交數(shù)據(jù)采集任務(wù)
????????exec.execute(new HoldOffsetRunnable(offset, logfile, channelProcessor, batchsize, positionfilepath));

????}

????/**
?????* 停止前要調(diào)用的方法
?????* 可以在這里做一些資源關(guān)閉清理工作
?????*/
????@Override
????public synchronized void stop() {
????????super.stop();

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

????/**
?????* 獲取配置文件中的參數(shù)，來配置本source實(shí)例
?????* <p>
?????* 要哪些參數(shù)：
?????* 偏移量記錄文件所在路徑
?????* 要采集的文件所在路徑
?????*
?????* @param context
?????*/
????public void configure(Context context) {

????????// 這是我們source用來記錄偏移量的文件路徑
????????this.positionfilepath = context.getString("positionfile", "./");

????????// 這是我們source要采集的日志文件的路徑
????????this.logfile = context.getString("logfile");

????????// 這是用戶配置的采集事務(wù)批次最大值
????????this.batchsize = context.getInteger("batchsize", 100);

????????// 如果日志文件路徑?jīng)]有指定，則拋異常
????????if (StringUtils.isBlank(logfile)) throw new RuntimeException("請(qǐng)配置需要采集的文件路徑");

????}

????/**
?????* 采集文件的具體工作線程任務(wù)類
?????*/
????private static class HoldOffsetRunnable implements Runnable {

????????long offset;
????????String logfilepath;
????????String positionfilepath;
????????ChannelProcessor channelProcessor; ?// channel提交器 （里面會(huì)調(diào)攔截器，會(huì)開啟寫入channel的事務(wù)）
????????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);

????????????????// 循環(huán)讀數(shù)據(jù)
????????????????String line = null;

????????????????// 記錄上一批提交的時(shí)間
????????????????long lastBatchTime = System.currentTimeMillis();
????????????????while (true) {
????????????????????line = raf.readLine();
????????????????????if(line == null ){
????????????????????????Thread.sleep(2000);
????????????????????????continue;
????????????????????}

????????????????????// 將數(shù)據(jù)轉(zhuǎn)成event
????????????????????Event event = EventBuilder.withBody(line.getBytes());
????????????????????// 裝入list batch
????????????????????synchronized (HoldOffesetSource.class) {
????????????????????????events.add(event);
????????????????????}

????????????????????// 判斷批次大小是否滿 或者 時(shí)間到了沒有
????????????????????if (events.size() >= batchsize || timeout(lastBatchTime)) {
????????????????????????// 滿足，則提交
????????????????????????channelProcessor.processEventBatch(events);

????????????????????????// 記錄提交時(shí)間
????????????????????????lastBatchTime = systemClock.currentTimeMillis();

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

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

????????????????????}

????????????????????// 不滿足，繼續(xù)讀
????????????????}
????????????} catch (FileNotFoundException e) {
????????????????logger.error("要采集的文件不存在");
????????????} catch (IOException e) {
????????????????logger.error("我也不知道怎么搞的，不好意思，我罷工了");
????????????} catch (InterruptedException e) {
????????????????logger.error("線程休眠出問題了");
????????????}
????????}

????????// 判斷是否批次間隔超時(shí)
????????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;
????// 數(shù)據(jù)體字段之間的分隔符
????String dataSplitBy;

????/**
?????* 構(gòu)造方法
?????* @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;
????}

????// 這個(gè)方法會(huì)被框架調(diào)用一次，用來做一些初始化工作
????public void initialize() {

????}

????// 攔截方法--對(duì)一個(gè)event進(jìn)行處理
????public Event intercept(Event event) {

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

????????// 數(shù)據(jù)的字段數(shù)組
????????String[] dataFieldsArr = dataStr.split(dataSplitBy);

????????// 需要加密的索引的數(shù)組
????????String[] idxArr = indices.split(idxSplitBy);

????????for (String s : idxArr) {

????????????int index = Integer.parseInt(s);
????????????// 取出要加密的字段的內(nèi)容
????????????String field = dataFieldsArr[index];
????????????// MD5加密這個(gè)字段
????????????String encryptedField = DigestUtils.md5Hex(field);
????????????// BASE64編碼
????????????byte[] bytes = Base64.decodeBase64(encryptedField);
????????????// 替換掉原來的未加密內(nèi)容
????????????dataFieldsArr[index] = new String(bytes);
????????}

????????// 將加密過的字段重新拼接成一條數(shù)據(jù)，并使用原來的分隔符
????????StringBuilder sb = new StringBuilder();
????????for (String field : dataFieldsArr) {
????????????sb.append(field).append(dataSplitBy);
????????}

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

????????// 返回加密后的字段所封裝的event對(duì)象
????????return EventBuilder.withBody(sb.toString().getBytes());
????}

????// 攔截方法--對(duì)一批event進(jìn)行處理
????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退出前，會(huì)調(diào)一次該方法，進(jìn)行需要的清理、關(guān)閉操作
????public void close() {

????}

????/**
?????* 攔截器的構(gòu)造器
?????*/
????public staticc lass EncryptInterceptorBuilder implements Interceptor.Builder{
????????// 要加密的字段索引s
????????String indices;
????????// 索引之間的分隔符
????????String idxSplitBy;
????????// 數(shù)據(jù)體字段之間的分隔符
????????String dataSplitBy;

????????// 構(gòu)造一個(gè)攔截器實(shí)例
????????public Interceptor build() {

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

????????// 獲取配置文件中的攔截器參數(shù)
????????public void configure(Context context) {
????????????// 要加密的字段索引s
????????????this.indices = context.getString(Constants.INDICES);
????????????// 索引之間的分隔符
????????????this.idxSplitBy = context.getString(Constants.IDX_SPLIT_BY);
????????????// 數(shù)據(jù)體字段之間的分隔符
????????????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

## 對(duì)兩個(gè)sink分配不同的優(yōu)先級(jí)

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

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

## 主sink失敗后，停用懲罰時(shí)間

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

## 數(shù)據(jù)格式

a1.sinks.k1.hdfs.serializer = TEXT