Flume数据采集

Flume概念

  • Flume 是 Cloudera 提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的软件。

  • Flume 的核心是把数据从数据源(source)收集过来，再将收集到的数据送到指定的目的地(sink)。为了保证输送的过程一定成功，在送到目的地(sink)之前，会先缓存数据(channel),待数据真正到达目的地(sink)后，flume 再删除自己缓存的数据。

  • Flume 支持定制各类数据发送方，用于收集各类型数据；同时，Flume 支持定制各种数据接受方，用于最终存储数据。一般的采集需求，通过对 flume 的简单配置即可实现。针对特殊场景也具备良好的自定义扩展能力。因此，flume 可以适用于大部分的日常数据采集场景。

Flume中的角色

Flume 系统中核心的角色是 agent，agent 本身是一个 Java 进程，一般运行在日志收集节点。

1540347254561

每一个 agent 相当于一个数据传递员，内部有三个组件：

Source：采集源，用于跟数据源对接，以获取数据；

Sink：下沉地，采集数据的传送目的，用于往下一级 agent 传递数据或者往最终存储系统传递数据；

Channel：agent 内部的数据传输通道，用于从 source 将数据传递到 sink；在整个数据的传输的过程中，流动的是 event，它是 Flume 内部数据传输的最基本单元。event 将传输的数据进行封装。如果是文本文件，通常是一行记录， event 也是事务的基本单位。event 从 source，流向 channel，再到 sink，本身为一个字节数组，并可携带 headers(头信息)信息。event 代表着一个数据的最小完整单元，从外部数据源来，向外部的目的地去。

安装部署Flume

前置条件

  • Java 1.7 or later
  • 为sources，channels和sinks提供充足的内存
  • 为channles提供充足的磁盘空间
  • 为agent提供读和写权限

安装过程

  • 上传安装包到数据源所在节点上
  • 然后解压 tar -zxvf apache-flume-1.8.0-bin.tar.gz -C /export/server
  • 然后进入 flume 的目录，修改 conf 下的 flume-env.sh，在里面配置 JAVA_HOME
  • 配置flume环境变量：
1
2
3
4
5
6
7
vim /etc/profile

export FLUME_HOME=/export/server/apache-flume-1.8.0-bin

export PATH=\$FLUME_HOME/bin:$PATH

source /etc/profile
  • 检查是否配置成功：flume-ng version 查看flume版本

根据数据采集需求配置采集方案，描述在配置文件中(文件名可任意自定义)

指定采集方案配置文件，在相应的节点上启动 flume agent

系统架构

简单结构

单个 agent 采集数据

复杂结构

  • 单个agent写入到不同的数据源中

整合Kafka

使用flume的关键就是定义配置文件，根据需求，首先定义以下 3 大要素

  • 采集源，即 source——监控文件 : /root/tmp/user_profile_65/datacollection/source_data/access-nginx-20230114

    • source的配置采集源，即 source——监控文件内容更新 : TAILDIR

  • 下沉目标，即 sink——kafka : KafkaSink

    • sink的配置

      • kafka.producer.acks
      • kafka.producer.compression.type snappy

    • 压缩算法 snappy谷歌开源的压缩算法, 压缩速度快, 压缩率低

  • source 和 sink 之间的传递通道——channel

    • channel的配置
      • capacity：默认该通道中最大的可以存储的 event 数量
      • trasactionCapacity：每次最大可以从 source 中拿到或者送到 sink 中的 event数量

配置文件信息

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
30
31
32
33
34
35
36
37
pro.sources = s1
pro.channels = c1
pro.sinks = k1
# 读取文件类型
pro.sources.s1.type = TAILDIR
# 定义监控的文件组
pro.sources.s1.filegroups = f1 f2
# 配置文件组的路径
pro.sources.s1.filegroups.f1 = /tmp/user_profile_488/com/itheima/online/datacollection/source_data/access-nginx-20230812
# 记录数据读取偏移量
pro.sources.s1.positionFile = /export/data/flume/taildir_position.json
# 每批设置多少条数据发送至channel
pro.sources.s1.batchSize = 10

# 设置channel类型
pro.channels.c1.type = memory
# 最大的缓存数据量
pro.channels.c1.capacity = 1000
# 每次传输的数据量
pro.channels.c1.transactionCapacity = 100

# 设置sink类型
pro.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
# 设置kafka输出的topic
pro.sinks.k1.kafka.topic = tfec_nginx_log
# 设置kafka连接地址
pro.sinks.k1.kafka.bootstrap.servers = up01:9092
# 设置每批传输的数据量
pro.sinks.k1.kafka.flumeBatchSize = 20
# 设置kafka的应答机制
pro.sinks.k1.kafka.producer.acks = 1
# 设置数据的压缩方法
pro.sinks.k1.kafka.producer.compression.type = snappy

# 将source channel sink连接起来
pro.sources.s1.channels = c1
pro.sinks.k1.channel = c1

启动Flume采集

启动采集

在flume 的 conf 目录下运行

1
nohup flume-ng agent -n pro -c ./ -f nginxlog2kafka2.conf &

-n : agent的name
-c：conf目录路径
-f：配置文件路径