【大数据环境准备】（七）flume 采集

按照规划，需要采集的用户行为日志文件分布在hadoop102，hadoop103两台日志服务器，故需要在hadoop102，hadoop103两台节点配置日志采集Flume。日志采集Flume需要采集日志文件内容，并对日志格式（JSON）进行校验，然后将校验通过的日志发送到Kafka。
此处可选择TaildirSource和KafkaChannel，并配置日志校验拦截器。
选择TailDirSource和KafkaChannel的原因如下：

1）TailDirSource

TailDirSource相比ExecSource、SpoolingDirectorySource的优势

TailDirSource：断点续传、多目录。Flume1.6以前需要自己自定义Source记录每次读取文件位置，实现断点续传。

ExecSource可以实时搜集数据，但是在Flume不运行或者Shell命令出错的情况下，数据将会丢失。

SpoolingDirectorySource监控目录，支持断点续传。

2）KafkaChannel

采用Kafka Channel，省去了Sink，提高了效率

创建拦截器

（1）创建Maven工程flume-interceptor
（2）创建包：org.apache.flume.interceptor HostInterceptor

（3）在pom.xml文件中添加如下配置

<dependencies>
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-core</artifactId>
        <version>1.9.0</version>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.62</version>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>2.3.2</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

拦截代码

package org.apache.flume.interceptor;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;

public class HostInterceptor implements Interceptor {
   
    /*
     * 通过异常判断是否是json字符串
     * 是：返回true  不是：返回false
     * */
    public  boolean isJSONValidate(String log){
   
        try {
   
            JSONObject.parseObject(log);
            return true;
        }catch (JSONException e){
   
            return false;
        }
    }
    @Override
    public void initialize() {
   

    }

    @Override
    public Event intercept(Event event) {
   

        //1、获取body当中的数据并转成字符串
        byte[] body = event.getBody();
        String log = new String(body, StandardCharsets.UTF_8);
        //2、判断字符串是否是一个合法的json，是：返回当前event；不是：返回null
        if (this.isJSONValidate(log)) {
   
            return event;
        } else {
   
            return null;
        }
    }

    @Override
    public List<Event> intercept(List<Event> list) {
   

        Iterator<Event> iterator = list.iterator();

        while (iterator.hasNext()){
   
            Event next = iterator.next();
            if(intercept(next)==null){
   
                iterator.remove();
            }
        }

        return list;
    }

    public static class Builder implements Interceptor.Builder{
   

        @Override
        public Interceptor build() {
   
            return new HostInterceptor();
        }
        @Override
        public void configure(Context context) {
   

        }

    }

    @Override
    public void close() {
   

    }
}

打包

复制文件

[centos@hadoop10 data]$ /data/xsync /data/module/flume/
[centos@hadoop10 data]$ /data/xsync /data/module/flume/lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar
[centos@hadoop10 data]$ vim f1.sh

#!/bin/bash
case $1 in
"start"){
   
        for i in hadoop10 hadoop11 hadoop12
        do
                echo " --------启动 $i 采集flume-------"
                ssh $i "nohup /data/module/flume/bin/flume-ng agent -n a1 -c /data/module/flume/conf/ -f /data/module/flume/job/file_to_kafka.conf >/dev/null 2>&1 &"
        done
};; 
"stop"){
   
        for i in hadoop10 hadoop11 hadoop12
        do
                echo " --------停止 $i 采集flume-------"
                ssh {
   mathJaxContainer[0]}2}' | xargs -n1 kill -9 "
        done
};;
esac

[centos@hadoop10 data]$ chmod 777 fl.sh 
[centos@hadoop10 data]$ ./fl.sh start