爬虫对接flume和hdfs

### 如何使用Flume将爬虫数据写入HDFS Flume 是一个分布式、可靠且高可用的日志采集与聚合系统，能够实时地将数据从各种来源传输到 HDFS 等目标存储中[^1]。以下内容详细说明了如何配置 Flume 将爬虫数据写入 HDFS。 #### 配置 Flume 以支持爬虫数据写入 HDFS 1. **安装与环境准备** 在开始之前，确保已经正确安装并配置了 Flume、Hadoop 和 HDFS。同时，需要将编译好的自定义 jar 包放置到 FLUME_HOME/lib 目录下[^3]。 2. **创建 Flume 配置文件** 配置文件是 Flume 的核心部分，用于定义数据源（source）、通道（channel）和接收器（sink）。以下是一个示例配置文件 `flume.conf`： ```properties # 定义 agent 名称 a1.sources = r1 a1.sinks = k1 a1.channels = c1 # 配置 source (r1) a1.sources.r1.type = exec a1.sources.r1.command = python /path/to/spider.py # 配置 channel (c1) a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # 配置 sink (k1) a1.sinks.k1.type = hdfs a1.sinks.k1.hdfs.path = hdfs://hdp1:9000/user/flume/ a1.sinks.k1.hdfs.fileType = DataStream a1.sinks.k1.hdfs.writeFormat = Text a1.sinks.k1.hdfs.rollInterval = 30 a1.sinks.k1.hdfs.rollSize = 0 a1.sinks.k1.hdfs.rollCount = 0 # 绑定 source 和 sink 到 channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 ``` - `a1.sources.r1.type = exec` 表示通过执行命令的方式获取数据，这里可以运行爬虫脚本。 - `a1.sinks.k1.type = hdfs` 表示数据将被写入 HDFS。 - `a1.sinks.k1.hdfs.path` 指定了 HDFS 中的数据存储路径。 3. **启动 Flume Agent** 使用以下命令启动 Flume Agent，确保后台运行并将日志输出重定向： ```bash nohup bin/flume-ng agent -n a1 -c conf -f conf/flume.conf >/dev/null 2>&1 & ``` 这里的 `-n a1` 表示使用名为 `a1` 的 agent，`-f conf/flume.conf` 指定了配置文件路径[^2]。 4. **验证数据传输** 在 HDFS 中检查数据是否成功写入： ```bash hdfs dfs -ls /user/flume/ ``` 如果配置正确，应该可以看到由 Flume 创建的文件。 5. **优化与扩展** - 如果需要更高的可靠性，可以将内存通道替换为文件通道（file channel）。 - 可以通过调整 `rollInterval`、`rollSize` 和 `rollCount` 参数来控制文件滚动策略。 #### 示例代码：将本地文件上传至 HDFS 如果需要手动将爬虫生成的文件上传至 HDFS，可以参考以下 Java 代码片段[^4]： ```java package com.zpark.kafka; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.io.IOException; import java.net.URI; import java.net.URISyntaxException; public class AccessTask { public static void main(String[] args) { URI uri = null; try { uri = new URI("hdfs://hdp1:9000"); Configuration conf = new Configuration(); conf.set("dfs.replication", "2"); conf.set("dfs.blocksize", "64m"); String user = "root"; FileSystem fs = FileSystem.get(uri, conf, user); Path src = new Path("/path/to/local/file.txt"); Path dst = new Path("/user/flume/uploaded_file.txt"); fs.copyFromLocalFile(src, dst); fs.close(); } catch (URISyntaxException | IOException | InterruptedException e) { e.printStackTrace(); } } } ``` 此代码实现了将本地文件上传至指定 HDFS 路径的功能。 ---