我有一壶酒，足以慰平生。

0%

Flume学习笔记

发表于 分类于 阅读次数： Disqus：

Flume

背景

flume是由cloudera软件公司产出的可分布式日志收集系统，后与2009年被捐赠了apache软件基金会，为hadoop相关组件之一。尤其近几年随着flume的不断被完善以及升级版本的逐一推出，特别是flume-ng;同时flume内部的各种组件不断丰富，用户在开发的过程中使用的便利性得到很大的改善，现已成为apache top项目之一。

概述

什么是flume?

Apache Flume 是一个从可以收集例如日志，事件等数据资源，并将这些数量庞大的数据从各项数据资源中集中起来存储的工具/服务，或者数集中机制。flume具有高可用，分布式，配置工具，其设计的原理也是基于将数据流，如日志数据从各种网站服务器上汇集起来存储到HDFS，HBase等集中存储器中。其结构如下图所示：

image-20200730101521762

Flume特性

  1. Flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。
  2. u Flume可以采集文件，socket数据包、文件、文件夹、kafka等各种形式源数据，又可以将采集到的数据(下沉sink)输出到HDFS、hbase、hive、kafka等众多外部存储系统中。
  3. 一般的采集需求，通过对flume的简单配置即可实现。
  4. Flume针对特殊场景也具备良好的自定义扩展能力，因此，flume可以适用于大部分的日常数据采集场景。

Flume原理

Flume组件详解

对于每一个Agent来说,它就是一共独立的守护进程(JVM),它从客户端接收数据，如下图所示flume的基本模型

image-20200730101854263

  1. Flume分布式系统中最核心的角色是**agent**，flume采集系统就是由一个个agent所连接起来形成。
  2. 每一个agent相当于一个数据(被封装成Event对象)传递员，内部有三个组件：
    • Source：采集组件，用于跟数据源对接，以获取数据
    • Sink：下沉组件，用于往下一级agent传递数据或者往最终存储系统传递数据
    • Channel：传输通道组件，用于从source将数据传递到sink

image-20200730102056423

首先来看一下flume官网中对Event的定义

image-20200730105303461

一行文本内容会被反序列化成一个event(*序列化是将对象状态转换为可保持或传输的格式的过程。与序列化相对的是反序列化，它将流转换为对象。这两个过程结合起来，可以轻松地存储和传输数据)*，event的最大定义为2048字节，超过，则会切割，剩下的会被放到下一个event中，默认编码是UTF-8。

官方案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

Flume采集结构图

简单结构

单个agent采集数据

image-20200730105234103

复杂结构

多级agent之间串联

image-20200730105418372

Flume安装部署

下载

官网

安装

1
tar -zxvf apache-flume-1.8.0-bin.tar.gz -C /user/local

配置环境变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
vi /etc/source
#添加如下内容
export FLUME_HOME=/usr/local/apache-flume-1.8.0-bin
export PATH=:$FLUME_HOME/bin
#更新
source /etc/profile

#分发
scp -r /usr/local/apache-flume-1.8.0-bin/  hadoop02:/usr/local
scp -r /usr/local/apache-flume-1.8.0-bin/  hadoop03:/usr/local
scp -r /etc/profile hadoop02:/etc
scp -r /etc/profile hadoop03:/etc

#flume-env.sh 添加JAVA_HOME
vi flume-env.sh

export JAVA_HOME=/usr/local/jdk1.8.0_211

Flume案例

启动命令封装shell脚本

==在bin 目录下创建，在flume根目录下执行==

1
2
3
4
5
6
7
vi start-myconf.sh
#内容如下


#!/bin/bash

bin/flume-ng agent -c conf -f myconf/$1 -n a1 -Dflume.root.logger=info,console

修改权限:

1
chmod 777 start-myconf.sh

执行样例：

1
[root@hadoop01 apache-flume-1.8.0-bin]# start-myconf.sh netcat-logger.conf

简单测试

image-20200730123301813

  1. 创建自己的conf

    1
    mkdir myconf

  2. 创建配置文件

    netcat-logger.conf

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    vi netcat-logger.conf
    #编辑内容如下（来自flume官网）

    # example.conf: A single-node Flume configuration

    # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1

    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = 192.168.88.110
    a1.sources.r1.port = 44444

    # Describe the sink
    a1.sinks.k1.type = logger

    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100

    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

  3. 启动

    1
    [root@hadoop01 apache-flume-1.8.0-bin]# start-myconf.sh netcat-logger.conf

    image-20200730124153050

  4. 测试

    先要往agent的source所监听的端口上发送数据，让agent有数据可采

    随便在一个能跟agent节点联网的机器上

    telnet anget-hostname port （telnet localhost 44444）

    image-20200730124021668

再去启动端可以看到相应的log日志，如下所示：

image-20200730124153050

采集目录到HDFS

image-20200730124517344

采集需求：某服务器的某特定目录下，会不断产生新的文件，每当有新文件出现，就需要把文件采集到HDFS中去

根据需求，首先定义以下3大要素

  • 数据源组件，即source ——监控文件目录 : spooldir

    spooldir特性：

    1. 监视一个目录，只要目录中出现新文件，就会采集文件中的内容

    2. 采集完成的文件，会被agent自动添加一个后缀：COMPLETED

    3. 所监视的目录中不允许重复出现相同文件名的文件

  • 下沉组件，即sink——HDFS文件系统 : hdfs sink

  • 通道组件，即channel——可用file channel 也可以用内存channel

配置文件编写：

spooldir-hdfs.conf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
#vi spooldir-hdfs.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /var/log/apache/flumeSpool
a1.sources.r1.fileHeader = true

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动：

1
[root@hadoop01 apache-flume-1.8.0-bin]# start-myconf.sh spooldir-hdfs.conf

动态给logs文件下添加东西：

1
cp wc.txt logs

在hdfs中查看下沉数据：

image-20200730133148612

当数据完全下沉后，会在数据末尾带上COMPLETED标志

image-20200730133253031

image-20200730113112375

您的支持是我继续创作的动力