小菜鸟

java菜鸟号正在起航

0%

flume采集kafka消息写入HDFS

发表于 更新于 分类于 阅读次数： 33 Valine： 0
本文字数： 6.2k 阅读时长 ≈ 6 分钟

flume实战：从 Kafka 采集消息写入 HDFS 的完整流程

在大数据架构中，Kafka 常作为实时数据的 “中转站”，而 HDFS 则是海量数据的 “持久化仓库”。Flume 作为连接两者的桥梁，能够高效地将 Kafka 中的消息同步到 HDFS，实现实时数据到离线存储的流转。本文将详细讲解如何配置 Flume，以 Kafka 为数据源、HDFS 为目标存储，完成端到端的数据同步，并解析核心参数与优化技巧。

方案架构：Kafka → Flume → HDFS 的数据流转

本方案的核心是通过 Flume 构建 “Kafka 消费 → 通道缓冲 → HDFS 写入” 的流水线，具体架构如下：

Kafka Source 消费
Kafka Channel 缓冲
HDFS Sink 写入
Kafka 主题 kafka-flume
Flume Agent
Kafka 通道主题 kafka-channel
HDFS 存储 按时间分区
  • Kafka Source：从 Kafka 主题（如 kafka-flume）消费消息；
  • Kafka Channel：使用 Kafka 作为 Flume 的通道，利用 Kafka 的可靠性暂存数据；
  • HDFS Sink：将数据按时间分区（如 %Y-%m-%d/%H）写入 HDFS，便于后续离线分析。

详细配置步骤

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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
# 1. 定义组件名称  
#事件源名称
agent1.sources = kafkaSource
#通道名称
agent1.channels = kafka-channel
#接收器名称
agent1.sinks = hdfsSink


# 2. 配置 Kafka Source（从 Kafka 消费数据）  

###############################事件源source###########################################
## flume-kafka-source-1.9.0.jar
# For each one of the sources, the type is defined
# 使用kafka事件源
agent1.sources.kafkaSource.type = org.apache.flume.source.kafka.KafkaSource
# Kafka 集群地址（多个 broker 用逗号分隔）  
agent1.sources.kafkaSource.kafka.bootstrap.servers = localhost:9092
# 要消费的 Kafka 主题（多个主题用逗号分隔，注意参数名是 topics 带 s！）  
# 注意这里的topic配置是topics  我当初没有写s导致事件源没有启动  数据进不去  找了很长时间问题
agent1.sources.kafkaSource.kafka.topics = kafka-flume
# 消费者组id（确保唯一，避免与其他消费者冲突）
agent1.sources.kafkaSource.kafka.consumer.group.id = kafka-flume-hdfs
# 批量写入的最大数量（一次拉取的消息量，平衡吞吐量和延迟）
agent1.sources.kafkaSource.kafka.batchSize = 1000
# # 批量消费最长等待时间（毫秒），若未达 batchSize 则超时后提交  
agent1.sources.kafkaSource.kafka.batchDurationMillis = 1000

# 3. 配置 Kafka Channel（用 Kafka 作为通道缓冲数据）  

#############################通道channel#############################################
## flume-kafka-channel-1.9.0.jar
agent1.channels.kafka-channel.type = org.apache.flume.channel.kafka.KafkaChannel
agent1.channels.kafka-channel.kafka.bootstrap.servers = localhost:9092
# 通道专用的 Kafka 主题（用于暂存数据，需提前创建） 
agent1.channels.kafka-channel.kafka.topic = kafka-channel
# 消费者组 ID（通道内部使用，需唯一）  
agent1.channels.kafka-channel.kafka.consumer.group.id = flume-channel-group  
# 生产者确认机制（确保数据写入通道主题，可选 1 或 -1）  
agent1.channels.kafka-channel.kafka.producer.acks = 1 


# 4. 配置 HDFS Sink（将数据写入 HDFS）  
##################################接收器sink##############################################
## flume-hdfs-sink-1.9.0.jar
## 采用HDFSEventSink，将KafkaChannel中的数据写入到HDFS中，以日期和小时来切割文件，即每天的每个小时生成一个子目录

agent1.sinks.hdfsSink.type = hdfs
# # HDFS 存储路径（按“年-月-日/小时”分区，便于后续查询） 
agent1.sinks.hdfsSink.hdfs.path = hdfs://localhost:9000/kafka-flume-hdfs/%Y-%m-%d/%H
# 文件前缀（区分不同来源的数据）
agent1.sinks.hdfsSink.hdfs.filePrefix = test
# 临时文件后缀（写入中文件的标识，避免读取未完成文件）  
agent1.sinks.hdfsSink.hdfs.inUseSuffix = .tmp
# 最终文件后缀（写入完成后重命名）  
agent1.sinks.hdfsSink.hdfs.fileSuffix = .txt
# 使用本地时间（默认 UTC，需改为本地时间避免时区偏移）
agent1.sinks.hdfsSink.hdfs.useLocalTimeStamp = true

# 5. HDFS 文件滚动策略（控制文件切割逻辑）  

# 若以时间切割文件时，滚动为目标文件之前的最大时间间隔，单位秒
# 如果为0，则表示不根据时间来滚动文件
# 每小时滚动一次文件（秒）  
agent1.sinks.hdfsSink.hdfs.rollInverval = 3600

# 若以大小切割文件时，滚动为目标文件之前的最多字节数
# 如果为0，则表示不根据临时文件大小来滚动文件
# 单个文件达到 128MB 滚动（字节）
agent1.sinks.hdfsSink.hdfs.rollSize = 134217728

# 配置当事件数据达到该数量时，将临时文件滚动成目标文件
# 如果为0，表示不根据事件数据来滚动文件
# 不按事件数量滚动 
agent1.sinks.hdfsSink.hdfs.rollCount = 0

# # 每 1000 条事件批量写入 HDFS 
agent1.sinks.hdfsSink.hdfs.batchSize = 1000

# 6. 其他 HDFS 配置
# 文件格式，默认为SequenceFile
agent1.sinks.hdfsSink.hdfs.fileType = DataStream

# # 文本写入格式 ，默认Writable
agent1.sinks.hdfsSink.hdfs.writeFormat = Text
# 配置当前被打开的临时文件在该参数指定的时间内，没有任何数据写入时则将该临时文件关闭并重命名为目标文件
agent1.sinks.hdfsSink.hdfs.idleTimeout = 0

# # HDFS 操作线程池大小，默认10
agent1.sinks.hdfsSink.hdfs.threadsPoolSize = 15
# 执行HDFS操作的超时时间，默认10s
agent1.sinks.hdfsSink.hdfs.callTimeout = 60000

# 7. 绑定组件关系（核心！连接 Source → Channel → Sink）
agent1.sinks.hdfsSink.channel = kafka-channel

# The channel can be defined as follows.
# 事件源的通道，绑定通道
agent1.sources.kafkaSource.channels = kafka-channel

核心参数解析

配置中以下参数直接影响数据可靠性和性能，需重点理解：

组件 参数 作用与建议值
Kafka Source kafka.topics 需消费的 Kafka 主题，注意参数名带 s（易踩坑点）
Kafka Source batchSize + batchDurationMillis 批量消费参数，建议根据数据量调整（如 1000 条 / 1 秒）
Kafka Channel kafka.topic 专用缓冲主题，需与 Source 消费的主题区分开
HDFS Sink hdfs.path 按时间分区（%Y-%m-%d/%H），便于后续 Spark/Flink 分析
HDFS Sink rollInterval + rollSize 文件滚动策略，避免单文件过大或过小（如 1 小时 / 128MB）

启动与验证流程

步骤 1：启动依赖组件

  1. 启动 ZooKeeper（Kafka 和 Hadoop 依赖）：

    1
    zkServer.sh start

  2. 启动 Kafka 集群：

    1
    kafka-server-start.sh /usr/local/etc/kafka/server.properties &

  3. 创建必要的 Kafka 主题：

    1
    2
    3
    4
    # 创建数据源主题（供 Source 消费）  
    kafka-topics.sh --bootstrap-server localhost:9092 --create --topic kafka-flume --partitions 3 --replication-factor 1  
    # 创建通道缓冲主题（供 Channel 使用）  
    kafka-topics.sh --bootstrap-server localhost:9092 --create --topic kafka-channel --partitions 3 --replication-factor 1
  1. 启动 Hadoop 集群：
1
2
# 启动 HDFS 和 YARN（在 Hadoop 安装目录的 sbin 下）  
./start-all.sh

步骤 2：启动 Flume Agent

执行以下命令启动 Flume，开始从 Kafka 消费数据并写入 HDFS：

1
2
3
4
5
flume-ng agent \  
  -c /usr/local/Cellar/flume/1.9.0_1/libexec/conf \  # Flume 配置目录  
  -f conf/kafka-to-hdfs.conf \  # 自定义配置文件路径  
  --name agent1 \  # Agent 名称（需与配置文件一致）  
  -Dflume.root.logger=INFO,console  # 控制台输出日志（调试用）

步骤 3：生成测试数据

使用kafka发送数据

1
kafka-console-producer --bootstrap-server localhost:9092 --topic kafka-flume

步骤 4：验证数据流转

验证 1：Kafka Channel 缓冲数据

查看通道主题 kafka-channel 的日志文件，确认数据已进入缓冲：

1
2
3
kafka-run-class kafka.tools.DumpLogSegments \  
  --files /usr/local/var/lib/kafka-logs/kafka-channel-0/00000000000000000000.log \  
  --print-data-log

输出中 payload 字段应为发送的测试数据。

验证 2：HDFS 写入结果

通过 HDFS 命令查看数据是否成功写入：

1
2
3
4
5
# 查看 HDFS 目录结构（按时间分区）  
hdfs dfs -ls /kafka-flume-hdfs/$(date +%Y-%m-%d)/$(date +%H)  

# 读取具体文件内容（替换实际文件名）  
hdfs dfs -cat /kafka-flume-hdfs/2024-07-22/15/test.1626900000000.txt

若配置正确，将输出测试数据内容。

进阶优化：提升可靠性与性能

1. 增强 Kafka Channel 可靠性

Kafka Channel 本身依赖 Kafka 的持久化机制，可通过以下参数提升可靠性：

1
2
3
4
# 通道消费者自动提交偏移量的间隔（毫秒）  
agent1.channels.kafka-channel.kafka.consumer.auto.commit.interval.ms = 5000  
# 通道生产者重试次数（失败后重试）  
agent1.channels.kafka-channel.kafka.producer.retries = 3

2. 优化 HDFS 写入效率

  • 批量写入：增大 hdfs.batchSize（如 5000），减少 HDFS 小文件数量；

  • 压缩存储：开启 HDFS 文件压缩（如 Snappy），节省存储空间：

    1
    agent1.sinks.hdfsSink.hdfs.codeC = snappy

  • 权限控制：设置 HDFS 文件权限，避免后续分析时权限不足：

1
2
agent1.sinks.hdfsSink.hdfs.filePermission = 0644  
agent1.sinks.hdfsSink.hdfs.dirPermission = 0755

3. 避免数据重复

  • Kafka 消费者偏移量：确保 kafka.consumer.group.id 唯一，避免重复消费；
  • HDFS 幂等性：通过 Flume 拦截器为每条数据添加唯一 ID，后续处理时去重。

常见问题排查

1. Flume 启动失败：Kafka 主题不存在

错误提示：Topic kafka-flume not found
解决：重新创建主题并确保名称正确：

1
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic kafka-flume --partitions 3 --replication-factor 1

2. 数据写入 HDFS 延迟或丢失

可能原因：

  • Kafka 消费受阻：检查 kafka.bootstrap.servers 是否正确，Kafka 集群是否正常；

  • HDFS 权限不足：确保 Flume 进程对 HDFS 路径有写入权限：

    1
    hdfs dfs -chmod 777 /kafka-flume-hdfs

  • 通道缓冲满：增大 Kafka Channel 主题的分区数或调整 batchSize。

3. HDFS 文件时区偏移

错误现象：HDFS 目录时间与本地时间不符（如差 8 小时）
解决：确保 hdfs.useLocalTimeStamp = true，使用本地时间而非 UTC 时间。

相关文章

欢迎关注我的其它发布渠道

表情 | 预览
快来做第一个评论的人吧~
Powered By Valine
v1.3.10