Flume 数据采集

  • Post author:
  • Post category:其他



Flume 数据采集


模块


1.


1


数据处理链路


1.2


环境准备


1


.


2


.


1


集群


进程查看


脚本

(1)在/home/bigdata_admin/bin目录下创建脚本xcall.sh


[bigdata_admin@hadoop102


bin]$ vim xcall.sh

(2)在脚本中编写如下内容

#! /bin/bash



for i in hadoop102 hadoop103 hadoop104

do

    echo --------- $i ----------

    ssh $i "$*"

done

(3)修改脚本执行权限


[bigdata_admin@hadoop102 bin


]$ chmod 777 xcall.sh

(4)启动脚本


[bigdata_admin@hadoop102 bin


]$ xcall.sh jps


1


.


2.2


H


adoop


安装



1)安装步骤




2)项目经验

(1)项目经验之HDFS存储多目录

虚拟机的项目不需要配置,我们只有一个磁盘。

1.生产环境服务器磁盘情况

2.在hdfs-site.xml文件中配置多目录,注意新挂载磁盘的访问权限问题。

HDFS的DataNode节点保存数据的路径由

dfs.datanode.data.dir

参数决定,其默认值为

file://${hadoop.tmp.dir}/dfs/data

,若服务器有多个磁盘,必须对该参数进行修改。如服务器磁盘如上图所示,则该参数应修改为如下的值。

<property>

    <name>dfs.datanode.data.dir</name>

<value>file:///dfs/data1,file:///hd2/dfs/data2,file:///hd3/dfs/data3,file:///hd4/dfs/data4</value>

</property>


注意:每台服务器挂载的磁盘不一样,所以每个节点的多目录配置可以不一致。单独配置即可

(2)项目经验之集群数据均衡

1节点间数据均衡

开启数据均衡命令。

start-balancer.sh -threshold 10

对于参数10,代表的是集群中各个节点的磁盘空间利用率相差不超过10%,可根据实际情况进行调整。

停止数据均衡命令。

stop-balancer.sh

2磁盘间数据均衡

生成均衡计划



(我们只有一块磁盘,不会生成计划)。


hdfs diskbalancer -plan hadoop103

执行均衡计划。

hdfs diskbalancer -execute hadoop103.plan.json

查看当前均衡任务的执行情况。

hdfs diskbalancer -query hadoop103

取消均衡任务。

hdfs diskbalancer -cancel hadoop103.plan.json

(3)项目经验之Hadoop参数调优

1.HDFS参数调优hdfs-site.xml

The number of Namenode RPC server threads that listen to requests from clients. If dfs.namenode.servicerpc-address is not configured then Namenode RPC server threads listen to requests from all nodes.

NameNode有一个工作线程池,用来处理不同DataNode的并发心跳以及客户端并发的元数据操作。

对于大集群或者有大量客户端的集群来说,通常需要增大参数dfs.namenode.handler.count的默认值10。

<property>

    <name>dfs.namenode.handler.count</name>

    <value>10</value>

</property>

dfs.namenode.handler.count=20*log(e)(ClusterSize),比如集群规模为8台时,此参数设置为41。可通过简单的python代码计算该值,代码如下。

[bigdata_admin@hadoop102 ~]$

python

Python 2.7.5 (default, Apr 11 2018, 07:36:10)

[GCC 4.8.5 20150623 (Red Hat 4.8.5-28)] on linux2

Type “help”, “copyright”, “credits” or “license” for more information.

>>>

import math

>>>

print int(20*math.log(8))

41

>>>

quit()

2.YARN参数调优yarn-site.xml



情景




描述


:总共7台机器,每天几亿条数据,数据源->Flume->Kafka->HDFS->Hive



面临问题


:数据统计主要用HiveSQL,没有数据倾斜,小文件已经做了合并处理,开启的JVM重用,而且IO没有阻塞,

内存用了不到50%。

但是还是跑的非常慢,而且数据量洪峰过来时,整个集群都会宕掉。基于这种情况有没有优化方案。



解决




办法:

内存利用率不够。这个一般是Yarn的2个配置造成的,单个任务可以申请的最大内存大小,和Hadoop单个节点可用内存大小。调节这两个参数能提高系统内存的利用率。

(a)yarn.nodemanager.resource.memory-mb

表示该节点上YARN可使用的物理内存总量,默认是8192(MB),注意,如果你的节点内存资源不够8GB,则需要调减小这个值,而YARN不会智能的探测节点的物理内存总量。

(b)yarn.scheduler.maximum-allocation-mb

单个任务可申请的最多物理内存量,默认是8192(MB)。


1


.


2.3


Zookeeper安装



1




)安装步骤




2









ZK




集群启动停止脚本

(1)在hadoop102的/home/bigdata_admin/bin目录下创建脚本


[bigdata_admin@hadoop102 bin]$ vim zk.sh

在脚本中编写如下内容。

#!/bin/bash



case $1 in

"start"){

for i in hadoop102 hadoop103 hadoop104

do

        echo ---------- zookeeper $i 启动 ------------

ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"

done

};;

"stop"){

for i in hadoop102 hadoop103 hadoop104

do

        echo ---------- zookeeper $i 停止 ------------    

ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"

done

};;

"status"){

for i in hadoop102 hadoop103 hadoop104

do

        echo ---------- zookeeper $i 状态 ------------    

ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"

done

};;

esac

(2)增加脚本执行权限


[bigdata_admin@hadoop102 bin]$ chmod 777 zk.sh

(3)Zookeeper集群启动脚本


[bigdata_admin@hadoop102 module]$ zk.sh start

(4)Zookeeper集群停止脚本


[bigdata_admin@hadoop102 module]$ zk.sh stop


1


.


2.4


Kafk


a


安装



1




)安装步骤



1


.2


.


5 Flume


安装

按照采集通道规划,需在hadoop102,hadoop103,hadoop104三台节点分别部署一个Flume。



可参照以下步骤先在hadoop






102






安装,然后再进行分发。




1)安装步骤




2)分发Flume到hadoop




103




、hadoop




104

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



3)项目经验

(1)堆内存调整

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

修改/opt/module/flume/conf/flume-env.sh文件,配置如下参数



(虚拟机环境暂不配置)



export JAVA_

OPTS

=”-Xms


4096


m -Xmx


4096


m -Dcom.sun.management.jmxremote”

注:

-Xms表示JVM Heap(堆内存)最小尺寸,初始分配;

-Xmx 表示JVM Heap(堆内存)最大允许的尺寸,按需分配。


1


.


3


日志采集Flume


1.


3


.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)Kafka




Channel


采用Kafka


Channel,


省去了Sink


,提高了


效率。

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


1


.3


.


2


日志采集Flume配置实操



1)创建Flume配置文件

(1)在hadoop102节点的Flume的job目录下创建file_to_kafka.conf。


[bigdata_admin@hadoop102 flume]$ mkdir job


[bigdata_admin@hadoop102 flume]$ vim job/file_to_kafka.conf

配置文件内容如下:

#为各组件命名

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.bigdata_admin.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



#绑定source和channel以及sink和channel的关系

a1.sources.r1.channels = c1

(2)分发配置文件到hadoop103


[bigdata_admin@hadoop102 flume]$ xsync job



2




)编写Flume拦截器

(1)创建Maven工程flume-interceptor

(2)创建包:com.bigdata_admin.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.bigdata_admin.flume.interceptor包下创建JSONUtils类

package com.bigdata_admin.flume.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;

public class JSONUtils {
    public static boolean isJSONValidate(String log){
        try {
            JSON.parse(log);
            return true;
        }catch (JSONException e){
            return false;
        }
    }
}

(5)在com.bigdata_admin.flume.interceptor包下创建ETLInterceptor类

package com.bigdata_admin.flume.interceptor;



import com.alibaba.fastjson.JSON;

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) {



        byte[] body = event.getBody();

        String log = new String(body, StandardCharsets.UTF_8);



        if (JSONUtils.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)打包

flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar

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


1


.3.3


日志采集Flume测试



1)启动Zookeeper、Kafka集群



2)启动hadoop




102




的日志采集Flume

[bigdata_admin@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

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



4




)生成模拟数据

[bigdata_admin@hadoop102 ~]$ lg.sh



5




)观察




Kafka




消费者是否能消费到数据


1


.3.4


日志采集Flume启停脚本



1




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

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

[bigdata_admin@hadoop102 flume]$ scp -r job hadoop103:/opt/module/flume/

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



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

(1)在hadoop102节点的/home/bigdata_admin/bin目录下创建脚本f1.sh

[bigdata_admin@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

(2)增加脚本执行权限


[bigdata_admin@hadoop102 bin]$ chmod 777 f1.sh

(3)f1启动


[bigdata_admin@hadoop102 module]$ f1.sh

start

(4)f2停止


[bigdata_admin@hadoop102 module]$ f1.sh stop


1


.


4


日志消费Flume


1


.4.1


日志消费Flume配置概述

按照规划,该Flume需将Kafka中topic_log的数据发往HDFS。并且对每天产生的用户行为日志进行区分,将不同天的数据发往HDFS不同天的路径。

此处选择KafkaSource、FileChannel、HDFSSink。

关键配置如下:


1


.4.2


日志消费Flume配置实操



1)创建Flume配置文件

在hadoop104节点的Flume的job目录下创建kafka_to_hdfs_log.conf。


[bigdata_admin@hadoop104 flume]$ vim job/kafka_to_hdfs_log.conf

配置文件内容如下:

## 组件

a1.sources=r1

a1.channels=c1

a1.sinks=k1



## source1

a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource

a1.sources.r1.batchSize = 5000

a1.sources.r1.batchDurationMillis = 2000

a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092

a1.sources.r1.kafka.topics=topic_log

a1.sources.r1.interceptors = i1

a1.sources.r1.interceptors.i1.type = com.bigdata_admin.flume.interceptor.TimeStampInterceptor$Builder



## channel1

a1.channels.c1.type = file

a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint/behavior1

a1.channels.c1.dataDirs = /opt/module/flume/data/behavior1/

a1.channels.c1.maxFileSize = 2146435071

a1.channels.c1.capacity = 1000000

a1.channels.c1.keep-alive = 6





## sink1

a1.sinks.k1.type = hdfs

a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d

a1.sinks.k1.hdfs.filePrefix = log-

a1.sinks.k1.hdfs.round = false





a1.sinks.k1.hdfs.rollInterval = 10

a1.sinks.k1.hdfs.rollSize = 134217728

a1.sinks.k1.hdfs.rollCount = 0



## 控制输出文件是原生文件。

a1.sinks.k1.hdfs.fileType = CompressedStream

a1.sinks.k1.hdfs.codeC = gzip



## 拼装

a1.sources.r1.channels = c1

a1.sinks.k1.channel= c1



注:配置优化

(1)FileChannel优化

通过配置

dataDirs


指向多个路径

,每个路径对应不同的硬盘,增大Flume吞吐量。

官方说明如下:

Comma

separated list of directories for storing log files. Using multiple directories on separate disks can improve file channel peformance


checkpointDir和backupCheckpointDir也尽量配置在不同硬盘对应的目录中,

保证checkpoint坏掉后,可以快速使用backupCheckpointDir恢复数据。

(2)HDFS Sink优化

①HDFS存入大量小文件,有什么影响?


元数据层面:

每个小文件都有一份元数据,其中包括文件路径,文件名,所有者,所属组,权限,创建时间等,这些信息都保存在Namenode内存中。所以小文件过多,会占用Namenode服务器大量内存,影响Namenode性能和使用寿命。


计算层面:

默认情况下MR会对每个小文件启用一个Map任务计算,非常影响计算性能。同时也影响磁盘寻址时间。

②HDFS小文件处理

官方默认的这三个参数配置写入HDFS后会产生小文件,hdfs.rollInterval、hdfs.rollSize、hdfs.rollCount。

基于以上hdfs.rollInterval=3600,hdfs.rollSize=134217728,hdfs.rollCount =0几个参数综合作用,效果如下:

  • 文件在达到128M时会滚动生成新文件
  • 文件创建超3600秒时会滚动生成新文件



2




)编写Flume拦截器

(1)在com.bigdata_admin.flume.interceptor包下创建TimeStampInterceptor类

package com.bigdata_admin.flume.interceptor;



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.ArrayList;

import java.util.List;

import java.util.Map;



public class TimeStampInterceptor implements Interceptor {



    private ArrayList<Event> events = new ArrayList<>();



    @Override

    public void initialize() {



    }



    @Override

    public Event intercept(Event event) {



        Map<String, String> headers = event.getHeaders();

        String log = new String(event.getBody(), StandardCharsets.UTF_8);



        JSONObject jsonObject = JSONObject.parseObject(log);



        String ts = jsonObject.getString("ts");

        headers.put("timestamp", ts);



        return event;

    }



    @Override

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

        events.clear();

        for (Event event : list) {

            events.add(intercept(event));

        }



        return events;

    }



    @Override

    public void close() {



    }



    public static class Builder implements Interceptor.Builder {

        @Override

        public Interceptor build() {

            return new TimeStampInterceptor();

        }



        @Override

        public void configure(Context context) {

        }

    }

}

(2)重新打包

flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar

(3)需要先将打好的包放入到hadoop104的/opt/module/flume/lib文件夹下面。


1


.4


.


3


日志消费Flume测试



1)启动Zookeeper、Kafka集群



2)启动日志采集Flume

[bigdata_admin@hadoop102 ~]$ f1.sh start



3)启动hadoop




104




的日志消费Flume

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



4




)生成模拟数据

[bigdata_admin@hadoop102 ~]$ lg.sh



5




)观察




HDFS




是否出现数据


1


.4


.


4


日志消费Flume启停脚本

若上述测试通过,为方便,此处创建一个Flume的启停脚本。



1




)在




hadoop102




节点的/home/bigdata_admin/bin目录




下创建




脚本




f2.sh

[bigdata_admin@hadoop102 bin]$ vim f2.sh

在脚本中填写如下内容:

#!/bin/bash



case $1 in

"start")

        echo " --------启动 hadoop104 日志数据flume-------"

        ssh hadoop104 "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf -f /opt/module/flume/job/kafka_to_hdfs_log.conf >/dev/null 2>&1 &"

;;

"stop")



        echo " --------停止 hadoop104 日志数据flume-------"

        ssh hadoop104 "ps -ef | grep kafka_to_hdfs_log | grep -v grep |awk '{print \$2}' | xargs -n1 kill"

;;

esac



2




)增加脚本执行




权限


[bigdata_admin@hadoop102 bin]$ chmod 777 f2.sh



3









f2




启动


[bigdata_admin@hadoop102 module]$ f2.sh start



4









f2




停止


[bigdata_admin@hadoop102 module]$ f2.sh stop


1


.


5


采集通道启动/停止脚本



1




)在/home/bigdata_admin/bin目录




下创建




脚本




cluster.sh

[bigdata_admin@hadoop102 bin]$ vim cluster.sh

在脚本中填写如下内容:

#!/bin/bash



case $1 in

"start"){

        echo ================== 启动 集群 ==================



        #启动 Zookeeper集群

        zk.sh start



        #启动 Hadoop集群

        cdh.sh start



        #启动 Kafka采集集群

        kf.sh start



        #启动 Flume采集集群

        f1.sh start



        #启动 Flume消费集群

        f2.sh start



        };;

"stop"){

        echo ================== 停止 集群 ==================



        #停止 Flume消费集群

        f2.sh stop



        #停止 Flume采集集群

        f1.sh stop



        #停止 Kafka采集集群

        kf.sh stop



        #停止 Hadoop集群

        cdh.sh stop



#循环直至 Kafka 集群进程全部停止

#xcall.sh 是我们写的脚本,作用是在集群的每个节点都执行一次后面的命令。此处 xcall.sh jps 的作用是查看所有节点的 java 进程

#grep Kafka 的作用是过滤所有 Kafka 进程

#wc -l 是统计行数,每个进程会在 jps 中单独占据一行,因此行数等于进程数

#$()的作用是将括号内命令的执行结果作为值取出来

#因此如下命令的作用是统计集群未停止的 Kafka 进程数然后将进程数赋值给 kafka_count 变量



kafka_count=$(xcall.sh jps | grep Kafka | wc -l)



#判断 kafka_count 变量的值是否大于零,如果是则说明仍有未停止的 Kafka 进程,此时不能停止 Zookeeper,因为 Kafka 的工作要依赖于 Zookeeper 的节点,如果在 Kafka 进程停止之前停止了 Zookeeper,可能会导致本次 Kafka 进程无法正常停止。所以当 Kafka 进程数大于零时进入循环,休眠一秒,然后重新统计 Kafka 进程数,直至 Kafka 进程数为零跳出循环,才能进行下一步(停止 Zookeeper 集群)

while [ $kafka_count -gt 0 ]

do

sleep 1

kafka_count=$( xcall.sh jps | grep Kafka | wc -l)

            echo "当前未停止的 Kafka 进程数为 $kafka_count"

done



        #停止 Zookeeper集群

        zk.sh stop

};;

esac



2




)增加脚本执行




权限


[bigdata_admin@hadoop102 bin]$ chmod u+x cluster.sh



3









cluster




集群启动脚本


[bigdata_admin@hadoop102 module]$ cluster.sh start



4









cluster




集群停止脚本


[bigdata_admin@hadoop102 module]$ cluster.sh stop



版权声明:本文为liuwei0376原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。