ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > 数据分析 > 针对flume中扇出复用（源exec）源码修改，并编译flume

针对flume中扇出复用（源exec）源码修改，并编译flume

原创 数据分析 作者：loveheping 时间：2018-04-14 22:41:19 0 删除 编辑
一、编写目地
  最近研究Flume组件中关于复用数据，多路复用可以根据设定的信息，进一步分流。通过对flume组件的了解，除使用拦截器外，还可以对源码修改或自定义源。本次测试使用的是对源码修改。使用虚拟搭建flume + kafka环境，编写功能简单的多路复用，主要是用flume的exec收集数据源，放入到logger、hdfs、kafka队列。在这期间遇到了很多问题，也学到了一些知识，打算做个笔记吧，帮助自己也帮助别人；

二、编译环境说明
1、jdk版本：jdk1.7.0_80  注：根据pom.xml中要求flume-ng-1.6.0需要在jdk1.7中编译，使jdk1.8可能会存在问题
2、flume版本：flume-ng-1.6.0-cdh5.7.0
3、maven版本：apache-maven-3.3.9

三、Flume源码的修改
3、用maven编译flume-ng-1.6.0-cdh5.7.0源码
3.1、下载flume
cd /opt/sourcecode
wget http://archive.cloudera.com/cdh5/cdh/5/flume-ng-1.6.0-cdh5.7.0-src.tar.gz
tar -zxvf flume-ng-1.6.0-cdh5.7.0-src.tar.gz

3.2、修改源码：
cd /opt/sourcecode/flume-ng-1.6.0-cdh5.7.0/flume-ng-core/src/main/java/org/apache/flume/source
在ExecSource.java中增加如下代码：
vi ExecSource.java
// 创建map(注：在源码中336行增加)

点击(此处)折叠或打开

  1. Map<String,String> map=new HashMap<String,String>();

// 截取每行数据前两个字节(根据个人业务逻辑修改或使用flume中的拦截器)
// (注：在源码中339行下增加)
// 其中"state"对应于selector.header，"channelFlag"对应于mapping.XXX(如：mapping=US)
//通过这样在从源获取数据时对数据做了封装，成为event并加入header作为分流标记。
// 当event要推送入channel时，会根据header的值将其推送入相应的channel中，详细在flume配置文件

点击(此处)折叠或打开

  1. String channelFlag = line.substring(0,2);
  2. map.put("state", channelFlag);

修改(注：源码中第340行)

点击(此处)折叠或打开

  1. eventList.add(EventBuilder.withBody(line.getBytes(charset)));
  2. 为
  3. eventList.add(EventBuilder.withBody(line.getBytes(charset),map));
保存退(:wq)

四、Flume源码的编译；
4.1、源码编译
cd /opt/sourcecode/flume-ng-1.6.0-cdh5.7.0
mvn install -DskipTests

点击(此处)折叠或打开

  1. [INFO] ------------------------------------------------------------------------
  2. [INFO] Reactor Summary:
  3. [INFO]
  4. [INFO] Apache Flume ....................................... SUCCESS [ 3.621 s]
  5. [INFO] Flume NG SDK ....................................... SUCCESS [ 5.543 s]
  6. [INFO] Flume NG Configuration ............................. SUCCESS [ 1.117 s]
  7. [INFO] Flume Auth ......................................... SUCCESS [ 3.623 s]
  8. [INFO] Flume NG Core ...................................... SUCCESS [ 5.092 s]
  9. [INFO] Flume NG Sinks ..................................... SUCCESS [ 0.180 s]
  10. [INFO] Flume NG HDFS Sink ................................. SUCCESS [ 1.932 s]
  11. [INFO] Flume NG IRC Sink .................................. SUCCESS [ 0.954 s]
  12. [INFO] Flume NG Channels .................................. SUCCESS [ 0.153 s]
  13. [INFO] Flume NG JDBC channel .............................. SUCCESS [ 0.989 s]
  14. [INFO] Flume NG file-based channel ........................ SUCCESS [ 1.078 s]
  15. [INFO] Flume NG Spillable Memory channel .................. SUCCESS [ 0.960 s]
  16. [INFO] Flume NG Node ...................................... SUCCESS [ 1.258 s]
  17. [INFO] Flume NG Embedded Agent ............................ SUCCESS [ 1.937 s]
  18. [INFO] Flume NG HBase Sink ................................ SUCCESS [ 2.021 s]
  19. [INFO] Flume NG ElasticSearch Sink ........................ SUCCESS [ 1.095 s]
  20. [INFO] Flume NG Morphline Solr Sink ....................... SUCCESS [ 6.976 s]
  21. [INFO] Flume Kafka Sink ................................... SUCCESS [ 1.065 s]
  22. [INFO] Flume NG Kite Dataset Sink ......................... SUCCESS [ 4.729 s]
  23. [INFO] Flume NG Hive Sink ................................. SUCCESS [ 0.841 s]
  24. [INFO] Flume Sources ...................................... SUCCESS [ 0.099 s]
  25. [INFO] Flume Scribe Source ................................ SUCCESS [ 0.791 s]
  26. [INFO] Flume JMS Source ................................... SUCCESS [ 0.847 s]
  27. [INFO] Flume Twitter Source ............................... SUCCESS [ 0.785 s]
  28. [INFO] Flume Kafka Source ................................. SUCCESS [ 0.707 s]
  29. [INFO] Flume Taildir Source ............................... SUCCESS [ 0.763 s]
  30. [INFO] flume-kafka-channel ................................ SUCCESS [ 0.695 s]
  31. [INFO] Flume legacy Sources ............................... SUCCESS [ 0.074 s]
  32. [INFO] Flume legacy Avro source ........................... SUCCESS [ 1.625 s]
  33. [INFO] Flume legacy Thrift Source ......................... SUCCESS [ 0.714 s]
  34. [INFO] Flume NG Clients ................................... SUCCESS [ 0.079 s]
  35. [INFO] Flume NG Log4j Appender ............................ SUCCESS [ 7.450 s]
  36. [INFO] Flume NG Tools ..................................... SUCCESS [ 0.843 s]
  37. [INFO] Flume NG distribution .............................. SUCCESS [ 30.321 s]
  38. [INFO] Flume NG Integration Tests ......................... SUCCESS [ 1.106 s]
  39. [INFO] ------------------------------------------------------------------------
  40. [INFO] BUILD SUCCESS
  41. [INFO] ------------------------------------------------------------------------
  42. [INFO] Total time: 01:36 min
  43. [INFO] Finished at: 2018-04-14T21:46:36+08:00
  44. [INFO] Final Memory: 106M/705M
  45. [INFO] ------------------------------------------------------------------------
最后将会在flume-ng-dist/target目录下产生文件：apache-flume-1.6.0-cdh5.7.0-bin.tar.gz

4.2、编译问题：
4.2.1、ua-parser-1.3.0下载失败
 [ERROR] Failed to execute goal on project flume-ng-morphline-solr-sink: Could not resolve dependencies
  for project org.apache.flume.flume-ng-sinks:flume-ng-morphline-solr-sink:jar:1.6.0: Failed to collect dependencies at 
  org.kitesdk:kite-morphlines-all:pom:1.0.0 -> org.kitesdk:kite-morphlines-useragent:jar:1.0.0 -> 
  ua_parser:ua-parser:jar:1.3.0: Failed to read artifact descriptor for ua_parser:ua-parser:jar:1.3.0:
   Could not transfer artifact ua_parser:ua-parser:pom:1.3.0 from/to maven-twttr (http://maven.twttr.com):
    maven.twttr.com: Unknown host maven.twttr.com -> [Help 1]
 解决办法：在flume源码的pom.xml下加个repository，资源地址可以自行选择
 点击(此处)折叠或打开
  1. <repository>
  2.    <id>p2.jfrog.org</id>
  3.    <url>http://p2.jfrog.org/libs-releases</url>
  4. </repository>
4.2.2、 
pentaho-aggdesigner-algorithm-5.1.5-hyde
下载失败
This problem persists with the current Hive-1.2.0 release since it depends on Calcite-1.3.0-incubating<https://github.com/apache/hive/blob/master/pom.xml#L104>
which in turn depends on artifacts like pentaho-aggdesigner-algorithm-5.1.5-hyde<https://github.com/apache/incubator-calcite/blob/branch-1.3/pom.xml#L264>
, There are also licensing issues that I am not completely sure of. For instance what is the
 解决办法：在flume源码的pom.xml下加个repository，资源地址可以自行选择
点击(此处)折叠或打开
  1. <repository>
  2.       <releases>
  3.         <enabled>true</enabled>
  4.         <updatePolicy>always</updatePolicy>
  5.         <checksumPolicy>warn</checksumPolicy>
  6.       </releases>
  7.       <id>conjars</id>
  8.       <name>Conjars</name>
  9.       <url>http://conjars.org/repo</url>
  10.       <layout>default</layout>
  11.     </repository>
   
五、测试
   详见博客：http://blog.itpub.net/31511218/viewspace-2152461/


六、总结


【来自@若泽大数据】





来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/31511218/viewspace-2152950/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： Flume与Kafka整合--扇入、扇出功能整合，其中扇出包括：复制流、复用流
下一篇： 大数据之数仓平台设计思路01
请登录后发表评论 登录
全部评论
loveheping

注册时间：2017-12-13

  • 博文量
    41
  • 访问量
    80862

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593