Flume_企业中日志处理

时间:2023-07-15 13:26:44

企业中的日志存放_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"