基于Hadoop与AI Agent的农产品数据清洗全流程实战

很多同学在毕设选题或者做项目实战时，一听“大数据+AI”就觉得必须上深度学习、上复杂模型，但真正动手后才发现，卡住进度的往往是数据本身。尤其是农产品数据这种典型的“脏数据”来源：你拿到一份交易记录，里面有乱码、缺失、单位不统一、重复记录、日期乱写、名称一会儿“土豆”一会儿“马铃薯”……如果不在进入数据分析流程之前做一次彻底的数据清洗，后续无论跑 Hadoop 离线统计、Hive 查询还是接入 AI Agent 问答，结果都会失真，甚至导致整个系统无法上线。

这篇文章围绕“基于 Hadoop 与 AI Agent 的农产品数据分析及助农甄选系统”完整项目中的数据清洗环节展开，讲清楚三件事：

1. 农产品数据清洗到底要处理哪些问题；
2. 怎么用 Hadoop MapReduce + Hive + Pandas 做一套可落地的清洗方案；
3. 清洗结果如何与 AI Agent 联动，让后面的“助农甄选”可信可用。

整套方案不依赖高端配置，单机伪分布式 Hadoop 环境就能跑通。无论是课程设计、毕业设计还是新入门大数据开发，都可以直接参考这套思路来复现。

1. 背景：数据清洗在这个项目里到底解决什么问题

1.1 一个完整的大数据 + AI 项目长什么样

先把这个项目的整体链路摆出来，方便后面理解数据清洗的位置。

项目最终目标是做一个“农产品数据分析及助农甄选系统”。它不只是做一个可视化大屏，而是需要回答几类实际问题：

• 某个脱贫县适合种什么农产品？
• 不同产区的农产品价格走势如何？
• 哪个渠道的销量最好、价格波动最大？
• 农户上传的数据是否可信、哪些产地需要帮扶推荐？

要实现这些能力，数据链路大致是：

code复制数据采集（爬虫/接口/手工导入）
    ↓
数据存储（HDFS / 本地文件）
    ↓
数据清洗（MapReduce / Hive / Pandas）
    ↓
数据仓库（Hive 分层表）
    ↓
离线统计与分析（Hive SQL / Spark / Python）
    ↓
指标服务（API / 结果表）
    ↓
AI Agent 问答与助农甄选推荐
    ↓
前端可视化与交互

可以看到，数据清洗是整个项目的“入口关卡”。这一关不过，后面所有统计指标、价格预测、助农推荐都会建立在不准确的数据上。

1.2 为什么用 Hadoop 做数据清洗，AI Agent 又负责什么

农产品交易数据通常有几个特点：

• 数据量大，可能一天有几百万条交易明细；
• 来源杂，线下农贸市场、电商平台、合作社台账、农户上报都有；
• 字段不统一，不同来源的表格结构、单位、商品名称都不一样；
• 质量差，缺失、重复、异常值多。

这种规模的数据，单靠 Excel 或者 Pandas 在本地处理，内存很快就会成为瓶颈。用 Hadoop MapReduce 做批式清洗，可以发挥分布式计算的能力，把清洗任务拆分到多台机器上并行执行。即使是伪分布式环境，也能用完整的 MapReduce 流程来学习清洗任务的编程模型。

AI Agent 在这个项目中不是做数据清洗的主力，它更多是清洗之后的分析层应用。你可以把 AI Agent 理解为一个连接“数据仓库”与“用户问题”的智能中间层。用户问“今年陕西苹果价格趋势怎么样”，AI Agent 先从清洗后的数据仓库中检索相关指标，再做归纳回答。如果清洗层做的不好，AI Agent 给出看似流畅但实际错误的回答，这在助农场景中会造成很严重的误导。所以先做数据清洗，再做 Agent 应用，顺序不能反。

1.3 数据清洗的常见误区

很多初学者容易犯几个认知错误：

误区一是把“去重”当作数据清洗的全部。实际上清洗还包括格式转换、缺失值处理、异常值识别、单位统一、名称映射等多个环节。

误区二是清洗时只盯着当前表看，不考虑后续分析需要哪些字段。比如后面要做“价格按省份、品类、日期的聚合”，清洗阶段就必须保证产地、品类、日期这些字段规范可解析。

误区三是清洗结果直接覆盖原始数据。这是生产环境的大忌。原始数据要保留，清洗数据要落在新的目录或表里，方便数据回溯和对账。

2. 环境准备与版本说明

2.1 基础环境说明

本节以一个典型的 Hadoop 伪分布式环境为例。版本不一定需要最新，重点是稳定和能跑通 MapReduce。

建议环境如下：

如果你的电脑内存只有 8G，运行完整集群会比较吃力，建议先按伪分布式模式配置。伪分布式模式下，所有守护进程都跑在同一台机器上，对于学习和验证 MapReduce 清洗流程完全够用。

关于 Hadoop 具体安装步骤，网上有大量“Hadoop 伪分布式搭建”教程，这里不再重复，只强调安装完成后必须确认三件事：

1. jps 能看到 NameNode、DataNode 等进程；
2. 能正常访问 http://localhost:9870（Hadoop 3.x）或 http://localhost:50070（Hadoop 2.x）；
3. 可以在 HDFS 上创建目录并上传文件。

2.2 项目目录结构

为了后面的实战代码不混乱，先约定一个标准目录结构：

code复制agri-select-system/
├── data/
│   ├── raw/                 # 原始脏数据
│   └── cleaned/             # 清洗后的数据输出
├── etl/
│   ├── mapreduce/           # MapReduce 清洗工程
│   └── scripts/             # Pandas 辅助脚本
├── hive/                    # Hive 建表与清洗 SQL
│   ├── create_tables.sql
│   └── clean_agri.sql
└── agent/                   # AI Agent 调用层（后续文章展开）

这个结构的核心思想是原始数据、清洗代码、清洗结果、分析代码分开管理，避免把所有内容堆在一个目录里。

2.3 数据样例设计

本文使用的农产品数据是模拟的业务样例，字段设计参考了常见的农产品交易台账。CSV/TSV 文件格式，字段之间用 \t 分隔，样例数据如下：

code复制record_id	product_name	category	origin_province	origin_city	price	unit	quantity	trade_date	channel	farmer_id	region_tag
20250101_A001	土豆	蔬菜	陕西	榆林	1.5	元/斤	5000	2025-01-01	电商平台	F001	重点帮扶县
20250101_A001	土豆	蔬菜	陕西	榆林	1.5	元/斤	5000	2025-01-01	电商平台	F001	重点帮扶县
20250101_B002	马铃薯	蔬菜	陕西	榆林	3.0	元/公斤	2600	2025-01-01	批发市场	F002	普通区县
20250102_C003	西红柿	蔬菜	山东	潍坊	2.8	元/斤	-100	2025-01-02	农贸市场	F003	重点帮扶县
20250102_C003	西红柿	蔬菜	山东	潍坊	99999	元/斤	2000	2025-01-13	农贸市场	F003	重点帮扶县
20250102_D004	 苹果 	 水果	�西	延安	6.5	元/斤	800	2025-01-02	电商平台	F004	普通区县
...

上一眼看上去问题就不少：重复、单位不统一、名称不统一、负价格、异常价格、乱码、首尾空格、日期部分合理部分可疑。这些脏数据，正好对应数据清洗要处理的典型问题。

3. 农产品数据的质量问题分析

3.1 数据从哪来

农产品数据分析项目的数据来源通常包含四类：

第一类是电商平台接口返回的商品快照，字段相对规范，但价格、销量可能存在刷单和异常值。

第二类是线下批发市场手工记录的台账，经常有漏填、错填、单位混用的问题。

第三类是农户或合作社上报的数据，涉及大量中文输入，错别字、简称、方言叫法很常见。

第四类是第三方行情网站爬取的数据，虽然相对规整，但乱码、空值、重复抓取不可避免。

所以在设计清洗方案之前，先要知道“数据从哪里来”，这样才能预判会碰到哪些问题。来源不同，清洗规则就要针对性设计。

3.2 常见脏数据形态

农产品数据中最常见的脏数据可以分成几类：

第一，重复数据。同一条交易记录可能因为系统重试、手工重复导入等原因出现多次。重复数据会导致统计结果虚高，比如总销量被重复累计。

第二，缺失数据。比如产地缺失、价格缺失、农户编号缺失。缺失字段会直接影响后续的维度分析和关联查询。

第三，异常数据。典型的是价格出现负值、单日价格超过正常范围、日期格式错误（比如 2025-13-45）、数量为负数等。

第四，单位不统一。有的记录用“元/斤”，有的用“元/公斤”，甚至还有“元/箱”。如果直接对价格做统计，结果毫无意义。

第五，名称不一致。同一个商品有“土豆”和“马铃薯”、“西红柿”和“番茄”、“车厘子”和“樱桃”等写法，如果不做标准化，分组统计会分成多个组。

第六，编码和乱码问题。比如省份字段出现“�西”，明显是字符集转换造成的问题。

第七，隐藏字符问题。比如字符串首尾有空格、TAB、BOM 头部，这些字符很隐蔽，会导致 equal 判断失效。

3.3 我们的清洗目标

针对上面这些问题，我们要在清洗阶段达到以下目标：

一是唯一性。清洗后每条记录应有唯一可识别的业务主键，重复记录被剔除或合并。

二是完备性。核心字段不允许缺失，允许缺失的次要字段要有标记或默认值。

三是准确性。价格、数量、日期必须在合理范围内，异常值要么剔除，要么修正。

四是一致性。单位、名称、地区编码必须统一到同一套标准上。

五是可追踪性。清洗过程中被剔除、被修正的记录数量要能通过计数器或日志看到，方便数据质量报告。

4. 清洗规则拆解

4.1 去重规则

去重不是简单地把重复行删除，而是要先识别“什么才算重复”。

在本项目里，定义业务唯一键为 record_id。同一渠道、同一农户、同一商品、同一天内产生的交易记录，应该只有一条。如果出现多条，就需要判断保留哪一条。

保留策略通常有三种：

一是保留第一条，适合重复导入导致的情况；
二是保留最后一条，适合数据被反复修改的情况；
三是保留质量评分最高的一条，需要给每个字段计算质量分。

本文 MapReduce 示例采用“保留第一条并统计重复数量”的策略。更严格的场景下，建议对每条记录计算一个质量分，比如字段越完整、数值越正常，质量分越高，去重时保留最高分记录。

4.2 缺失值处理

缺失值处理要区分字段类型。

主键字段 record_id 缺失时，不能简单丢弃，可以用 MD5(整行数据) 生成一个新主键，保证记录不丢。

价格字段缺失时，需要结合业务决定。如果是统计均价，缺少价格会导致均值和总额失真。比较稳妥的做法是：先用同品类、同省份、最近一周的中位数填充，并打上“填充”标记。如果没有任何可参考数据，则直接剔除。

产地字段缺失时，可以考虑保留记录并打上 unknown 标记，后续分析可以按“未知产地”单独聚合，而不是直接删掉。

数量字段缺失时，可以用同渠道同商品的平均数量填充，或者同样打标记。

需要强调的是，缺失值处理一定不能对所有字段一刀切。像 farmer_id、region_tag 这类字段缺失时，影响的是辅助分析，不适合直接删除整条记录。

4.3 异常值处理

异常值识别的核心是设置合理的业务阈值。

价格字段的判定：先看是否为负数，再看是否超过同品类平均价格的三倍标准差。对于农产品，不同品类价格差异极大，苹果一斤可能几块钱，松茸可能上百块，所以不能设置全局统一的上下限，必须按品类分组判断。

日期字段的判定：匹配 yyyy-MM-dd 格式，月份 01-12，日期 01-31，同时日期不能晚于当天。对于那些日期显然乱写的记录，建议直接剔除。

数量字段的判定：必须大于 0。出现负数或零，很可能是系统错误或者输入错误。

异常值处理完成后，建议输出一份异常值统计报告，比如“价格负数剔除 200 条”“重复记录 500 条”等。MapReduce 的 Counter 天然支持这种统计，后面会展示。

4.4 单位统一与名称标准化

单位统一是一个容易忽略但影响极大的清洗步骤。

本项目约定核心价格单位统一为 元/公斤。如果原始单位是 元/斤，按 1 斤 = 0.5 公斤换算，即价格乘以 2。如果原始单位是 元/箱，则需要知道箱子规格后才能换算，无法直接转换的记录需要打上标记保留原单位，避免强行转换导致错误。

名称标准化需要维护一张映射表：

code复制土豆 -> 土豆
马铃薯 -> 土豆
西红柿 -> 西红柿
番茄 -> 西红柿
车厘子 -> 樱桃
...

在 MapReduce 里可以通过加载 DistributedCache 的方式把映射表分发到各个 Map Task，也可以先用 Python 脚本做一次预清洗，再进入 MapReduce。对于毕设项目，把映射表直接放在 Java 代码的静态 Map 里最方便理解。

4.5 派生字段与质检标记

清洗不只是“删”和“改”，还要生成一些对分析有意义的派生字段。

本项目添加三个派生字段：

一是 month，从 trade_date 提取月份，方便后续做月度价格分析；
二是 season，根据月份映射春夏秋冬，方便季节性分析；
三是 is_trusted，综合记录质量打出的可信标记。一般情况下，经过完整校验且没有缺失值填充的记录标记为 1，有填充或修正的记录标记为 0。

is_trusted 字段非常有用。后续 AI Agent 做助农推荐时，可以优先基于 is_trusted=1 的高质量数据回答问题，避免把填充和修正的数据当作真实数据使用。

5. 核心实战：MapReduce 清洗任务

5.1 设计思路

接下来进入代码实战环节。我们用 Java 编写一个 MapReduce 清洗任务，完成以下功能：

• 读取原始 TSV 文件；
• 剔除字段数不足的记录；
• 去除字符串首尾空格和 BOM；
• 校验日期与价格；
• 生成缺失的主键；
• 以 record_id 作为输出的 key，同一 key 的重复记录在 Reducer 阶段只保留第一条；
• 使用 Counter 统计重复、异常、正常记录数量。

先约定输入输出的 KeyValue 类型。Mapper 输出 key 为清洗后的 record_id，value 为清洗后的整行数据。Reducer 负责去重，输出相同格式。

5.2 Mapper 代码

在 Maven 工程中新建 Java 类 /src/main/java/com/agri/clean/CleanMapper.java。

核心代码如下：

java复制package com.agri.clean;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.security.MessageDigest;

public class CleanMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        String line = value.toString();
        // 去除 UTF-8 BOM 和首尾空白
        line = line.replace("\uFEFF", "").trim();
        if (line.isEmpty()) {
            return;
        }

        // 按 Tab 分隔
        String[] fields = line.split("\t", -1);

        // 字段数量校验：本项目固定 12 个字段
        if (fields.length < 12) {
            context.getCounter(CleanCounter.BAD_RECORD).increment(1);
            return;
        }

        // 去掉每个字段的首尾空格
        for (int i = 0; i < fields.length; i++) {
            if (fields[i] != null) {
                fields[i] = fields[i].trim();
            }
        }

        // 日期校验
        String date = fields[9];
        if (!isValidDate(date)) {
            context.getCounter(CleanCounter.BAD_DATE).increment(1);
            return;
        }

        // 价格校验
        double price = parsePrice(fields[6]);
        if (price <= 0 || price > 100000) {
            context.getCounter(CleanCounter.BAD_PRICE).increment(1);
            return;
        }

        // 数量校验
        double quantity = parsePrice(fields[8]);
        if (quantity <= 0) {
            context.getCounter(CleanCounter.BAD_QUANTITY).increment(1);
            return;
        }

        // 主键缺失时生成 MD5 主键
        String recordId = fields[0];
        if (recordId == null || recordId.isEmpty()) {
            recordId = md5(line);
        }

        // 输出 key=record_id, value=清洗后的整行
        String cleanLine = String.join("\t", fields);
        context.write(new Text(recordId), new Text(cleanLine));
        context.getCounter(CleanCounter.OK_RECORD).increment(1);
    }

    private boolean isValidDate(String date) {
        return date != null && date.matches("\\d{4}-\\d{2}-\\d{2}");
    }

    private double parsePrice(String s) {
        try {
            return Double.parseDouble(s);
        } catch (NumberFormatException e) {
            return -1;
        }
    }

    private String md5(String input) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] digest = md.digest(input.getBytes("UTF-8"));
            StringBuilder sb = new StringBuilder();
            for (byte b : digest) {
                sb.append(String.format("%02x", b));
            }
            return sb.toString();
        } catch (Exception e) {
            return String.valueOf(System.currentTimeMillis());
        }
    }
}

这里有几个细节需要注意：

line.split("\t", -1) 中 -1 的作用是保留末尾空字段，避免因为尾部 Tab 导致字段丢失。

CleanCounter 是一个枚举类型，用来记录各种统计数量：

java复制package com.agri.clean;

public enum CleanCounter {
    OK_RECORD,
    BAD_RECORD,
    BAD_DATE,
    BAD_PRICE,
    BAD_QUANTITY,
    DUP_RECORD
}

5.3 Reducer 去重代码

新建 /src/main/java/com/agri/clean/CleanReducer.java。

java复制package com.agri.clean;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class CleanReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        boolean first = true;
        // key 相同的记录会进入同一个 reduce 调用
        // 这里只保留第一条，其余计为重复记录
        for (Text value : values) {
            if (first) {
                context.write(key, value);
                first = false;
            } else {
                context.getCounter(CleanCounter.DUP_RECORD).increment(1);
            }
        }
    }
}

这个 Reducer 逻辑非常简单，适合入门理解。它的局限在于：如果有多条重复记录，每条质量不一样，它并没有选择质量最高的那条。实际工程项目里，可以在 Mapper 阶段给每条记录打分，并作为一个二次排序字段，让 Reducer 排序后自然取到最优记录。这属于进阶优化，在本文第 8 节会进一步说明。

5.4 Job 提交与运行

新建主类 /src/main/java/com/agri/clean/CleanJob.java。

java复制package com.agri.clean;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class CleanJob {

    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: CleanJob <input> <output>");
            System.exit(2);
        }

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "agri-data-clean");

        job.setJarByClass(CleanJob.class);
        job.setMapperClass(CleanMapper.class);
        job.setReducerClass(CleanReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

pom.xml 中需要引入 Hadoop 客户端依赖：

xml复制<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.3.6</version>
    </dependency>
</dependencies>

如果使用 Hadoop 2.x，请将版本号改成对应的 2.x 版本，例如 2.7.7。注意版本需要和你本机的 Hadoop 环境匹配，否则提交任务时可能会出现序列化类不一致或者协议版本问题。

编译打包：

bash复制mvn clean package -DskipTests

假设打包产物是 etl-mapreduce-1.0.jar，先启动 Hadoop 相关进程，然后把原始数据上传到 HDFS：

bash复制# 如果 HDFS 还没有相关目录，先创建
hdfs dfs -mkdir -p /agri/input

# 上传原始数据
hdfs dfs -put data/raw/agri_raw.tsv /agri/input/

# 提交 MapReduce 任务
hadoop jar etl-mapreduce-1.0.jar com.agri.clean.CleanJob \
  /agri/input/agri_raw.tsv /agri/output/clean_result_1

# 查看输出
hdfs dfs -cat /agri/output/clean_result_1/part-r-00000

5.5 结果验证

运行完成后，终端会输出 Counter 统计信息：

code复制com.agri.clean.CleanCounter
        OK_RECORD=5
        BAD_RECORD=1
        BAD_DATE=1
        BAD_PRICE=1
        BAD_QUANTITY=1
        DUP_RECORD=1

通过 Counter 可以直观看到记录被剔除的原因和数量。抽看输出结果时，重点检查三个点：

1. 是否还有乱码和空格；
2. 重复记录是否只剩一条；
3. 日期和价格是否已经是合法值。

如果输出结果符合预期，说明第一轮清洗完成。

6. Hive 清洗与 Pandas 辅助预处理

6.1 为什么还要做一层 Hive 清洗

MapReduce 清洗解决了“单条记录是否合法”的问题，但没有解决“跨记录计算”的问题。比如中位数填充、同品类均值判断异常值，这类需要参考其他记录的计算，用 Hive SQL 更方便。

另外，MapReduce 产出的清洗结果还只是普通文本文件，后续做数据统计时，用 Hive 建表关联查询更高效。所以方案上是用 MapReduce 先做“行级清洗”，再用 Hive 做“表级清洗和入库”。

6.2 建表与分区

在 Hive 中先建立原始数据外部表，指向 HDFS 中的清洗结果：

sql复制CREATE EXTERNAL TABLE IF NOT EXISTS dwd_agri_clean (
    record_id STRING,
    product_name STRING,
    category STRING,
    origin_province STRING,
    origin_city STRING,
    price DOUBLE,
    unit STRING,
    quantity INT,
    trade_date STRING,
    channel STRING,
    farmer_id STRING,
    region_tag STRING,
    month STRING,
    season STRING,
    is_trusted INT
)
PARTITIONED BY (dt STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE
LOCATION '/agri/dwd/agri_clean';

这里使用外部表而不是内部表，好处是删除表不会误删 HDFS 底层数据文件，对生产和毕设都比较安全。

6.3 清洗 SQL

下面这段 SQL 完成几件事：

1. 从 MapReduce 清洗结果中查询基础数据；
2. 过滤掉价格异常明显的记录；
3. 补全 month、season、is_trusted 派生字段；
4. 写入分区表。

sql复制INSERT OVERWRITE TABLE dwd_agri_clean PARTITION(dt='2025-01-15')
SELECT
    record_id,
    product_name,
    category,
    origin_province,
    origin_city,
    price,
    unit,
    quantity,
    trade_date,
    channel,
    farmer_id,
    region_tag,
    substr(trade_date, 1, 7) AS month,
    CASE
        WHEN substr(trade_date, 6, 2) IN ('03','04','05') THEN 'spring'
        WHEN substr(trade_date, 6, 2) IN ('06','07','08') THEN 'summer'
        WHEN substr(trade_date, 6, 2) IN ('09','10','11') THEN 'autumn'
        ELSE 'winter'
    END AS season,
    CASE
        WHEN price > 0 AND quantity > 0 AND origin_province != 'unknown' THEN 1
        ELSE 0
    END AS is_trusted
FROM ods_agri_mapreduce_result
WHERE dt = '2025-01-15'
  AND price > 0
  AND quantity > 0
  AND trade_date >= '2020-01-01';

这里有一处可以优化：season 字段的月份判断只提取了 MM 部分，没有校验 trade_date 格式。实际开发中，建议先用正则或 regexp_replace 将日期转成标准格式再提取，避免脏日期绕过校验。

如果 Hive 中还没建 ods_agri_mapreduce_result 表，可以用下面语句把 MapReduce 输出目录关联进来：

sql复制CREATE EXTERNAL TABLE IF NOT EXISTS ods_agri_mapreduce_result (
    record_id STRING,
    product_name STRING,
    category STRING,
    origin_province STRING,
    origin_city STRING,
    price DOUBLE,
    unit STRING,
    quantity INT,
    trade_date STRING,
    channel STRING,
    farmer_id STRING,
    region_tag STRING
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE
LOCATION '/agri/output/clean_result_1';

6.4 Pandas 辅助脚本

有些清洗逻辑放在 SQL 里写不方便，比如单位转换映射表、中文名称标准化，用 Python Pandas 更灵活。下面脚本可以在数据量不大时作为补充。

python复制# 文件路径：etl/scripts/preprocess_agri.py
import pandas as pd

df = pd.read_csv("data/raw/agri_raw.tsv", sep="\t", dtype=str)

# 去除 BOM 和首尾空格
for col in df.columns:
    df[col] = df[col].str.replace("\ufeff", "").str.strip()

# 商品名称标准化映射
name_map = {
    "马铃薯": "土豆",
    "番茄": "西红柿",
    "洋芋": "土豆",
    "圣女果": "小番茄"
}
df["product_name"] = df["product_name"].replace(name_map)

# 单位统一为元/公斤
def convert_unit(price_str, unit_str):
    try:
        price = float(price_str)
    except:
        return price_str
    if unit_str == "元/斤":
        return str(round(price * 2, 2))
    elif unit_str == "元/公斤":
        return str(price)
    return price_str

df["price"] = df.apply(
    lambda row: convert_unit(row["price"], row["unit"]),
    axis=1
)
df["unit"] = "元/公斤"

# 保存清洗中间结果
df.to_csv("data/cleaned/agri_cleaned_pandas.tsv", sep="\t", index=False)
print("清洗完成，记录数:", len(df))

Pandas 脚本适合数据量在百万行以下、内存能扛住的场景。如果数据量达到上亿级别，建议以 MapReduce + Hive 为主，Pandas 只做样本抽查和质量分析。

7. 清洗结果如何被 AI Agent 使用

7.1 数据分层与存储

经过 MapReduce 和 Hive 清洗后，数据有了比较清晰的层次：

• 原始层：/agri/input/ 保存未清洗的原始数据；
• 明细层：dwd_agri_clean 保存清洗后的明细数据；
• 聚合层：后续根据业务需求生成“省份-品类-月份价格统计表”“渠道销量统计表”等。

AI Agent 通常不会直接读取明细表，因为明细数据量太大，也不适合自然语言问答。更合理的做法是把聚合结果表导出为轻量级的数据文件或数据库表，再通过 API 提供给 AI Agent 调用。

7.2 AI Agent 的问答链路

AI Agent 的问答链路大致如下：

code复制用户提问 -> 意图识别 -> 检索聚合数据 -> 组装上下文 -> 生成回答

比如用户问：“陕西榆林最近三个月土豆价格是多少？”Agent 需要先判断用户想查“价格趋势”，然后从聚合表中检索“榆林、土豆、最近三个月”的数据，再把数据作为上下文交给语言模型生成回答。

如果清洗阶段没有把“土豆/马铃薯”统一，Agent 检索时只查“土豆”，就会漏掉大量以“马铃薯”为名称的记录，回答必然不完整。

7.3 数据质量标记的价值

清洗时生成的 is_trusted 字段，在助农甄选场景里作用非常大。

比如系统要回答“哪个重点帮扶县的农产品销量最好”，如果某些记录的价格和数量是用填充值补上的，就不应该作为推荐依据。AI Agent 在检索时可以通过 is_trusted=1 过滤数据，保证所有结论都来自真实可靠的原始记录。

这也体现了数据清洗与下游智能应用的深度耦合：清洗不只是让数据“好看”，而是让下游应用在行动上更可靠。

8. 常见问题与排查思路

下面整理这套清洗方案里最容易遇到的问题。

遇到问题时，建议按以下顺序排查：

1. 先看自己的原始数据长什么样，不要假设数据格式符合预期；
2. 再确认 Hadoop 服务进程都正常，HDFS 路径存在；
3. 然后查看 MapReduce 日志中的 Counter 统计，判断数据在哪一步被丢弃；
4. 最后抽查输出目录的随机样本，验证清洗质量。

9. 最佳实践与工程建议

9.1 数据安全与授权

农产品数据中可能包含农户编号、合作社信息、手机号等敏感字段。在开发阶段，建议使用脱敏后的模拟数据，不要直接使用真实生产数据。在生产环境中，对数据文件的访问要做用户权限隔离，遵循最小权限原则。

本项目是教学和课程设计场景，因此可以使用模拟数据。但如果是真实数据项目，必须在清洗脚本中明确脱敏步骤，比如对手机号、身份证号做掩码处理。

9.2 清洗任务必须可重跑

数据清洗最忌讳“跑一次就丢文件”。建议所有清洗任务的输出目录都带上批次或日期，例如：

code复制/agri/output/clean_result_20250115_001

如果清洗规则做了调整，可以重新运行任务，不影响历史结果，也方便对比前后两轮清洗的差异。

同时在 Hive 表中使用分区字段 dt，每次按日期覆盖当天分区，回滚时只需要删掉错误分区。

9.3 清洗规则配置化

不要把所有清洗规则硬编码在 Java 代码里。至少把以下内容外部化配置：

• 字段分隔符；
• 必填字段；
• 价格上下限；
• 名称标准化映射表；
• 单位换算表。

在 Hadoop 的 Configuration 中直接传递简单参数，映射表可以放到 HDFS 上并用 DistributedCache 分发。配置化之后，业务人员调整清洗规则时不需要重新编译代码。

9.4 血缘追踪与日志

开发阶段可能觉得“格式规范一点就行”，但工程化之后必须考虑数据血缘。

最简单的做法是将清洗日志写到独立目录，内容包括：任务 ID、输入路径、输出路径、Counter 统计结果、运行时间。这样数据质检时能回答“这批数据为什么少了 1000 条”“是哪一步丢掉的”。

在 Hive 表注释中添加数据来源说明，也可以帮助后续维护人员快速理解表结构。

9.5 性能优化方向

当前这套流程可能跑在伪分布式环境，一旦迁移到真实集群，有几个优化点很关键：

一是使用压缩格式存储，比如清洗结果直接写 Parquet 或 ORC，能显著减少存储空间和查询时间。

二是 MapReduce 阶段可以做小文件合并，因为大量小文件会让 HDFS 的 NameNode 压力很大。

三是 Hive 查询尽量按分区裁剪，避免全表扫描。在清洗 SQL 中，加上 dt 分区过滤条件是最基础但最有效的优化。

四是如果数据量持续增大，考虑引入 Spark 或 Flink 做实时清洗。但作为毕设项目，掌握 Hadoop MapReduce 的清洗思路，再迁移到 Spark，思路是通用的。

10. 总结：7天项目里的数据清洗进度表

如果要把这个“大数据 + AI”毕设项目压缩到 7 天完成，数据清洗这部分建议按下面节奏推进：

数据清洗不是整个项目里最“炫”的部分，但它决定了后面所有分析结论是否可信。如果你现在能亲手把一份带乱码、重复、单位混乱的农产品数据，通过 MapReduce 洗成一张干净规范的表，再去接 AI Agent 做问答，整个项目的完成度就已经超过大部分只跑通 Demo 的同学了。

动手的时候建议从原始数据开始，先手动做一次小样本清洗，明确每一条规则，再写到代码里。规则越清楚，代码越简单。后续文章会继续拆解 Hive 指标层和 AI Agent 联调部分，感兴趣的话可以接着往下实践。