flume采集实时数据到hdfs

### 使用 Apache Flume 将实时数据采集到 HDFS 的配置教程 #### 1. 配置概述 Apache Flume 是一种分布式的、可靠的、高可用的服务，专门设计用于大规模的日志收集和传输。通过 Flume 可以将来自不同源的数据流式地写入目标存储系统，比如 HDFS。 以下是基于引用内容构建的一个完整的 Flume 数据采集到 HDFS 的配置示例[^1]。 --- #### 2. 创建必要的目录结构 为了确保 Flume 能够正常工作并记录状态信息，需预先创建所需的文件夹： - 在 `/usr/local/soft/apache-flume-1.9.0-bin/` 下创建 `log` 文件夹。 - 在该路径下进一步创建 `taildir_position.json` 文件，此文件用来保存读取文件的偏移量位置，防止重启后重复读取已处理过的数据[^3]。 ```bash mkdir -p /usr/local/soft/apache-flume-1.9.0-bin/log/ touch /usr/local/soft/apache-flume-1.9.0-bin/log/taildir_position.json ``` --- #### 3. 编辑 Flume 配置文件 假设我们希望监控某个日志文件的变化并将新增内容发送至 HDFS，则可以按照如下方式编写配置文件 `flume-conf.properties`: ```properties # 定义 agent 名称为 a1 a1.sources = r1 a1.sinks = k1 a1.channels = c1 # 配置 source (TailDir Source) a1.sources.r1.type = TAILDIR a1.sources.r1.positionFile = /usr/local/soft/apache-flume-1.9.0-bin/log/taildir_position.json a1.sources.r1.filegroups = f1 a1.sources.r1.filegroups.f1 = /usr/local/data/log/demo.log.* # 配置 channel (Memory Channel) a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # 配置 sink (HDFS Sink) a1.sinks.k1.type = hdfs a1.sinks.k1.hdfs.path = hdfs://namenode:8020/user/flume/logs/%Y%m%d a1.sinks.k1.hdfs.rollInterval = 60 a1.sinks.k1.hdfs.rollSize = 0 a1.sinks.k1.hdfs.rollCount = 0 a1.sinks.k1.hdfs.writeFormat = Text a1.sinks.k1.hdfs.fileType = DataStream # 绑定 source 和 sink 到同一个 channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 ``` 上述配置说明： - **Source**: 使用 TailDir Source 来监听指定路径下的多个日志文件变化，并自动跟踪新产生的文件[^5]。 - **Channel**: Memory Channel 存储临时事件缓冲区，支持快速吞吐但不持久化数据。 - **Sink**: HDFS Sink 把接收到的消息存放到远程 HDFS 中，按日期分区存放以便后续分析。 需要注意的是如果输入数据包含 JSON 格式字符串则应验证其合法性以免引发异常[^4]。 --- #### 4. 启动 Flume Agent 完成以上设置之后即可启动 Flume 并开始执行任务： ```bash bin/flume-ng agent --conf conf/ --name a1 --conf-file flume-conf.properties -Dflume.root.logger=INFO,console ``` 此时任何向 `/usr/local/data/log/demo.log.*` 添加的新行都会被立即捕获并通过管道传送到设定好的 HDFS 地址上。 --- #### 5. 常见问题排查 假如在实际操作过程中遇到诸如 `com.alibaba.fastjson.JSONException: not close json text...` 类型报错时，请核查原始日志是否存在非法字符或未闭合的大括号等问题。必要情况下可增加预处理器来清洗脏数据后再提交给下游组件处理。 ---