【Spark七十五】Spark Streaming整合Flume-NG三之接入log4j
编程技术 / houtizong 发布于 3年前 95
先来一段废话：
实际工作中，业务系统的日志基本上是使用Log4j写入到日志文件中的，问题的关键之处在于业务日志的格式混乱，这给对日志文件中的日志进行统计分析带来了极大的困难，或者说，基本上无法进行分析，每个人写日志的习惯不同，导致日志行的格式五花八门，最后只能通过grep来查找特定的关键词缩小范围，但是在集群环境下，每个机器去grep一遍，分析一遍，这个效率如何可想之二，大好光阴都浪费在这上面了。
更合理的做法时，对重要日志进行统计分析，写入关系型数据库或者NoSQL数据库，一方面将重要的日志整合到一起，同时使用这些数据库的查询能力快速的找到相关的日志。
这就涉及到一个日志格式的问题，对于需要进行统计分析的日志，应该使用专门的logger以及appender，这里就是使用FlumeAppender，将日志发送到Flume的输入源，然后经过Channel和Sink进入处理和分析的环节中。另一方面，针对这种的日志，需要根据业务的分析目标，严格定义其结构。
本文分析使用log4j将业务产生的日志通过FlumeAppender写到Flume的日志输入源(source)，最后流出到Spark Streaming，交由Spark Streaming
1. log4j配置
###日志名称和级别log4j.rootLogger=INFO,Flume####未log4j定义的Flume专用Appender类log4j.appender.Flume=org.apache.flume.clients.log4jappender.Log4jAppender###将数据发往localhost的19999端口，此端口由Flume的一个Agent监听，该Agent接收Flume发送过来的数据log4j.appender.Flume.Hostname= localhostlog4j.appender.Flume.Port=19999log4j.appender.Flume.UnsafeMode=false###输出格式log4j.appender.Flume.layout=org.apache.log4j.PatternLayoutlog4j.appender.Flume.layout.ConversionPattern=%d{ABSOLUTE} %-5p [%c] %m% 2. 应用程序依赖的jar
这里的应用程序不是指Spark提交的程序，而是指的是使用log4j输出日志的业务系统，因为业务系统使用了Flume的专用Appender，因此需要把这些依赖的jar加到classpath上
avro-1.7.3.jar jackson-mapper-asl-1.9.3.jar slf4j-api-1.6.1.jaravro-ipc-1.7.3.jar flume-ng-core-1.5.2.jar slf4j-log4j12-1.6.1.jarcommons-collections-3.2.1.jar flume-ng-log4jappender-1.5.2.jar log4j-1.2.17.jarcommons-lang-2.5.jar flume-ng-sdk-1.5.2.jar commons-logging-1.1.1.jar jackson-core-asl-1.9.3.jar netty-3.5.12.Final.jar
3. Flume配置
Flume的配置与Spark Streaming与之前的处理一样，Flume Agent的source监听于19999端口，Spark Streaming的Worker Thread监听于9999端口，Flume Agent的sink往9999端口写入数据(或者直接写到KafkaSink，Spark Streaming从Kafka读取数据)，
通过上面的配置可以看出来，Flume的各个组件时独立的，可以任意的搭配，使用Flume的Log4j Appender仅仅改变了Flume获取数据源的方式，获取到数据后，之前的操作都是一样的
问题：
Flume的source使用avro的方式从19999获取数据，而数据是通过Log4j Appender写入到19999端口的，之前是使用avro client的方式将数据写入到19999端口的，Log4jAppender输入的数据格式和avro client输入的数据一样吗？即两种方式写入到19999端口，能否被Flume source所识别。从上面可以看到Flume的Log4jAppender依赖于avro和avro ipc库，因此有理由相信，Flume的Log4jAppender也是采用类似avro-client的方式，以avro方式将数据进行包装后写到19999中的。实验验证也确实如此
a1.sources = r1a1.sinks = k1a1.channels = c1a1.sources.r1.type = avroa1.sources.r1.channels = c1a1.sources.r1.bind = 0.0.0.0a1.sources.r1.port = 19999a1.sinks = k1a1.sinks.k1.type = avroa1.sinks.k1.hostname = localhosta1.sinks.k1.port = 9999a1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1a1.sinks.k1.channel = c1
4. 验证
写一个java程序， 定时的写日志，然后Flume的Log4j Appender将数据发送到19999端口，作为Flume的输入源，Flume通过sink将数据写到9999端口，这正是Spark Streaming监听的端口，Spark Streaming读取到数据后，即可进行分析
4.1 Java代码
package com.tom.flume.log4j.Example;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;public class FlumeLog4j {private static Log LOG = LogFactory.getLog(FlumeLog4j.class);public static void main(String[] args) {int loop = 60;int interval = 1000;if (args != null && args.length > 0) {interval = Integer.parseInt(args[0]);}if (args != null && args.length > 1) {loop = Integer.parseInt(args[1]);}try {int i = 0;while (i++ < loop) {System.out.println(i);LOG.info("This is the log " + i); //Spark Streaming收到这个日志Thread.sleep(interval); //暂停interval毫秒}} catch (Exception e) {e.printStackTrace();}}} 4.2 log4j.properties
log4j.rootLogger=INFO,Flumelog4j.appender.Flume=org.apache.flume.clients.log4jappender.Log4jAppenderlog4j.appender.Flume.Hostname= localhostlog4j.appender.Flume.Port=19999log4j.appender.Flume.UnsafeMode=falselog4j.appender.Flume.layout=org.apache.log4j.PatternLayoutlog4j.appender.Flume.layout.ConversionPattern=%d{ABSOLUTE} %-5p [%c] %m% 4.3 程序启动脚本launch.sh
java -classpath ".:./*" com.tom.flume.log4j.Example.FlumeLog4j
将log4j.properties以及前面提到的14个jar以及FlumeLog4j这个类所打成的jar包放到launch.sh的同一个目录下
5.运行
5.1 启动Spark Streaming，监听于9999
5.2 启动Flume Agent a1,监听于19999等待数据输入作为数据源
5.3 通过launch.sh启动java程序，想19999端口写入数据
5.4 Flume接收到来自19999端口的写入数据后，通过sink写向9999，Spark Streaming接收到数据，注意Spark Streaming接收的数据格式为
02:01:49,255 INFO [com.tom.flume.log4j.Example.FlumeLog4j] This is the log 34
可见Log4j根据appender的PatternLayout加了一些前缀，需要根据需要决定是否需要这个，需要的话就需要额外的解析工作。
请勿发布不友善或者负能量的内容。与人为善，比聪明更重要！
技术博客集 - 网站简介：
前后端技术：
后端基于Hyperf2.1框架开发,前端使用Bootstrap可视化布局系统生成
网站主要作用：
1.编程技术分享及讨论交流，内置聊天系统;
2.测试交流框架问题，比如：Hyperf、Laravel、TP、beego;
3.本站数据是基于大数据采集等爬虫技术为基础助力分享知识，如有侵权请发邮件到站长邮箱，站长会尽快处理;
4.站长邮箱：[email protected];
文章归档
文章标签
友情链接
