±à¼ÍƼö: |
±¾ÎÄÀ´×ÔÓÚzhanshi258.iteye.com/blog,½éÉÜÁË·¢Õ¹Ê·£¬¶¨Òå°²×°£¬Ìص㼰ÕûÌå¼Ü¹¹¡£ |
|
ÈÕÖ¾ÊÇϵͳÊý¾ÝµÄ»ùʯ£¬¶ÔÓÚϵͳµÄ°²È«À´Ëµ·Ç³£ÖØÒª£¬Ëü¼Ç¼ÁËϵͳÿÌì·¢ÉúµÄ¸÷ÖÖ¸÷ÑùµÄÊÂÇ飬Óû§¿ÉÒÔͨ¹ýËüÀ´¼ì²é´íÎó·¢ÉúµÄÔÒò£¬»òÕßѰÕÒÊܵ½¹¥»÷ʱ¹¥»÷ÕßÁôϵĺۼ£¡£ÈÕÖ¾Ö÷ÒªµÄ¹¦ÄÜÊÇÉó¼ÆºÍ¼à²â¡£Ëü»¹¿ÉÒÔʵʱµØ¼à²âϵͳ״̬£¬¼à²âºÍ×·×ÙÇÖÈëÕß¡£ÏÖÔÚ»¥ÁªÍøÉÏ´æÔÚµÄÈÕÖ¾×é¼þ¸÷ÖÖ¸÷Ñù£¬ÎÒÃÇÕâÀïÖ÷Òª½²µÄÊÇFlume¡£
Flume ·¢Õ¹ÀúÊ·
Cloudera ¿ª·¢µÄ·Ö²¼Ê½ÈÕÖ¾ÊÕ¼¯ÏµÍ³ Flume£¬ÊÇ hadoop Öܱß×é¼þÖ®Ò»¡£Æä¿ÉÒÔʵʱµÄ½«·Ö²¼ÔÚ²»Í¬½Úµã¡¢»úÆ÷ÉϵÄÈÕÖ¾ÊÕ¼¯µ½
hdfs ÖС£Flume ³õʼµÄ·¢Ðа汾Ŀǰ±»Í³³ÆÎª Flume OG£¨original generation£©£¬ÊôÓÚ
cloudera¡£µ«Ëæ×Å FLume ¹¦ÄܵÄÀ©Õ¹£¬Flume OG ´úÂ빤³ÌÓ·Öס¢ºËÐÄ×é¼þÉè¼Æ²»ºÏÀí¡¢ºËÐÄÅäÖò»±ê×¼µÈȱµã±©Â¶³öÀ´£¬ÓÈÆäÊÇÔÚ
Flume OG µÄ×îºóÒ»¸ö·¢Ðа汾 0.94.0 ÖУ¬ÈÕÖ¾´«Êä²»Îȶ¨µÄÏÖÏóÓÈΪÑÏÖØ£¬Õâµã¿ÉÒÔÔÚ BigInsights
²úÆ·ÎĵµµÄ troubleshooting °å¿é·¢ÏÖ¡£ÎªÁ˽â¾öÕâЩÎÊÌ⣬2011 Äê 10 Ô 22
ºÅ£¬cloudera Íê³ÉÁË Flume-728£¬¶Ô Flume ½øÐÐÁËÀï³Ì±®Ê½µÄ¸Ä¶¯£ºÖع¹ºËÐÄ×é¼þ¡¢ºËÐÄÅäÖÃÒÔ¼°´úÂë¼Ü¹¹£¬Öع¹ºóµÄ°æ±¾Í³³ÆÎª
Flume NG£¨next generation£©£»¸Ä¶¯µÄÁíÒ»ÔÒòÊǽ« Flume ÄÉÈë apache
ÆìÏ£¬cloudera Flume ¸ÄÃûΪ Apache Flume¡£µ±È»ÎÒÃÇÏÖÔÚÓõÄÊÇFlume
NG£¬ËùÒÔ²»ÔÙ½²Flume OGµÄÄÚÈÝ¡£
Flume¶¨Òå
FlumeÊÇÒ»¸ö¸ß¿ÉÓ㬸߿ɿ¿£¬·Ö²¼Ê½º£Á¿ÈÕÖ¾²É¼¯¡¢¾ÛºÏºÍ´«Êäϵͳ£¬FlumeÖ§³ÖÔÚÈÕ־ϵͳÖж¨ÖƸ÷ÀàÊý¾Ý·¢ËÍ·½£¬ÓÃÓÚÊÕ¼¯Êý¾Ý£»Í¬Ê±£¬FlumeÌṩ¶ÔÊý¾Ý½øÐмòµ¥´¦Àí£¬²¢Ð´µ½¸÷ÖÖÊý¾Ý½ÓÊÜ·½£¨¿É¶¨ÖÆ£©µÄÄÜÁ¦¡£
Flume¼Ü¹¹½éÉÜ
FlumeÈÕÖ¾ÊÕ¼¯½á¹¹Í¼

Flume µÄºËÐÄÊǰÑÊý¾Ý´ÓÊý¾ÝÔ´ÊÕ¼¯¹ýÀ´£¬ÔÙË͵½Ä¿µÄµØ¡£
ΪÁ˱£Ö¤ÊäËÍÒ»¶¨³É¹¦£¬ÔÚË͵½Ä¿µÄµØÖ®Ç°£¬»áÏÈ»º´æÊý¾Ý£¬´ýÊý¾ÝÕæÕýµ½´ïÄ¿µÄµØºó£¬É¾³ý×Ô¼º»º´æµÄÊý¾Ý¡£
Flume ´«ÊäµÄÊý¾ÝµÄ»ù±¾µ¥Î»ÊÇ Event£¬Èç¹ûÊÇÎı¾Îļþ£¬Í¨³£ÊÇÒ»ÐмǼ£¬ÕâÒ²ÊÇÊÂÎñµÄ»ù±¾µ¥Î»¡£
Event ´Ó Source£¬Á÷Ïò Channel£¬ÔÙµ½ Sink£¬±¾ÉíΪһ¸ö byte Êý×飬²¢¿ÉЯ´ø
headers ÐÅÏ¢¡£
Event ´ú±í×ÅÒ»¸öÊý¾ÝÁ÷µÄ×îСÍêÕûµ¥Ôª£¬´ÓÍⲿÊý¾ÝÔ´À´£¬ÏòÍⲿµÄÄ¿µÄµØÈ¥¡£
Flume ÔËÐеĺËÐÄÊÇ Agent¡£ËüÊÇÒ»¸öÍêÕûµÄÊý¾ÝÊÕ¼¯¹¤¾ß£¬º¬ÓÐÈý¸öºËÐÄ×é¼þ£¬·Ö±ðÊÇ Source¡¢Channel¡¢Sink¡£
Source ¿ÉÒÔ½ÓÊÕÍⲿԴ·¢Ë͹ýÀ´µÄÊý¾Ý¡£²»Í¬µÄ Source£¬¿ÉÒÔ½ÓÊܲ»Í¬µÄÊý¾Ý¸ñʽ¡£±ÈÈçÓÐĿ¼³Ø(spooling
directory)Êý¾ÝÔ´£¬¿ÉÒÔ¼à¿ØÖ¸¶¨Îļþ¼ÐÖеÄÐÂÎļþ±ä»¯£¬Èç¹ûĿ¼ÖÐÓÐÎļþ²úÉú£¬¾Í»áÁ¢¿Ì¶ÁÈ¡ÆäÄÚÈÝ¡£
Channel ÊÇÒ»¸ö´æ´¢µØ£¬½ÓÊÕ Source µÄÊä³ö£¬Ö±µ½ÓÐ Sink Ïû·Ñµô Channel
ÖеÄÊý¾Ý¡£
Channel ÖеÄÊý¾ÝÖ±µ½½øÈëµ½ÏÂÒ»¸öChannelÖлòÕß½øÈëÖն˲Żᱻɾ³ý¡£
µ± Sink дÈëʧ°Üºó£¬¿ÉÒÔ×Ô¶¯ÖØÆô£¬²»»áÔì³ÉÊý¾Ý¶ªÊ§£¬Òò´ËºÜ¿É¿¿¡£
Sink »áÏû·Ñ Channel ÖеÄÊý¾Ý£¬È»ºóË͸øÍⲿԴ»òÕ߯äËû Source¡£ÈçÊý¾Ý¿ÉÒÔдÈëµ½
HDFS »òÕß HBase ÖС£
Flume ºËÐĸÅÄîÕûÀí
Agent AgentÖаüº¬¶à¸ösourcesºÍsinks¡£
Client Éú²úÊý¾Ý£¬ÔËÐÐÔÚÒ»¸ö¶ÀÁ¢µÄÏ̡߳£
Source ´ÓClientÊÕ¼¯Êý¾Ý£¬´«µÝ¸øChannel¡£ÓÃÀ´Ïû·Ñ´«µÝµ½¸Ã×é¼þµÄEvent¡£
Sink ´ÓChannelÊÕ¼¯Êý¾Ý£¬½«Event´«µÝµ½Flow PipelineÖеÄÏÂÒ»¸öAgent¡£
Channel ÖÐתEventÁÙʱ´æ´¢£¬±£´æSource´«µÝ¹ýÀ´Event£¬Á¬½Ó sources ºÍ
sinks ¡£
Events Ò»¸öÊý¾Ýµ¥Ôª£¬´øÓÐÒ»¸ö¿ÉÑ¡µÄÏûϢͷ¡£¿ÉÒÔÊÇÈÕÖ¾¼Ç¼¡¢ avro ¶ÔÏóµÈ¡£
Flume ÌØµã
flumeµÄÊý¾ÝÁ÷ÓÉʼþ(Event)¹á´©Ê¼ÖÕ¡£Ê¼þÊÇFlumeµÄ»ù±¾Êý¾Ýµ¥Î»£¬ËüЯ´øÈÕÖ¾Êý¾Ý(×Ö½ÚÊý×éÐÎʽ)²¢ÇÒЯ´øÓÐÍ·ÐÅÏ¢£¬ÕâЩEventÓÉAgentÍⲿµÄSourceÉú³É£¬µ±Source²¶»ñʼþºó»á½øÐÐÌØ¶¨µÄ¸ñʽ»¯£¬È»ºóSource»á°ÑʼþÍÆÈë(µ¥¸ö»ò¶à¸ö)ChannelÖС£Äã¿ÉÒÔ°ÑChannel¿´×÷ÊÇÒ»¸ö»º³åÇø£¬Ëü½«±£´æÊ¼þÖ±µ½Sink´¦ÀíÍê¸Ãʼþ¡£Sink¸ºÔð³Ö¾Ã»¯ÈÕÖ¾»òÕß°ÑʼþÍÆÏòÁíÒ»¸öSource¡£
AgentÊÇFlumeÖÐ×îСµÄÔËÐе¥Î»£¬Ò»¸öAgentÖÐÓÉSource¡¢SinkºÍChannelÈý¸ö×é¼þ¹¹³É¡£
EventÊÇFlumeÖлù±¾Êý¾Ýµ¥Î»£¬EventÖаüº¬Óд«ÊäÊý¾Ý¼°Êý¾ÝÍ·Êý¾Ý°ü
ÈçÏÂͼËùʾ£º

ÖµµÃ×¢ÒâµÄÊÇ£¬FlumeÌṩÁË´óÁ¿ÄÚÖõÄSource¡¢ChannelºÍSinkÀàÐÍ¡£²»Í¬ÀàÐ͵ÄSource,ChannelºÍSink¿ÉÒÔ×ÔÓÉ×éºÏ¡£×éºÏ·½Ê½»ùÓÚÓû§ÉèÖõÄÅäÖÃÎļþ£¬·Ç³£Áé»î¡£
±ÈÈ磺Channel¿ÉÒÔ°ÑʼþÔÝ´æÔÚÄÚ´æÀҲ¿ÉÒԳ־û¯µ½±¾µØÓ²ÅÌÉÏ¡£Sink¿ÉÒÔ°ÑÈÕ־дÈëHDFS,
HBase£¬ÉõÖÁÊÇÁíÍâÒ»¸öSourceµÈµÈ¡£FlumeÖ§³ÖÓû§½¨Á¢¶à¼¶Á÷£¬Ò²¾ÍÊÇ˵£¬¶à¸öagent¿ÉÒÔÐͬ¹¤×÷£¬²¢ÇÒÖ§³ÖFan-in¡¢Fan-out¡¢Contextual
Routing¡¢Backup Routes£¬ÕâÒ²ÕýÊÇNBÖ®´¦¡£
ÈçÏÂͼËùʾ:

Flume ÕûÌå¼Ü¹¹×ܽá
Flume¼Ü¹¹ÕûÌåÉÏ¿´¾ÍÊÇ source-->channel-->sink µÄÈý²ã¼Ü¹¹£¬ÀàËÆÉú³ÉÕߺÍÏû·ÑÕߵļܹ¹£¬ËûÃÇÖ®¼äͨ¹ýqueue£¨channel£©´«Ê䣬½âñî¡£
Source:Íê³É¶ÔÈÕÖ¾Êý¾ÝµÄÊÕ¼¯£¬·Ö³É transtion ºÍ event ´òÈëµ½channelÖ®ÖС£
Channel:Ö÷ÒªÌṩһ¸ö¶ÓÁеŦÄÜ£¬¶ÔsourceÌṩÖеÄÊý¾Ý½øÐмòµ¥µÄ»º´æ¡£
Sink:È¡³öChannelÖеÄÊý¾Ý£¬½øÐÐÏàÓ¦µÄ´æ´¢Îļþϵͳ£¬Êý¾Ý¿â£¬»òÕßÌá½»µ½Ô¶³Ì·þÎñÆ÷¡£
¶ÔÏÖÓгÌÐò¸Ä¶¯×îСµÄʹÓ÷½Ê½ÊÇʹÓÃÊÇÖ±½Ó¶ÁÈ¡³ÌÐòÔÀ´¼Ç¼µÄÈÕÖ¾Îļþ£¬»ù±¾¿ÉÒÔʵÏÖÎÞ·ì½ÓÈ룬²»ÐèÒª¶ÔÏÖÓгÌÐò½øÐÐÈκθ͝¡£
Flume ÏÂÔØ¡¢°²×°
°²×°JDK
1.½«ÏÂÔØºÃµÄJDK°ü½âѹ£¬±ÈÈçÎҵĽâѹµ½ /home/liuqing/jdk1.7.0_72 Ŀ¼ÏÂ
2.ÅäÖû·¾³±äÁ¿
ÔÚ/etc/profile ÎļþÖÐÌí¼Ó
Java´úÂë
export JAVA_HOME=/home/liuqing/jdk1.7.0_72
export PATH=$JAVA_HOME/bin:$PATH
export CLASS_PATH=.:$JAVA_HOME/lib/dt.jar: $JAVA_HOME/lib/tools.jar: $CLASS_PATH
|
3.Ö´ÐÐsource profile
4.ÔÚÃüÁîÐÐÊäÈë java -version
³öÏÖ£º
java version "1.7.0_72"
Java(TM) SE Runtime Environment (build 1.7.0_72-b14)
Java HotSpot(TM) 64-Bit Server VM (build 24.72-b04,
mixed mode)
±íʾ°²×°³É¹¦
°²×°Flume
1. ´Ó¹ÙÍø http://flume.apache.org/download.html ÏÂÔØ×îÐµİ²×°°ü
2. ½âѹËõ£¬±ÈÈçÎҵĽâѹµ½ /home/liuqing/hadoop/flumeĿ¼
3. ÐÞ¸Ä flume-env.sh ÅäÖÃÎļþ,Ö÷ÒªÊÇJAVA_HOME±äÁ¿ÉèÖÃ
JAVA_HOME=/home/liuqing/jdk1.7.0_72
4. ÑéÖ¤ÊÇ·ñ°²×°³É¹¦
root@ubuntu:/home/liuqing/hadoop/flume/bin# ./flume-ng
version
³öÏÖ£º
Flume 1.6.0
Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
±íʾ°²×°³É¹¦
°¸Àý
°¸Àý1. µ¥½Úµã FlumeÅäÖÃ
1. н¨ÅäÖÃÎļþ£¬ÅäÖÃÎļþʾÀý
Java´úÂë
# example.conf:
A single-node Flume configuration
# agent×é¼þÃû³Æ
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# source ÅäÖÃ
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# sink ÅäÖÃ
a1.sinks.k1.type = logger
# ʹÓÃÄÚ´æÖÐBuffer Event Channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# °ó¶¨ source ºÍ sink µ½channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1 |
½«ÉÏÊöÅäÖôæÎª£º/home/liuqing/hadoop/flume/conf/example.conf
2. È»ºóÎÒÃǾͿÉÒÔÆô¶¯ Flume ÁË£º
ÔÚ/home/liuqing/hadoop/flume·¾¶ÏÂÔËÐУº
Java´úÂë
bin/flume-ngagent
--conf conf --conf-file conf/example.conf --name
a1 -Dflume.root.logger=INFO,console |
ÆäÖÐ -c/--conf ºó¸úÅäÖÃĿ¼£¬-f/--conf-file ºó¸ú¾ßÌåµÄÅäÖÃÎļþ£¬-n/--name
Ö¸¶¨agentµÄÃû³Æ
3. È»ºóÎÒÃÇÔÙ¿ªÒ»¸ö shell ÖÕ¶Ë´°¿Ú£¬telnet ÉÏÅäÖÃÖÐÕìÌýµÄ¶Ë¿Ú£¬¾Í¿ÉÒÔ·¢ÏûÏ¢¿´µ½Ð§¹ûÁË£º
Java´úÂë
$ telnet localhost
44444
Trying 127.0.0.1...
Connected to localhost.localdomain (127.0.0.1).
Escape character is '^]'.
Hello world! <ENTER>
OK |
4.Flume ÖÕ¶Ë´°¿Ú´Ëʱ»á´òÓ¡³öÈçÏÂÐÅÏ¢£¬¾Í±íʾ³É¹¦ÁË
Java´úÂë
12/06/1915: 32:19
INFO source.NetcatSource: Source starting
12/06/1915:32: 19 INFO source.NetcatSource: Created
serverSocket: sun.nio.ch.ServerSocketChannelImpl [/127.0.0.1:44444]
12/06/1915:32:34 INFO sink.LoggerSink: Event:
{ headers:{} body: 48 65 6C 6C 6F 20 77 6F 72
6C 64 21 0D Hello world!. } |
ÖÁ´Ë£¬ÔÛÃǵĵÚÒ»¸ö Flume Agent ËãÊDz¿Êð³É¹¦ÁË£¡
°¸Àý2. ½áºÏʵ¼ÊÏîÄ¿
²Î¿¼£ºhttps://github.com/gilt/logback-flume-appender
1. ÔÚ/home/liuqing/hadoop/flume/conf/ÏÂн¨ÅäÖÃÎļþ test.conf
Java´úÂë
agent1.sources
= source1
agent1.sinks = sink1
agent1.channels = channel1
# Describe/configure source1
agent1.sources.source1.type = avro
agent1.sources.source1.bind = 0.0.0.0
agent1.sources.source1.port = 44444
# Describe sink1
#ÈÕÖ¾Îļþ°´Ê±¼äÉú³É
#agent1.sinks.sink1.type = FILE_ROLL
#agent1.sinks.sink1.sink.directory = /home/liuqing/hadoop/flume/flume-out
#agent1.sinks.sink1.sink.rollInterval = 1800
#agent1.sinks.sink1.batchSize = 5
#ÈÕÖ¾Îļþ¸ù¾Ý´óСÉú³É
#Éú³ÉĿ¼ÔÚconfÎļþ¼ÐϵÄlog4j.properties¿ÉÒÔÅäÖÃ
agent1.sinks.sink1.type = logger
# Use a channel which buffers events in memory
agent1.channels.channel1.type = memory
agent1.channels.channel1.capacity = 1000
agent1.channels.channel1.transactionCapactiy =
100
# Bind the source and sink to the channel
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1 |
2. ÏîÄ¿ÒѾÅäºÃÁËlogback.xml Îļþ
ÔÚlogback.xmlÎļþÖÐÌí¼Ó
Xml´úÂë
<appender name="flumeApplender"
class= "com.xxx.hd.extended.log.flume. FlumeLogstashV1Appender">
<flumeAgents>
192.168.23.235:44444,
</flumeAgents>
<flumeProperties>
connect-timeout=4000;
request-timeout=8000
</flumeProperties>
<batchSize>2048</batchSize>
<reportingWindow>20480</reportingWindow>
<additionalAvroHeaders>
myHeader=myValue
</additionalAvroHeaders>
<application>ProjectName</application>
<layout class="ch.qos.logback.classic.PatternLayout">
<pattern>%d{HH:mm:ss.SSS} %-5level %logger{36}
- \(%file:%line\) - %msg%n%ex< /pattern>
</layout>
</appender> |
3. È»ºóÎÒÃǾͿÉÒÔÆô¶¯ Flume ÁË£º
ÔÚ/home/liuqing/hadoop/flume·¾¶ÏÂÔËÐУº
Java´úÂë
bin/flume-ng agent
--conf ./conf/ -f conf/lqtest.conf -n agent1 |
4. ÏÖÔÚÈÕÖ¾»á´òÓ¡µ½/home/liuqing/hadoop/flume/logsĿ¼ÏÂ
ÈÕÖ¾ÎļþÂú128M¾Í»á×Ô¶¯½¨Ò»¸öÐ嵀 |