思享工具箱

flume使用之httpSource

摘要：
exe、kafka。。。有一个非常简单的源代码httpsource。Flume将在启动后启动web服务以侦听指定的IP和端口。

flume自带很长多的source，如：exe、kafka...其中有一个非常简单的source——httpsource，使用httpSource，flume启动后会拉起一个web服务来监听指定的ip和port。常用的使用场景：对于有些应用环境中，不能部署Flume SDK及其依赖项，可以在代码中通过HTTP而不是Flume的PRC发送数据的情况，此时HTTP SOURCE可以用来将数据接收到Flume中。

1、httpsource 参数：

配置参数默认值描述
type http (org.apache.fluem.source.httpSource）
bind 绑定的IP地址或主机名
port 绑定的端口号
enableSSLfalse 
keystore 使用的keystore文件的路径
keystorePassword 能够进入keystore的密码
handlerJSONHandlerHTTP SOURCE使用的处理程序类
handler.* 传给处理程序类的任何参数 可以 通过使用此参数（*）配置传入

1）handler：

Flume使用一个可插拔的“handler”程序来实现转换，如果不指定默认是：JSONHandler，它能处理JSON格式的事件，格式如下。此外用户可以自定义handler，必须实现HTTPSourceHandler接口。

json数据格式：

[html] view plain copy
 
  1. [ { "headers":{"":"","":""  
  2.                  },  
  3.      "body":"the first event"  
  4.    },  
  5.    { "headers":{"":"","":""  
  6.                  },  
  7.      "body":"the second event"  
  8.    }  
  9.      
  10. ]  


2、简单介绍一下flume的logger sink：

记录INFO级别的日志，一般用于调试。本文将使用这种类型的sink，配置的属性：

  • type  logger
  • maxBytesToLog    16    Maximum number of bytes of the Event body to log

注意：要求必须在 --conf 参数指定的目录下有 log4j的配置文件，可以通过-Dflume.root.logger=INFO,console在命令启动时手动指定log4j参数。

3、简单的httpSource实例：

1）下载flume、解压：

[html] view plain copy
 
  1. cd /usr/local/  
  2. wget http://mirror.bit.edu.cn/apache/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz  
  3. tar -xvzf apache-flume-1.7.9-bin.tar.gz  

配置flume的环境变量：

[html] view plain copy
 
  1. vim /etc/profile  
  2.   
  3. export PS1="[u@`/sbin/ifconfig eth0|grep 'inet '|awk -F'[: ]+' '{print $4}'` W]"'$ '  
  4. export FLUME_HOME=/usr/local/apache-flume-1.6.0-bin  
  5. export PATH=$PATH:$FLUME_HOME/bin  


2）安装jdk、配置环境变量；

3）配置flume：

[html] view plain copy
 
  1. cd /usr/local/flume/conf  
  2. vim flume-env.sh  

指定java_home，同时放入如下log4j.properties

[html] view plain copy
 
  1. ### set log levels ###  
  2. log4j.rootLogger = info,stdout ,  D ,  E  
  3.   
  4. ###  
  5. log4j.appender.stdout = org.apache.log4j.ConsoleAppender  
  6. log4j.appender.stdout.Target = System.out  
  7. log4j.appender.stdout.layout = org.apache.log4j.PatternLayout  
  8. log4j.appender.stdout.layout.ConversionPattern =  [%d{MM-dd HH:mm:ss}] [%p] [%c:%L] %m%n  
  9.    
  10. ### 输出到日志文件 ###  
  11. log4j.appender.D = org.apache.log4j.DailyRollingFileAppender  
  12. log4j.appender.D.File = /data/logs/flume/flume.log  
  13. log4j.appender.D.Append = true  
  14. log4j.appender.D.Threshold = info  
  15. log4j.appender.D.layout = org.apache.log4j.PatternLayout  
  16. log4j.appender.D.layout.ConversionPattern = [%d{MM-dd HH:mm:ss}] [%p] [%c:%L] %m%n  
  17.    
  18. ### 保存异常信息到单独文件 ###  
  19. log4j.appender.E = org.apache.log4j.DailyRollingFileAppender  
  20. log4j.appender.E.File =/data/logs/flume/flume_error.log  
  21. log4j.appender.E.Append = true  
  22. log4j.appender.E.Threshold = ERROR  
  23. log4j.appender.E.layout = org.apache.log4j.PatternLayout  
  24. log4j.appender.E.layout.ConversionPattern = [%d{MM-dd HH:mm:ss}] [%p] [%c:%L] %m%n  
  25.   
  26. ### sink  
  27. log4j.logger.com.iqiyi.ttbrain.log.flume.sink.MysqlSink= INFO, F, EE  
  28. log4j.additivity.com.iqiyi.ttbrain.log.flume.sink.MysqlSink = false  
  29. log4j.appender.F= org.apache.log4j.DailyRollingFileAppender  
  30. log4j.appender.F.File=/data/logs/flume/flume_sink.log  
  31. log4j.appender.F.Append = true  
  32. log4j.appender.F.Threshold = info  
  33. log4j.appender.F.layout=org.apache.log4j.PatternLayout    
  34. log4j.appender.F.layout.ConversionPattern= [%d{MM-dd HH:mm:ss}] [%p] [%c:%L] %m%n  
  35.   
  36. log4j.appender.EE= org.apache.log4j.DailyRollingFileAppender  
  37. log4j.appender.EE.File=/data/logs/flume/flume_sink_error.log  
  38. log4j.appender.EE.Append = true  
  39. log4j.appender.EE.Threshold = ERROR  
  40. log4j.appender.EE.layout=org.apache.log4j.PatternLayout    
  41. log4j.appender.EE.layout.ConversionPattern= [%d{MM-dd HH:mm:ss}] [%p] [%c:%L] %m%n  

4）配置httpSource：

[html] view plain copy
 
  1. cd /usr/local/flume/conf  
  2. vim http_test.conf  
  3.   
  4. a1.sources=r1  
  5. a1.sinks=k1  
  6. a1.channels=c1  
  7.   
  8. a1.sources.r1.type=http  
  9. a1.sources.r1.bind=localhost  
  10. a1.sources.r1.port=50000  
  11. a1.sources.r1.channels=c1  
  12.   
  13. a1.sinks.k1.type=logger  
  14. a1.sinks.k1.channel=c1  
  15.   
  16. a1.channels.c1.type=memory  
  17. a1.channels.c1.capacity=1000  
  18. a1.channels.c1.transactionCapacity=100  


5）启动flume：

[html] view plain copy
 
  1. flume-ng agent -c /usr/local/flume/conf/ -f /usr/local/flume/conf/http_test.conf -n a1  


6）测试：

开一个shell窗口，输入命令：

[html] view plain copy
 
  1. curl -X POST -d'[{"headers":{"h1":"v1","h2":"v2"},"body":"hello body"}]'  http://localhost:50000  

在/data/log/flume/flume.log 文件中可以看到：

[html] view plain copy
 
  1. [09-29 10:31:12] [INFO] [org.apache.flume.sink.LoggerSink:94] Event: { headers:{h1=v1, h2=v2} body: 68 65 6C 6C 6F 20 62 6F 64 79                   hello body }  


4、自定义handler：

假定xml请求格式，期望格式如下：

[html] view plain copy
 
  1. <events>  
  2.  <event>  
  3.      <headers><header1>value1</header1></headers>  
  4.      <body>test</body>  
  5.  </event>  
  6.  <event>  
  7.     <headers><header1>value1</header1></headers>  
  8.     <body>test2</body>  
  9.   </event>  
  10.  </events>  


1）pom.xml

[html] view plain copy
 
  1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
  2.   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">  
  3.   <modelVersion>4.0.0</modelVersion>  
  4.   <groupId>org.pq</groupId>  
  5.   <artifactId>flume-demo</artifactId>  
  6.   <packaging>jar</packaging>  
  7.   <version>1.0</version>  
  8.   <name>flume-demo Maven jar</name>  
  9.   <url>http://maven.apache.org</url>  
  10.   <dependencies>  
  11.     <dependency>  
  12.       <groupId>junit</groupId>  
  13.       <artifactId>junit</artifactId>  
  14.       <version>4.8.2</version>  
  15.       <scope>test</scope>  
  16.     </dependency>  
  17.     <dependency>  
  18.       <groupId>org.slf4j</groupId>  
  19.       <artifactId>slf4j-log4j12</artifactId>  
  20.       <version>1.7.7</version>  
  21.       <scope>compile</scope>  
  22.     </dependency>  
  23.     <dependency>  
  24.       <groupId>org.apache.flume</groupId>  
  25.       <artifactId>flume-ng-core</artifactId>  
  26.       <version>1.6.0</version>  
  27.       <scope>compile</scope>  
  28.     </dependency>  
  29.   </dependencies>  
  30.   <build>  
  31.     <finalName>flume-demo</finalName>  
  32.   </build>  
  33. </project>  

2）自定义handler：

[html] view plain copy
 
  1. package org.pq.flumeDemo.sources;  
  2. import com.google.common.base.Preconditions;  
  3. import org.apache.flume.Context;  
  4. import org.apache.flume.Event;  
  5. import org.apache.flume.event.EventBuilder;  
  6. import org.apache.flume.source.http.HTTPBadRequestException;  
  7. import org.apache.flume.source.http.HTTPSourceHandler;  
  8. import org.slf4j.Logger;  
  9. import org.slf4j.LoggerFactory;  
  10. import org.w3c.dom.Document;  
  11. import org.w3c.dom.Element;  
  12. import org.w3c.dom.Node;  
  13. import org.w3c.dom.NodeList;  
  14. import org.xml.sax.SAXException;  
  15.   
  16. import javax.servlet.http.HttpServletRequest;  
  17. import javax.xml.parsers.DocumentBuilder;  
  18. import javax.xml.parsers.DocumentBuilderFactory;  
  19. import java.util.ArrayList;  
  20. import java.util.HashMap;  
  21. import java.util.List;  
  22. import java.util.Map;  
  23.   
  24. public class HTTPSourceXMLHandler implements HTTPSourceHandler {  
  25.     private final String ROOT = "events";  
  26.     private final String EVENT_TAG = "event";  
  27.     private final String HEADERS_TAG = "headers";  
  28.     private final String BODY_TAG = "body";  
  29.   
  30.     private final String CONF_INSERT_TIMESTAMP = "insertTimestamp";  
  31.     private final String TIMESTAMP_HEADER = "timestamp";  
  32.     private final DocumentBuilderFactory documentBuilderFactory  
  33.             = DocumentBuilderFactory.newInstance();  
  34.   
  35.     // Document builders are not thread-safe.  
  36.     // So make sure we have one for each thread.  
  37.     private final ThreadLocal<DocumentBuilder> docBuilder  
  38.             = new ThreadLocal<DocumentBuilder>();  
  39.   
  40.     private boolean insertTimestamp;  
  41.     private static final Logger LOG = LoggerFactory.getLogger(HTTPSourceXMLHandler.class);  
  42.   
  43.   
  44.     public List<Event> getEvents(HttpServletRequest httpServletRequest) throws HTTPBadRequestException, Exception {  
  45.         if (docBuilder.get() == null) {  
  46.             docBuilder.set(documentBuilderFactory.newDocumentBuilder());  
  47.         }  
  48.         Document doc;  
  49.         final List<Event> events;  
  50.         try {  
  51.             doc = docBuilder.get().parse(httpServletRequest.getInputStream());              
  52.             Element root = doc.getDocumentElement();          
  53.   
  54.             root.normalize();  
  55.             // Verify that the root element is "events"  
  56.             Preconditions.checkState(  
  57.                     ROOT.equalsIgnoreCase(root.getTagName()));  
  58.   
  59.             NodeList nodes = root.getElementsByTagName(EVENT_TAG);  
  60.             LOG.info("get nodes={}",nodes);  
  61.   
  62.             int eventCount = nodes.getLength();  
  63.             events = new ArrayList<Event>(eventCount);  
  64.             for (int i = 0; i < eventCount; i++) {  
  65.                 Element event = (Element) nodes.item(i);  
  66.                 // Get all headers. If there are multiple header sections,  
  67.                 // combine them.  
  68.                 NodeList headerNodes  
  69.                         = event.getElementsByTagName(HEADERS_TAG);  
  70.                 Map<String, String> eventHeaders  
  71.                         = new HashMap<String, String>();  
  72.                 for (int j = 0; j < headerNodes.getLength(); j++) {  
  73.                     Node headerNode = headerNodes.item(j);  
  74.                     NodeList headers = headerNode.getChildNodes();  
  75.                     for (int k = 0; k < headers.getLength(); k++) {  
  76.                         Node header = headers.item(k);  
  77.   
  78.                         // Read only element nodes  
  79.                         if (header.getNodeType() != Node.ELEMENT_NODE) {  
  80.                             continue;  
  81.                         }  
  82.                         // Make sure a header is inserted only once,  
  83.                         // else the event is malformed  
  84.                         Preconditions.checkState(  
  85.                                 !eventHeaders.containsKey(header.getNodeName()),  
  86.                                 "Header expected only once " + header.getNodeName());  
  87.                         eventHeaders.put(  
  88.                                 header.getNodeName(), header.getTextContent());  
  89.                     }  
  90.                 }  
  91.                 Node body = event.getElementsByTagName(BODY_TAG).item(0);  
  92.                 if (insertTimestamp) {  
  93.                     eventHeaders.put(TIMESTAMP_HEADER, String.valueOf(System  
  94.                             .currentTimeMillis()));  
  95.                 }  
  96.                 events.add(EventBuilder.withBody(  
  97.                         body.getTextContent().getBytes(  
  98.                                 httpServletRequest.getCharacterEncoding()),  
  99.                         eventHeaders));  
  100.             }  
  101.         } catch (SAXException ex) {  
  102.             throw new HTTPBadRequestException(  
  103.                     "Request could not be parsed into valid XML", ex);  
  104.         } catch (Exception ex) {  
  105.             throw new HTTPBadRequestException(  
  106.                     "Request is not in expected format. " +  
  107.                             "Please refer documentation for expected format.", ex);  
  108.         }  
  109.         return events;  
  110.     }  
  111.   
  112.     public void configure(Context context) {  
  113.         insertTimestamp = context.getBoolean(CONF_INSERT_TIMESTAMP,  
  114.                 false);  
  115.     }  
  116. }  

打包成dependency，然后放到flume的lib下。

3）flume配置文件：

[html] view plain copy
 
  1. a1.sources=r1  
  2. a1.sinks=k1  
  3. a1.channels=c1  
  4.   
  5. a1.sources.r1.type=http  
  6. a1.sources.r1.bind=localhost  
  7. a1.sources.r1.port=50000  
  8. a1.sources.r1.channels=c1  
  9. a1.sources.r1.handler=org.pq.flumeDemo.sources.HTTPSourceXMLHandler  
  10. a1.sources.r1.insertTimestamp=true  
  11.   
  12. a1.sinks.k1.type=logger  
  13. a1.sinks.k1.channel=c1  
  14.   
  15. a1.channels.c1.type=memory  
  16. a1.channels.c1.capacity=1000  
  17. a1.channels.c1.transactionCapacity=100  

4）启动：

[html] view plain copy
 
    1. $ bin/flume-ng agent -c conf -f conf/http_test.conf  -n a1 -Dflume.root.logger=INFO,console  

标签：#flume#:#appender#文件#handler | 浏览：268 | 发布日期：

免责声明：文章转载自《flume使用之httpSource》仅用于学习参考。如对内容有疑问，请及时联系本站处理。

上篇3.深入k8s：Deployment控制器promise使用场景下篇

宿迁高防，2C2G15M，22元/月；香港BGP，2C5G5M，25元/月 雨云优惠码：MjYwNzM=

相关文章

ECNUOJ 2147 字符环

字符环 Time Limit:1000MS Memory Limit:65536KBTotal Submit:562 Accepted:146 Description 字符环:就是将给定的一个字符串，首尾相连而形成的一个环。现在我们来玩一个有趣的两人游戏:每次甲给出两个字符串A和B,然后让乙判断字符串B是否能够被字符环A所包含.例如,A = aabcd ,...

使用jest进行单元测试

以前，写完一段代码我也是直接调用或者实例化一下，发现过了就把测试相关部分删了。今年的不幸与坎坷使我有很长一段时间去思考人生，不想将就了，鲁棒健壮的程序，开发和测试应该是分得很开的，于是我选择jest去做单元测试这件事。 为什么要做单元测试 在开始之前，我们先思考这样一个问题，我们为什么要做单元测试？ 不扯犊子直接说吧，第一点，用数据、用茫茫多的测试用例去告...

rsync常用命令及格式

rsync在同步文件夹内容这个工作上应用非常广泛，但是rsync本身命令还是比较复杂，本文总结一下： rsync = remote sync的简称 ，它 被用于在linux/unix系统中执行备份操作。rsnync用于从一个位置到另外一个位置同步文件和文件夹。备份的地址可以是本地也可以是remote server。 rsync的重要功能： speed 首次...

用UIButton实现页面跳转（AppDelegate ＋NavigationViewController）

1.AppDelegate.h 定义一个UINavigationController  #import <UIKit/UIKit.h> #import "MainViewController.h"   @interface AppDelegate : UIResponder <UIApplicationDelegate>   @pr...

关于SVN提交强制加入注释

一、摘要 场景：     在这次开发项目过程中，团队中总是有人忘记添加注释。 问题：     1：其他成员不知道你提交是什么代码, 给回滚操作带来很多不必要的麻烦。     2：这个工作需要有一个人天天提醒大家在提交代码的时候加入注释。     3：这个是重复的，定时的，固定的 工作。 思路：     1：我们可以把这项工作交给电脑来完成岂不是更好。  ...

Linux上的ftp服务器 vsftpd 之配置满天飞--设置匿名用户访问(不弹出用户名密码框）以及其他用户可正常上传

一、问题背景 没事谁折腾这鬼玩意哦。。。还不是因为bug。 我们的应用，用户头像是存在ftp上的。之前的ftp服务器是一台windows，我们后台服务器程序收到用户上传头像的请求时，会用一个ROOT/ROOT的账户，连接ftp服务器。 获取到连接后，即进行上传。 上传：上传文件则存放在ROOT用户的home 目录（在windows上装的是server-U来...

最新文章

随机推荐

思享工具箱导航