Hadoop数据库怎样进行数据清洗与预处理

在大数据场景下，Hadoop 本身不是一个传统意义上的“数据库”，而是一个分布式存储与计算平台。数据清洗与预处理通常是在 Hadoop 生态组件（Hive、Spark、MapReduce、HBase、HDFS） 中完成的。下面从流程、常用工具、典型做法几个层面系统说明。

一、Hadoop 生态中数据清洗/预处理的核心思路

清洗：去除或修正异常、脏数据
预处理：转换、标准化、增强数据，便于后续分析或建模

典型处理流程：

原始数据
   ↓
HDFS 存储
   ↓
格式解析（结构化 / 半结构化）
   ↓
数据清洗（去重、补全、纠错）
   ↓
数据转换（标准化、特征工程）
   ↓
输出到 Hive / HBase / 数据湖

二、常用 Hadoop 生态工具对比

三、典型数据清洗任务及实现方式

1️⃣ 数据去重

Hive 示例

INSERT OVERWRITE TABLE clean_table
SELECT *
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY id ORDER BY ts DESC) as rn
  FROM raw_table
) t
WHERE rn = 1;

Spark（Scala / PySpark）

df = spark.read.parquet("hdfs:///data/raw")
df_clean = df.dropDuplicates(["id"])

2️⃣ 缺失值处理

Hive

SELECT
  COALESCE(age, 0) AS age,
  CASE WHEN name IS NULL THEN 'unknown' ELSE name END AS name
FROM raw_table;

Spark

from pyspark.sql.functions import when, col

df = df.fillna({"age": 0, "name": "unknown"})

3️⃣ 异常值处理

例如：年龄 < 0 或 > 120

WHERE age BETWEEN 0 AND 120

或

df = df.filter((col("age") >= 0) & (col("age") <= 120))

4️⃣ 数据格式转换

时间格式统一

SELECT unix_timestamp(time_str, 'yyyy-MM-dd') AS ts
FROM raw_table;

JSON / 日志解析

SELECT get_json_object(log, '$.user_id') AS user_id
FROM raw_log;

5️⃣ 字段标准化 / 枚举映射

CASE
  WHEN gender = '1' THEN 'M'
  WHEN gender = '0' THEN 'F'
  ELSE 'Unknown'
END

6️⃣ 大规模文本清洗（日志、爬虫数据）

Spark 示例：

from pyspark.sql.functions import regexp_replace

df = df.withColumn(
    "content",
    regexp_replace("content", "\\s+", " ")
)

四、MapReduce 清洗示例（理解原理）

map(LongWritable key, Text value) {
    String line = value.toString();
    if (line.contains("error")) return;
    context.write(new Text(line), NullWritable.get());
}

✅ 适合：

• 超大规模
• 逻辑复杂
• 无 SQL 表达能力的场景

五、数据清洗的最佳实践

✅ 层级化清洗

• ODS（原始层）
• DWD（清洗层）
• DWS（汇总层）

✅ 可重跑

• 清洗逻辑幂等
• 使用分区表（如 dt）

✅ 数据质量监控

• 空值率
• 重复率
• 字段分布变化

✅ 小文件问题

• 使用 coalesce / repartition
• HDFS 合并小文件

六、典型 Hadoop 数据预处理架构

Kafka / Flume
     ↓
Spark Streaming / Flink
     ↓
HDFS / Hive（ETL后）
     ↓
Hive / Spark SQL
     ↓
数据仓库 / 模型训练

七、总结一句话

Hadoop 本身不做数据清洗，真正做清洗的是 Hive、Spark、MapReduce 等计算框架。

如果你愿意，我可以：

• ✅ 给你一个完整的 Hadoop 数据清洗项目示例
• ✅ 根据你用的 Hive / Spark / HBase 给出更具体方案
• ✅ 结合 数据仓库（ODS → DWD → DWS） 给你设计

你可以直接告诉我：你现在的 Hadoop 版本、数据格式和清洗目标。