6、Flume安装部署

news2024/11/15 15:31:50

按照采集通道规划，需在hadoop102，hadoop103，hadoop104三台节点分别部署一个Flume。可参照以下步骤先在hadoop102安装，然后再进行分发。

1、Flume入门

1.1、 Flume安装部署

1.1.1、安装地址

（1） Flume官网地址：Welcome to Apache Flume — Apache Flume

（2）文档查看地址：Flume 1.11.0 User Guide — Apache Flume

（3）下载地址：Index of /dist/flume

1.1.2、安装部署

（1）将apache-flume-1.9.0-bin.tar.gz上传到linux的/opt/software目录下

（2）解压apache-flume-1.9.0-bin.tar.gz到/opt/module/目录下

[shuidi@hadoop102 software]$ tar -zxf /opt/software/apache-flume-1.9.0-bin.tar.gz -C /opt/module/

（3）修改apache-flume-1.9.0-bin的名称为flume

[shuidi@hadoop102 module]$ mv /opt/module/apache-flume-1.9.0-bin /opt/module/flume

（4）将lib文件夹下的guava-11.0.2.jar删除以兼容Hadoop 3.1.3

[shuidi@hadoop102 module]$ rm /opt/module/flume/lib/guava-11.0.2.jar

注意：删除guava-11.0.2.jar的服务器节点，一定要配置hadoop环境变量。否则会报如下异常。

Caused by: java.lang.ClassNotFoundException: com.google.common.collect.Lists
        at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
        ... 1 more

（5）修改conf目录下的log4j.properties配置文件，配置日志文件路径

[shuidi@hadoop102 conf]$ vim log4j.properties

flume.log.dir=/opt/module/flume/logs

1.2、分发Flume

[shuidi@hadoop102 ~]$ xsync /opt/module/flume/

1.3、项目经验

（1）堆内存调整

Flume堆内存通常设置为4G或更高，配置方式如下：

修改/opt/module/flume/conf/flume-env.sh文件，配置如下参数（虚拟机环境暂不配置）

export JAVA_OPTS="-Xms4096m -Xmx4096m -Dcom.sun.management.jmxremote"
注：
-Xms表示JVM Heap(堆内存)最小尺寸，初始分配；
-Xmx 表示JVM Heap(堆内存)最大允许的尺寸，按需分配。

2、日志采集Flume

2.1、日志采集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，提高了效率。

日志采集Flume关键配置如下：

3、日志采集Flume配置实操

1）创建Flume配置文件

在hadoop102节点的Flume的job目录下创建file_to_kafka.conf

[shuidi@hadoop104 flume]$ mkdir job
[shuidi@hadoop104 flume]$ vim job/file_to_kafka.conf

2）配置文件内容如下

#定义组件
a1.sources = r1
a1.channels = c1

#配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/module/applog/log/app.*
a1.sources.r1.positionFile = /opt/module/flume/taildir_position.json
a1.sources.r1.interceptors =  i1
a1.sources.r1.interceptors.i1.type = com.atguigu.gmall.flume.interceptor.ETLInterceptor$Builder

#配置channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092
a1.channels.c1.kafka.topic = topic_log
a1.channels.c1.parseAsFlumeEvent = false

#组装 
a1.sources.r1.channels = c1

3）编写Flume拦截器

（1）创建Maven工程flume-interceptor

（2）创建包：com.atguigu.gmall.flume.interceptor

（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>

（4）在com.atguigu.gmall.flume.utils包下创建JSONUtil类

package com.atguigu.gmall.flume.utils;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONException;

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

（5）在com.atguigu.gmall.flume.interceptor包下创建ETLInterceptor类

package com.atguigu.gmall.flume.interceptor;

import com.atguigu.gmall.flume.utils.JSONUtil;
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 ETLInterceptor implements Interceptor {

    @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 (JSONUtil.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 ETLInterceptor();
        }
        @Override
        public void configure(Context context) {

        }

    }

    @Override
    public void close() {

    }
}

（6）打包

（7）需要先将打好的包放入到hadoop102的/opt/module/flume/lib文件夹下面。

4、日志采集Flume测试

1）启动Zookeeper、Kafka集群

2）启动hadoop102的日志采集Flume

[shuidi@hadoop102 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/file_to_kafka.conf -Dflume.root.logger=info,console

3）启动一个Kafka的Console-Consumer

[shuidi@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_log

4）生成模拟数据

[shuidi@hadoop102 ~]$ lg.sh

5）观察Kafka消费者是否能消费到数据

5、日志采集Flume启停脚本

1）分发日志采集Flume配置文件和拦截器

若上述测试通过，需将hadoop102节点的Flume的配置文件和拦截器jar包，向另一台日志服务器发送一份。

[shuidi@hadoop102 flume]$ scp -r job hadoop103:/opt/module/flume/
[shuidi@hadoop102 flume]$ scp lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar hadoop103:/opt/module/flume/lib/

2）方便起见，此处编写一个日志采集Flume进程的启停脚本

在hadoop102节点的/home/shuidi/bin目录下创建脚本f1.sh

[shuidi@hadoop102 bin]$ vim f1.sh

在脚本中填写如下内容

#!/bin/bash

case $1 in
"start"){
        for i in hadoop102 hadoop103
        do
                echo " --------启动 $i 采集flume-------"
                ssh $i "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf/ -f /opt/module/flume/job/file_to_kafka.conf >/dev/null 2>&1 &"
        done
};; 
"stop"){
        for i in hadoop102 hadoop103
        do
                echo " --------停止 $i 采集flume-------"
                ssh $i "ps -ef | grep file_to_kafka | grep -v grep |awk  '{print \$2}' | xargs -n1 kill -9 "
        done

};;
esac

3）增加脚本执行权限