{"id":131912,"date":"2020-07-06T08:06:56","date_gmt":"2020-07-06T00:06:56","guid":{"rendered":"http:\/\/4563.org\/?p=131912"},"modified":"2020-07-06T08:06:56","modified_gmt":"2020-07-06T00:06:56","slug":"flume-%e5%ad%a6%e4%b9%a0%e4%b9%8b%e8%b7%af%ef%bc%88%e4%b8%89%ef%bc%89flume%e7%9a%84source%e7%b1%bb%e5%9e%8b","status":"publish","type":"post","link":"http:\/\/4563.org\/?p=131912","title":{"rendered":"Flume \u5b66\u4e60\u4e4b\u8def\uff08\u4e09\uff09Flume\u7684Source\u7c7b\u578b"},"content":{"rendered":"

\u5b98\u65b9\u6587\u6863\u4ecb\u7ecd\uff1ahttp:\/\/flume.apache.org\/FlumeUserGuide.html#flume-sources<\/a><\/p>\n

<\/a>Avro Source<\/h3>\n

\u5185\u7f6e Avro Server\uff0c\u53ef\u63a5\u53d7 Avro \u5ba2\u6237\u7aef\u53d1\u9001\u7684\u6570\u636e<\/p>\n

\u5b98\u65b9\u6587\u6863\u63cf\u8ff0\uff1a
\"Flume<\/p>\n

\"Flume<\/p>\n

\u793a\u4f8b\uff1a<\/p>\n

\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
<\/pre>\n<\/td>\n
\n
vim avro_source.properties<\/span>
<\/span>
a1.sources = s1<\/span>
a1.sinks = k1<\/span>
a1.channels = c1<\/span>
<\/span>
// \u914d\u7f6esource<\/span>
a1.sources.s1.channels = c1<\/span>
a1.sources.s1.type = avro<\/span>
a1.sources.s1.bind = 0.0.0.0<\/span>
a1.sources.s1.port = 6666<\/span>
<\/span>
// \u914d\u7f6echannels<\/span>
a1.channels.c1.type = memory<\/span>
<\/span>
// \u914d\u7f6esinks<\/span>
a1.sinks.k1.channel = c1<\/span>
a1.sinks.k1.type = logger<\/span>
<\/span>
// \u4e3asources\u548csinks\u7ed1\u5b9achannels<\/span>
a1.sources.s1.channels = c1<\/span>
a1.sinks.k1.channel = c1<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u542f\u52a8 Flume NG:<\/p>\n
\n\n\n
\n
1<\/span>
<\/pre>\n<\/td>\n
\n
bin/flume-ng agent -c conf/ -f conf/avro_source.properties -n a1 -Dflume.root.logger=INFO,console<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u5f00\u59cb\u8f93\u5165\u6d4b\u8bd5\u6570\u636e:<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
<\/pre>\n<\/td>\n
\n
vim  666.txt <\/span>
<\/span>
123<\/span>
123<\/span>
123<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u5ba2\u6237\u7aef\u8f93\u5165\uff1a<\/p>\n
\n\n\n
\n
1<\/span>
<\/pre>\n<\/td>\n
\n
bin/flume-ng avro-client -c conf/ -H bigdata -p 6666 -F 666.txt<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>Thrift Source<\/h3>\n
\u5185\u7f6e Thrift Server\uff0c\u53ef\u63a5\u53d7 Thrift \u5ba2\u6237\u7aef\u53d1\u9001\u7684\u6570\u636e\u3002ThriftSource \u4e0eAvro Source \u57fa\u672c\u4e00\u81f4\u3002\u53ea\u8981\u628asource\u7684\u7c7b\u578b\u6539\u6210thrift\u5373\u53ef<\/p>\n
<\/a>Exec Source<\/h3>\n
\u6267\u884c\u6307\u5b9a\u7684shell\uff0c\u5e76\u4ece\u8be5\u547d\u4ee4\u6807\u51c6\u8f93\u51fa\u4e2d\u83b7\u53d6\u6570\u636e\uff0c\u4f46\u8003\u8651\u5230\u8be5flume agent\u4e0d\u8fd0\u884c\u6216\u8005\u6307\u4ee4\u51fa\u9519\u65f6\uff0c\u5c06\u65e0\u6cd5\u6536\u96c6\u5230\u6570\u636e\uff0c\u6545\u751f\u4ea7\u73af\u5883\u5f88\u5c11\u91c7\u7528 \u3002<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
<\/pre>\n<\/td>\n
\n
vim exec_source.properties<\/span>
<\/span>
// \u914d\u7f6e\u6587\u4ef6<\/span>
a1.sources= s1  <\/span>
a1.sinks= k1  <\/span>
a1.channels= c1  <\/span>
   <\/span>
// \u914d\u7f6esources<\/span>
a1.sources.s1.type = exec  <\/span>
a1.sources.s1.command = tail -F /tmp/test.log  <\/span>
a1.sources.s1.channels = c1  <\/span>
   <\/span>
// \u914d\u7f6esinks <\/span>
a1.sinks.k1.type= logger  <\/span>
a1.sinks.k1.channel= c1  <\/span>
   <\/span>
// \u914d\u7f6echannel<\/span>
a1.channels.c1.type= memory<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u542f\u52a8 Flume NG:<\/p>\n
\n\n\n
\n
1<\/span>
<\/pre>\n<\/td>\n
\n
bin/flume-ng agent --conf conf/ -f conf/exec_source.properties --name a1 -Dflume.root.logger=DEBUG,console<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u751f\u6210\u6d4b\u8bd5\u6570\u636e\uff1a<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
<\/pre>\n<\/td>\n
\n
cd /tmp/test.log<\/span>
<\/span>
echo "hello world" >>  test.log<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>Spooling Directory Source<\/h3>\n
\u76d1\u542c\u4e00\u4e2a\u6587\u4ef6\u5939\u4e0b\u65b0\u4ea7\u751f\u7684\u6587\u4ef6\uff0c\u5e76\u8bfb\u53d6\u5185\u5bb9\uff0c\u53d1\u81f3 channel\u3002\u4f7f\u7528\u8be5 Source \u9700\u8981\u6ce8\u610f\u4e24\u70b9\uff1a\u7b2c\u4e00\u4e2a\u662f\u62f7\u8d1d\u5230 spool \u76ee\u5f55\u4e0b\u7684\u6587\u4ef6\u4e0d\u53ef\u4ee5\u518d\u6253\u5f00\u7f16\u8f91\uff0c\u7b2c\u4e8c\u4e2a\u662f spool \u76ee\u5f55\u4e0b\u4e0d\u53ef\u5305\u542b\u76f8\u5e94\u7684\u5b50\u76ee\u5f55\u3002\u8fd9\u4e2a\u4e3b\u8981\u7528\u9014\u4f5c\u4e3a\u5bf9\u65e5\u5fd7\u7684\u51c6\u5b9e\u65f6\u76d1\u63a7\u3002\u7531\u4e8e\u8be5Source\u53ef\u9760\u6027\u548c\u7a33\u5b9a\u6027\u8f83\u597d\uff0c\u88ab\u4e0d\u5c11\u516c\u53f8\u91c7\u7528\u3002<\/p>\n
\u793a\u4f8b\uff1a<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
<\/pre>\n<\/td>\n
\n
vim spooling_dir_source.properties  <\/span>
<\/span>
a1.sources = s1  <\/span>
a1.sinks = k1  <\/span>
a1.channels = c1  <\/span>
   <\/span>
// Describe/configure the source  <\/span>
a1.sources.s1.type =spooldir  <\/span>
a1.sources.s1.spoolDir =/tmp/logs  <\/span>
a1.sources.s1.fileHeader= true  <\/span>
a1.sources.s1.channels =c1  <\/span>
   <\/span>
// Describe the sink  <\/span>
a1.sinks.k1.type = logger  <\/span>
a1.sinks.k1.channel = c1  <\/span>
   <\/span>
// Use a channel which buffers events inmemory  <\/span>
a1.channels.c1.type = memory<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u542f\u52a8 Flume NG:<\/p>\n
\n\n\n
\n
1<\/span>
<\/pre>\n<\/td>\n
\n
bin/flume-ng agent --conf conf/ -f conf/spooling_dir_source.properties --name a1 -Dflume.root.logger=DEBUG,console<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u751f\u6210\u6d4b\u8bd5\u6570\u636e<\/p>\n
\n\n\n
\n
1<\/span>
<\/pre>\n<\/td>\n
\n
mv test.log logs<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>KafKa Source<\/h3>\n
\u5185\u7f6e\u4e86Kafka Consumer\uff0c\u53ef\u4ece KaFka Broker \u4e2d\u8bfb\u53d6\u67d0\u4e2a topic\u7684\u6570\u636e\uff0c\u5199\u5165Channel\u3002<\/p>\n
<\/a>Taildir Source<\/h3>\n
\u53ef\u4ee5\u5b9e\u65f6\u76d1\u63a7\u4e00\u4e2a\u76ee\u5f55\u4e0b\u6587\u4ef6\u7684\u53d8\u5316\uff0c\u5e76\u5b9e\u65f6\u8bfb\u53d6\u65b0\u589e\u6570\u636e\uff0c\u8bb0\u5f55\u65ad\u70b9\uff0c\u4fdd\u8bc1\u91cd\u542f Agent \u540e\u6570\u636e\u4e0d\u4e22\u5931\u6216\u88ab\u91cd\u590d\u4f20\u8f93\u3002<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
30<\/span>
31<\/span>
32<\/span>
33<\/span>
34<\/span>
35<\/span>
<\/pre>\n<\/td>\n
\n
vim tail_dir_source.properties<\/span>
<\/span>
a1.sources = r1<\/span>
a1.channels = c1<\/span>
a1.sinks = s1<\/span>
<\/span>
a1.sources.r1.type = TAILDIR<\/span>
// positionFile \u68c0\u67e5\u70b9\u6587\u4ef6\u8def\u5f84<\/span>
a1.sources.r1.positionFile = /tmp/flume/taildir_position.json<\/span>
// \u6307\u5b9afilegroups\uff0c\u53ef\u4ee5\u6709\u591a\u4e2a\uff0c\u4ee5\u7a7a\u683c\u5206\u9694\uff1b\uff08TailSource\u53ef\u4ee5\u540c\u65f6\u76d1\u63a7tail\u591a\u4e2a\u76ee\u5f55\u4e2d\u7684\u6587\u4ef6\uff09<\/span>
a1.sources.r1.filegroups = f1<\/span>
// \u6307\u5b9a\u76d1\u63a7\u7684\u6587\u4ef6\u76ee\u5f55f1,\u5339\u914d\u7684\u6587\u4ef6\u4fe1\u606f<\/span>
a1.sources.r1.filegroups.f1 = /tmp/baihe-flume/log_.*.log<\/span>
a1.sources.r1.fileHeader = true<\/span>
a1.sources.ri.maxBatchCount = 1000<\/span>
<\/span>
//  sink1 \u914d\u7f6e<\/span>
a1.sinks.s1.type = file_roll<\/span>
a1.sinks.s1.sink.directory = /tmp/flumefiles<\/span>
a1.sinks.s1.sink.rollInterval = 0<\/span>
<\/span>
 <\/span>
// fileChannel \u914d\u7f6e<\/span>
a1.channels.c1.type = file<\/span>
// -->\u68c0\u6d4b\u70b9\u6587\u4ef6\u6240\u5b58\u50a8\u7684\u76ee\u5f55<\/span>
a1.channels.c1.checkpointDir = /tmp/flume/checkpoint/baihe/<\/span>
// -->\u6570\u636e\u5b58\u50a8\u6240\u5728\u7684\u76ee\u5f55\u8bbe\u7f6e<\/span>
a1.channels.c1.dataDirs = /tmp/flume/data/baihe/<\/span>
// -->\u96a7\u9053\u7684\u6700\u5927\u5bb9\u91cf<\/span>
a1.channels.c1.capacity = 10000<\/span>
// -->\u4e8b\u52a1\u5bb9\u91cf\u7684\u6700\u5927\u503c\u8bbe\u7f6e<\/span>
a1.channels.c1.transactionCapacity = 200<\/span>
<\/span>
a1.sources.r1.channels = c1<\/span>
a1.sinks.s1.channel = c1<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u542f\u52a8 Flume NG\uff1a<\/p>\n
\n\n\n
\n
1<\/span>
<\/pre>\n<\/td>\n
\n
bin/flume-ng agent --conf conf/ -f conf/tail_dir_source.properties --name a1 -Dflume.root.logger=DEBUG,console<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u751f\u6210\u6d4b\u8bd5\u6570\u636e\uff1a<\/p>\n
\u91c7\u96c6\u76ee\u5f55\u4e0b\u751f\u6210\u6b63\u5219\u5bf9\u5e94\u7684\u6587\u4ef6\u4fe1\u606f\uff0c\u5199\u5165\u6570\u636e<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
<\/pre>\n<\/td>\n
\n
cd /tmp/baihe-flume<\/span>
<\/span>
echo "message1" >> log_20190915.log<\/span>
echo "message12" >> log_20190915.log<\/span>
<\/span>
// \u67e5\u770bsink\u6587\u4ef6<\/span>
tail  -f /tmp/flumefiles/1568530522032-1<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
TailSource\u4f7f\u7528\u4e86RandomAccessFile\u6765\u6839\u636epositionFile\u4e2d\u4fdd\u5b58\u7684\u6587\u4ef6\u4f4d\u7f6e\u6765\u8bfb\u53d6\u6587\u4ef6\u7684\uff0c\u5728agent\u91cd\u542f\u4e4b\u540e\uff0c\u4ea6\u4f1a\u5148\u4ecepositionFile\u4e2d\u627e\u5230\u4e0a\u6b21\u8bfb\u53d6\u7684\u6587\u4ef6\u4f4d\u7f6e\uff0c\u4fdd\u8bc1\u5185\u5bb9\u4e0d\u4f1a\u91cd\u590d\u53d1\u9001<\/p>\n
<\/a>Syslog<\/h3>\n
Syslog \u5206\u4e3a Tcp Source\u548c UDP Source\u4e24\u79cd\uff0c\u5206\u522b\u63a5\u53d7tcp\u548cudp\u534f\u8bae\u53d1\u8fc7\u6765\u7684\u6570\u636e\uff0c\u5199\u5165Channel<\/p>\n
<\/a>HTTP source<\/h3>\n
\u53ef\u63a5\u53d7HTTP\u534f\u8bae\u53d1\u6765\u7684\u6570\u636e\uff0c\u5bbe\u5199\u5165Channel<\/p>\n
<\/a>\u5982\u4f55\u9009\u62e9Flume Source?<\/h3>\n","protected":false},"excerpt":{"rendered":"
\u5b98\u65b9\u6587\u6863\u4ecb\u7ecd\uff1ahttp:\/\/flu…<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[],"tags":[],"_links":{"self":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/131912"}],"collection":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=131912"}],"version-history":[{"count":0,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/131912\/revisions"}],"wp:attachment":[{"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=131912"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=131912"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=131912"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}