Flume_企业中日志处理

Posted 2020-08-20

企业中的日志存放_1

201611/20161112.log.tmp
　　第二天文件变为20161112.log与20161113.log.tmp
拷贝一份flume-conf.properties.template改名为dir-mem-hdfs.properties
实现监控某一目录,如有新文件产生则上传至hdfs,另外过滤掉新文件中tmp文件
dir-mem-hdfs.properties
　　a1.sources = s1
　　a1.channels = c1
　　a1.sinks = k1
　　# defined the source
　　a1.sources.s1.type = spooldir
　　a1.sources.s1.spoolDir = /opt/data/log_hive/20161109
　　a1.sources.s1.includePattern = ([^ ]*\.log$) # 包含某些字段
　　a1.sources.s1.ignorePattern = ([^ ]*\.tmp$)  # 忽略某些字段
　　# defined the channel
　　a1.channels.c1.type = memory
　　a1.channels.c1.capacity = 1000
　　a1.channels.c1.transactionCapacity = 1000
　　# defined the sink
　　a1.sinks.k1.type = hdfs
　　a1.sinks.k1.hdfs.useLocalTimeStamp = true
　　a1.sinks.k1.hdfs.path = /flume/spdir
　　a1.sinks.k1.hdfs.fileType = DataStream 
　　a1.sinks.k1.hdfs.rollInterval = 0
　　a1.sinks.k1.hdfs.rollSize = 20480
　　a1.sinks.k1.hdfs.rollCount = 0
　　# The channel can be defined as follows.
　　a1.sources.s1.channels = c1
　　a1.sinks.k1.channel = c1
flmue目录下执行
　　bin/flume-ng agent -c conf/ -n a1 -f conf/dir-mem-hdfs.properties -Dflume.root.logger=INFO,console
　　这里使用了memory channel,可以使用file channel更加安全

 企业中的日志存放_2

201611/20161112.log
　　第二天文件继续往20161112.log写
这样,既要使用exec和spoolingdir,如何处理
编译flume1.7版tail dir source,并集成到我们已有的flume环境
　　1. window上下载安装git 
　　2. 在某个目录下加一个空的文件夹(文件夹路径尽量不要有中文),例GitHub
　　3. 使用github常用命令
　　　　$ pwd
　　　　$ ls
　　　　$ cd /C/Users/Administrator/Desktop/GitHub
　　　　$ git clone (https|git)://github.com/apache/flume.git
　　　　$ cd flume
　　　　$ git branch -r # 查看有哪些分支
　　　　$ git branch -r # 查看当前属于哪个分支
　　　　$ git checkout origin/flume-1.7 #别换分支
　　拷贝flume\flume-ng-sources\flume-taildir-source
　　使用eclipse导入flume-taildir-source项目
　　修改pom.xml
　　<repositories>
　　　　<repository>
　　　　　　<id>cloudera</id>
　　　　　　<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
　　　　</repository>
　　</repositories>
　　<modelVersion>4.0.0</modelVersion>
　　<groupId>org.apache.flume.flume-ng-sources</groupId>
　　<artifactId>flume-taildir-source</artifactId>
　　<version>1.5.0-cdh5.3.6</version>
　　<name>Flume Taildir Source</name>
　　<build>
　　　　<plugins>
　　　　　　<plugin>
　　　　　　　　<groupId>org.apache.maven.plugins</groupId>
　　　　　　　　<artifactId>maven-compiler-plugin</artifactId>
　　　　　　　　<version>2.3.2</version>
　　　　　　　　<configuration>
　　　　　　　　　　<source>1.7</source>
　　　　　　　　　　<target>1.7</target>
　　　　　　　　</configuration>
　　　　　　</plugin>
　　　　</plugins>
　　</build>
　　<dependencies>
　　　　<dependency>
　　　　　　<groupId>org.apache.flume</groupId>
　　　　　　<artifactId>flume-ng-core</artifactId>
　　　　　　<version>1.5.0-cdh5.3.6</version>
　　　　</dependency>
　　　　<dependency>
　　　　　　<groupId>junit</groupId>
　　　　　　<artifactId>junit</artifactId>
　　　　　　<version>4.10</version>
　　　　　　<scope>test</scope>
　　　　</dependency>
　　</dependencies>
　　4. MAVEN_BULID项目,获取jar包并放到当前flume的环境中(lib目录) 
　　5. 创建文件夹和文件
　　　　$ mkdir -p /opt/cdh-5.6.3/apache-flume-1.5.0-cdh5.3.6-bin/position
　　　　$ mkdir -p /opt/data/tail/hadoop-dir/
　　　　$ echo "" > /opt/data/tail/hadoop.log
　　　　拷贝一份flume-conf.properties.template改名为tail-mem-hdfs.properties
　　　　可从源码看出需要的参数
　　　　　　a1.sources = s1
　　　　　　a1.channels = c1
　　　　　　a1.sinks = k1
　　　　　　# defined the source
　　　　　　a1.sources.s1.type = org.apache.flume.source.taildir.TaildirSource
　　　　　　a1.sources.s1.positionFile = /opt/cdh-5.6.3/apache-flume-1.5.0-cdh5.3.6-bin/position/taildir_position.json
　　　　　　a1.sources.s1.filegroups = f1 f2
　　　　　　a1.sources.s1.filegroups.f1 = /opt/data/tail/hadoop.log
　　　　　　a1.sources.s1.filegroups.f2 = /opt/data/tail/hadoop-dir/.*
　　　　　　a1.sources.s1.headers.f1.headerKey1 = value1
　　　　　　a1.sources.s1.headers.f2.headerKey1 = value2-1
　　　　　　a1.sources.s1.headers.f2.headerKey2 = value2-2
　　　　　　a1.sources.s1.fileHeader = true
　　　　　　# defined the channel
　　　　　　a1.channels.c1.type = memory
　　　　　　a1.channels.c1.capacity = 1000
　　　　　　a1.channels.c1.transactionCapacity = 1000
　　　　　　# defined the sink
　　　　　　a1.sinks.k1.type = hdfs
　　　　　　a1.sinks.k1.hdfs.useLocalTimeStamp = true
　　　　　　a1.sinks.k1.hdfs.path = /flume/spdir
　　　　　　a1.sinks.k1.hdfs.fileType = DataStream 
　　　　　　a1.sinks.k1.hdfs.rollInterval = 0
　　　　　　a1.sinks.k1.hdfs.rollSize = 20480
　　　　　　a1.sinks.k1.hdfs.rollCount = 0
　　　　　　# The channel can be defined as follows.
　　　　　　a1.sources.s1.channels = c1
　　　　　　a1.sinks.k1.channel = c1
　　flmue目录下执行
　　　　bin/flume-ng agent -c conf/ -n a1 -f conf/tail-mem-hdfs.properties -Dflume.root.logger=INFO,console
　　　　测试文件或新数据

 企业中常用架构 Flume多sink

同一份数据采集到不同框架处理
采集source: 一份数据
管道channel: 多个
目标sink: 多个
如果多个sink从一个channel取数据将取不完整,而source会针对channel分别发送
设计: source--hive.log channel--file sink--hdfs(不同路径)
拷贝一份flume-conf.properties.template改名为hive-file-sinks.properties
hive-file-sinks.properties
　　a1.sources = s1
　　a1.channels = c1 c2
　　a1.sinks = k1 k2
　　# defined the source
　　a1.sources.s1.type = exec
　　a1.sources.s1.command = tail -F /opt/cdh-5.6.3/hive-0.13.1-cdh5.3.6/logs/hive.log
　　a1.sources.s1.shell = /bin/sh -c
　　# defined the channel 1
　　a1.channels.c1.type = file
　　a1.channels.c1.checkpointDir = /opt/cdh-5.6.3/apache-flume-1.5.0-cdh5.3.6-bin/datas/checkp1
　　a1.channels.c1.dataDirs = /opt/cdh-5.6.3/apache-flume-1.5.0-cdh5.3.6-bin/datas/data1
　　# defined the channel 2
　　a1.channels.c2.type = file
　　a1.channels.c2.checkpointDir = /opt/cdh-5.6.3/apache-flume-1.5.0-cdh5.3.6-bin/datas/checkp2
　　a1.channels.c2.dataDirs = /opt/cdh-5.6.3/apache-flume-1.5.0-cdh5.3.6-bin/datas/data2
　　# defined the sink 1
　　a1.sinks.k1.type = hdfs
　　a1.sinks.k1.hdfs.path = /flume/hdfs/sink1
　　a1.sinks.k1.hdfs.fileType = DataStream 
　　# defined the sink 2
　　a1.sinks.k2.type = hdfs
　　a1.sinks.k2.hdfs.path = /flume/hdfs/sink2
　　a1.sinks.k2.hdfs.fileType = DataStream 
　　# The channel can be defined as follows.
　　a1.sources.s1.channels = c1 c2
　　a1.sinks.k1.channel = c1
　　a1.sinks.k2.channel = c2		
flmue目录下执行
　　bin/flume-ng agent -c conf/ -n a1 -f conf/hive-file-sinks.properties -Dflume.root.logger=INFO,console
hive目录下执行
　　bin/hive -e "show databases"