Seal^_^头像
关注
Flume 多维数据源采集实战：数据库、日志与埋点的统一接入之道封面图

Flume 多维数据源采集实战：数据库、日志与埋点的统一接入之道

Flume 多维数据源采集实战：数据库、日志与埋点的统一接入之道


1. Flume 架构概述与多维数据源接入意义


Apache Flume 是一个高可用、高可靠、分布式的海量日志采集、聚合和传输的系统，专为日志收集中设计。在企业级数据中台建设过程中，通常需要从多种异构数据源采集数据，如数据库变更日志、服务器系统日志、应用程序埋点数据等。通过 Flume 的统一接入能力，可以实现不同类型数据的标准化采集，简化数据管道架构，提高数据处理效率。


Flume 的核心概念包括：

  • Agent：一个独立的 Flume 进程，包含 Source、Channel 和 Sink 三大组件
  • Source：数据收集组件，从数据源采集数据
  • Channel：数据传输组件，连接 Source 和 Sink
  • Sink：数据发送组件，将数据写入目的地


在实际应用中，通过合理配置这三个组件，可以实现多种数据源的统一接入与处理。


2. 数据库 Binlog 采集配置与实战


MySQL 数据库的 Binlog 记录了所有更改数据的 SQL 语句，是数据变更审计和实时数据同步的重要来源。Flume 可以通过 Debezium Source 插件或自定义 MySQL Binlog Source 实现 Binlog 采集。


配置步骤：


  1. 启用 MySQL Binlog：
-- 在 MySQL 配置文件中添加以下内容
[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server-id=1


  1. 创建 Flume 配置文件 (mysql-binlog-flume.conf)：
# 定义名为 mysql-binlog 的源
agent.sources = mysql-binlog

# 配置 mysql-binlog 源
agent.sources.mysql-binlog.type = org.apache.flume.source.exec.ExecSource
agent.sources.mysql-binlog.command = mysqlbinlog --read-from-remote-server --host=127.0.0.1 --port=3306 --user=flume --password=password --raw --stop-never mysql-bin.000001
agent.sources.mysql-binlog.shell = /bin/bash -c
agent.sources.mysql-binlog.batchSize = 1000
agent.sources.mysql-binlog.channels = memory-channel

# 定义内存通道
agent.channels = memory-channel
agent.channels.memory-channel.type = memory
agent.channels.memory-channel.capacity = 10000

# 定义 Kafka Sink
agent.sinks = kafka-sink
agent.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka-sink.brokerList = localhost:9092
agent.sinks.kafka-sink.topic = binlog-topic
agent.sinks.kafka-sink.channel = memory-channel
agent.sinks.kafka-sink.requiredAcks = 1
agent.sinks.kafka-sink.batchSize = 1000


  1. 启动 Flume Agent：
flume-ng agent --conf ./conf --conf-file ./mysql-binlog-flume.conf --name agent -Dflume.root.logger=INFO,console


关键点说明：

  • 使用 mysqlbinlog 命令直接读取 MySQL 的 Binlog
  • 通过内存通道作为中间缓冲，平衡数据采集速率和写入速率
  • 最终将数据写入 Kafka，实现高吞吐和持久化存储


3. 系统日志与应用埋点采集实现


3.1 系统日志采集


系统日志（如 Nginx 访问日志、系统日志等）通常采用文件 Source 采集：


配置示例 (syslog-flume.conf)：

# 定义 source
agent.sources = syslog-source

# 配置 syslog source
agent.sources.syslog-source.type = exec
agent.sources.syslog-source.command = tail -F /var/log/nginx/access.log
agent.sources.syslog-source.channels = memory-channel

# 定义通道
agent.channels = memory-channel
agent.channels.memory-channel.type = memory
agent.channels.memory-channel.capacity = 10000

# 定义 HDFS Sink
agent.sinks = hdfs-sink
agent.sinks.hdfs-sink.type = hdfs
agent.sinks.hdfs-sink.hdfs.path = hdfs://namenode:8020/logs/%Y%m%d/%H
agent.sinks.hdfs-sink.hdfs.fileType = DataStream
agent.sinks.hdfs-sink.hdfs.writeFormat = Text
agent.sinks.hdfs-sink.hdfs.rollInterval = 3600
agent.sinks.hdfs-sink.hdfs.rollSize = 134217728
agent.sinks.hdfs-sink.hdfs.rollCount = 0
agent.sinks.hdfs-sink.channel = memory-channel


3.2 应用埋点采集


应用埋点数据（如 JSON 格式的用户行为数据）可以通过 HTTP Source 接收：


配置示例 (app-metrics-flume.conf)：

# 定义 source
agent.sources = http-source

# 配置 HTTP source
agent.sources.http-source.type = org.apache.flume.http.HTTPSource
agent.sources.http-source.port = 5140
agent.sources.http-source.handler = org.apache.flume.http.JSONHandler
agent.sources.http-source.channels = memory-channel
agent.sources.http-source.processor.type = default
agent.sources.http-source.processor.maxThreads = 8

# 定义通道
agent.channels = memory-channel
agent.channels.memory-channel.type = memory
agent.channels.memory-channel.capacity = 10000

# 定义 Elasticsearch Sink
agent.sinks = elasticsearch-sink
agent.sinks.elasticsearch-sink.type = org.apache.flume.sink.elasticsearch.ElasticSearchSink
agent.sinks.elasticsearch-sink.hostNames = elasticsearch:9200
agent.sinks.elasticsearch-sink.indexName = app-metrics
agent.sinks.elasticsearch-sink.indexType = logs
agent.sinks.elasticsearch-sink.channel = memory-channel
agent.sinks.elasticsearch-sink.serializer = org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer


4. 多源数据汇聚与统一处理


当需要将多种数据源汇聚到同一目的地时，可以使用 Flume 的 Interceptor 机制进行数据预处理和统一格式化：


配置多源汇聚 (multi-source-flume.conf)：

# 定义多个 source
agent.sources = binlog-source syslog-source http-source

# 配置 binlog source
agent.sources.binlog-source.type = exec
agent.sources.binlog-source.command = mysqlbinlog --read-from-remote-server --host=127.0.0.1 --port=3306 --user=flume --password=password --raw --stop-never mysql-bin.000001
agent.sources.binlog-source.channels = memory-channel

# 配置 syslog source
agent.sources.syslog-source.type = exec
agent.sources.syslog-source.command = tail -F /var/log/nginx/access.log
agent.sources.syslog-source.channels = memory-channel

# 配置 HTTP source
agent.sources.http-source.type = org.apache.flume.http.HTTPSource
agent.sources.http-source.port = 5140
agent.sources.http-source.handler = org.apache.flume.http.JSONHandler
agent.sources.http-source.channels = memory-channel

# 定义通道
agent.channels = memory-channel
agent.channels.memory-channel.type = memory
agent.channels.memory-channel.capacity = 10000

# 定义 Kafka Sink
agent.sinks = kafka-sink
agent.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka-sink.brokerList = localhost:9092
agent.sinks.kafka-sink.topic = unified-topic
agent.sinks.kafka-sink.channel = memory-channel


添加 Interceptor 进行数据格式化：

# 为每个 source 添加 interceptor
agent.sources.binlog-source.interceptors = i1
agent.sources.binlog-source.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder

agent.sources.syslog-source.interceptors = i1
agent.sources.syslog-source.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder

agent.sources.http-source.interceptors = i1
agent.sources.http-source.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder


使用 Avro 实现多 Agent 级联：

# 在第一级 Agent 中
agent.sources = avro-source
agent.sources.avro-source.type = avro
agent.sources.avro-source.bind = 0.0.0.0
agent.sources.avro-source.port = 41414
agent.sources.avro-source.channels = memory-channel

# 在第二级 Agent 中
agent.sources = exec-source avro-source
agent.sources.exec-source.type = exec
agent.sources.exec-source.command = tail -F /var/log/application.log
agent.sources.exec-source.channels = memory-channel

agent.sources.avro-source.type = avro
agent.sources.avro-source.bind = 0.0.0.0
agent.sources.avro-source.port = 41414
agent.sources.avro-source.channels = memory-channel


5. 完整示例与关键注意事项


完整的多源采集配置示例：

# 定义 sources
agent.sources = binlog-source syslog-source http-source

# 配置 binlog source
agent.sources.binlog-source.type = exec
agent.sources.binlog-source.command = mysqlbinlog --read-from-remote-server --host=127.0.0.1 --port=3306 --user=flume --password=password --raw --stop-never mysql-bin.000001
agent.sources.binlog-source.interceptors = i1
agent.sources.binlog-source.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
agent.sources.binlog-source.channels = memory-channel

# 配置 syslog source
agent.sources.syslog-source.type = exec
agent.sources.syslog-source.command = tail -F /var/log/nginx/access.log
agent.sources.syslog-source.interceptors = i1
agent.sources.syslog-source.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
agent.sources.syslog-source.channels = memory-channel

# 配置 HTTP source
agent.sources.http-source.type = org.apache.flume.http.HTTPSource
agent.sources.http-source.port = 5140
agent.sources.http-source.handler = org.apache.flume.http.JSONHandler
agent.sources.http-source.interceptors = i1
agent.sources.http-source.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
agent.sources.http-source.channels = memory-channel

# 定义通道
agent.channels = memory-channel
agent.channels.memory-channel.type = memory
agent.channels.memory-channel.capacity = 10000

# 定义 Kafka Sink
agent.sinks = kafka-sink
agent.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka-sink.brokerList = localhost:9092
agent.sinks.kafka-sink.topic = unified-topic
agent.sinks.kafka-sink.channel = memory-channel


注意事项：

  1. 内存使用：内存通道容量设置需考虑可用内存，避免溢出
  2. 背压处理：当 Sink 无法及时处理时，需要合理配置 Channel 和 Source 的参数
  3. 数据格式统一：使用 Interceptor 统一不同数据源的时间戳和格式
  4. 高可用性：通过配置多个 Agent 和负载均衡实现高可用
  5. 监控告警：配置 JMX 监控 Agent 运行状态，及时发现问题
  6. 数据清洗：可在 Source 和 Sink 之间添加自定义 Channel Processor 进行数据清洗
  7. 批量处理：合理设置批量处理参数，平衡实时性和吞吐量


最小可直接运行示例：

# 简单的日志采集到控制台
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Source 配置
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /var/log/syslog

# Sink 配置
a1.sinks.k1.type = logger

# Channel 配置
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000

# 绑定 Source 和 Channel 到 Sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1


执行命令：

flume-ng agent --conf ./conf --conf-file ./simple-flume.conf --name a1 -Dflume.root.logger=INFO,console


MySQL数据库
Binlog

Flume Binlog Source

系统日志文件

Flume Exec Source

应用埋点API

Flume HTTP Source

Memory Channel

Kafka Sink

Kafka集群

实时数据处理引擎

数据仓库

数据湖


Flume 多维数据源采集是企业数据平台建设的重要环节，通过合理配置可以实现高效、可靠的数据采集。在实际应用中，需要根据业务需求调整配置参数，并结合监控和告警机制确保数据管道的稳定运行。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/qq_41840843/article/details/164194622

文章来源转载

评论

赞0

评论列表

微信小程序
QQ小程序

关于作者

点赞数：0
关注数：0
粉丝：0
文章：0
关注标签：0
加入于：--