flume采集kafka消息写入HDFS

发表于 更新于 分类于 Waline： 阅读次数： 本文字数： 6.5k 阅读时长 ≈ 6 分钟

Kafka 里的数据怎么落到 HDFS？Flume 这套配置我跑通了

Kafka 和 HDFS 是大数据里最常见的两个存储，一个管实时、一个管离线。中间怎么打通？Flume 就是干这个的。

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

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

注意这里有个特殊的地方：Channel 用的是 KafkaChannel，不是 Memory 或 File。也就是说，Flume 中间缓冲的数据也放在 Kafka 里，利用 Kafka 自身的持久化保证可靠性。这个方案的优点是可靠，缺点是得多维护一个 Kafka topic。

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

详细配置步骤

# 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 存储路径（按“年-月-日/小时”分区，便于后续查询） 
# %Y-%m-%d/%H 表示按天和小时建目录，比如 2026-07-01/14/。这样存的好处是，后面用 Hive 或 Spark 查的时候，按分区过滤效率高很多
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

为什么 Channel 也用 Kafka？ 因为可靠。Memory Channel 重启丢数据，File Channel 性能一般。KafkaChannel 把数据存在 Kafka 里，Flume 挂了重启之后还能接着消费，数据不丢。代价就是多维护一个 topic，我觉得值。

注意这个 topic 跟 Source 消费的 topic 不是同一个：Source 消费 kafka-flume，Channel 暂存到 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 依赖）：

    zkServer.sh start

  2. 启动 Kafka 集群：

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

  3. 创建必要的 Kafka 主题：

    Flume 不会帮你创建 Kafka topic，所以得提前建好两个：

    # 创建数据源主题（供 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 集群：
# 启动 HDFS 和 YARN（在 Hadoop 安装目录的 sbin 下）  
./start-all.sh

步骤 2：启动 Flume Agent

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

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发送数据

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

步骤 4：验证数据流转

验证 1：Kafka Channel 缓冲数据

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

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 命令查看数据是否成功写入：

# 查看 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 的持久化机制，可通过以下参数提升可靠性：

# 通道消费者自动提交偏移量的间隔（毫秒）  
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 的 RPC 请求次数，吞吐能上来不少。但别调太大，不然单次事务处理时间太长，出问题重试代价大。

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

    agent1.sinks.hdfsSink.hdfs.codeC = snappy

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

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

不设的话，默认用的是 HDFS 的 umask，可能写进去的文件下游读不了。

3. 避免数据重复

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

常见问题排查

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

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

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 路径有写入权限：

    hdfs dfs -chmod 777 /kafka-flume-hdfs

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

3. HDFS 文件时区偏移

错误现象：HDFS 目录时间与本地时间不符（如差 8 小时）
解决：确保 hdfs.useLocalTimeStamp = true，使用本地时间而非 UTC 时间。默认用的是 UTC 时间，如果你的服务器在东八区，生成的文件目录会差 8 个小时。