已经装好 Hadoop + Spark，接下来核心是跑通数据链路（把文本数据存入 HDFS → Spark 清洗 → 导出供模型训练），这是连接 Hadoop 和大模型的关键一步，我给你极简、可直接复制的操作步骤，10 分钟就能搞定。

第一步：准备测试数据（本地→HDFS）

先在 ECS 上创建一个简易的「客服文本分类」数据集，这是后续模型训练的数据源，内容简单、易验证。

1. 本地创建数据文件

# 新建文件夹存放数据
mkdir -p ~/paddle_demo && cd ~/paddle_demo

# 创建测试数据文件（格式：文本,分类标签）
echo -e "我的订单怎么还没发货,物流咨询\n申请退款多久到账,退款咨询\n产品保质期多久,产品咨询\n快递丢了怎么办,物流咨询\n退款审核要几天,退款咨询\n产品怎么使用,产品咨询\n空行过滤测试,\n,无效标签\n产品咨询,产品咨询" > customer_service.csv

2. 上传到 HDFS

启动Hadoop服务，执行

# 2. 启动 HDFS 集群
start-dfs.sh
# 3. 启动 YARN 集群
start-yarn.sh

把本地文件上传到 Hadoop 的 HDFS 中（伪分布式也能正常用）：

# 1. 在 HDFS 创建目录（如果已存在会提示，忽略即可）
hadoop fs -mkdir -p /user/demo/data/raw

# 2. 上传本地文件到 HDFS
hadoop fs -put customer_service.csv /user/demo/data/raw/

# 3. 验证上传成功（查看 HDFS 文件）
hadoop fs -ls /user/demo/data/raw/
hadoop fs -cat /user/demo/data/raw/customer_service.csv

✅ 预期输出：能看到你创建的文本行，说明数据已存入 HDFS。

第二步：用 Spark 清洗数据（核心：过滤无效数据）

写一个极简的 PySpark 脚本，完成「去空行、去重、过滤无效标签」，输出干净的数据集供模型训练，这是 Hadoop 给大模型提供高质量数据的核心环节。

1. 编写 Spark 清洗脚本

# 创建清洗脚本
vim ~/paddle_demo/clean_data.py

把下面代码粘贴进去（按 i 编辑，编辑完按 ESC + :wq 保存）：

from pyspark.sql import SparkSession

# 初始化 Spark 会话（对接 Hadoop 伪分布式）
spark = SparkSession.builder \
    .appName("DemoDataClean") \
    .master("yarn") \
    .getOrCreate()

# 1. 读取 HDFS 中的原始数据
# 注意替换 <你的HDFS地址>：伪分布式通常是 hdfs://localhost:9000 或 hdfs://<ECS内网IP>:9000
df = spark.read.csv(
    "hdfs://localhost:9000/user/demo/data/raw/customer_service.csv",
    header=False,  # 无表头
    encoding="utf-8"
)

# 2. 数据清洗：过滤空行、无效标签、去重
# _c0=文本列，_c1=标签列
clean_df = df \
    .filter(df._c0.isNotNull())  # 过滤文本为空
    .filter(df._c1.isNotNull())  # 过滤标签为空
    .filter(df._c1.isin(["物流咨询", "退款咨询", "产品咨询"]))  # 只保留有效标签
    .dropDuplicates()  # 去重

# 3. 保存清洗后的数据到 HDFS
clean_df.write.csv(
    "hdfs://localhost:9000/user/demo/data/clean",
    header=False,
    mode="overwrite",  # 覆盖已有文件
    encoding="utf-8"
)

# 4. 验证清洗结果（打印前10行）
print("清洗后的数据：")
clean_df.show(10)

spark.stop()

2. 执行 Spark 清洗任务

# 提交 Spark 任务（伪分布式用 yarn 模式）
spark-submit --master yarn ~/paddle_demo/clean_data.py

✅ 预期输出：打印出清洗后的 6 行有效数据（过滤了空行、无效标签），且 HDFS 生成 clean 目录。

如果内存不够，请使用本地模式执行：

spark-submit --master local[1] --driver-memory 512m ~/paddle_demo/clean_data.py

3. 验证清洗结果

# 查看 HDFS 清洗后的数据文件
hadoop fs -ls /user/demo/data/clean/

# 查看清洗后的数据内容
hadoop fs -cat /user/demo/data/clean/part-*

✅ 预期输出：只有 6 行有效数据（3 类标签各 2 条），无空行、无重复。

第三步：导出清洗后的数据（供模型训练）

把 HDFS 中的干净数据下载到本地，或直接让后续的 Python 代码读取 HDFS 数据，这里先下载到本地做备份，方便后续调试：

# 创建本地目录存放清洗后的数据
mkdir -p ~/paddle_demo/clean_data

# 从 HDFS 下载清洗后的数据到本地
hadoop fs -get /user/demo/data/clean/* ~/paddle_demo/clean_data/

# 查看本地数据
cat ~/paddle_demo/clean_data/part-*

第四步：安装模型训练依赖（连接数据和大模型）

现在数据链路已通，下一步要安装飞桨（PaddlePaddle）和相关依赖，为训练文本分类模型做准备：

# 安装 pip（如果没装）
apt update && apt install python3-pip -y

# 安装飞桨（CPU 版本，适配 ECS 无 GPU 场景）
pip3 install paddlepaddle==2.5.2 -i https://pypi.tuna.tsinghua.edu.cn/simple

# 安装飞桨 NLP 库（含轻量模型、分词器）
pip3 install paddlenlp -i https://pypi.tuna.tsinghua.edu.cn/simple

# 安装 HDFS 读取依赖（让 Python 能直接读 HDFS 数据）
pip3 install hdfs3 -i https://pypi.tuna.tsinghua.edu.cn/simple

总结（核心下一步）

1. 你已完成「Hadoop+Spark 环境搭建」，现在核心是跑通数据链路：本地数据 → HDFS 存储 → Spark 清洗 → 输出干净数据；
2. 上述步骤已帮你完成数据清洗，接下来只需：
   • 验证 HDFS 清洗后的数据是否正确；
   • 安装飞桨依赖后，编写模型训练脚本（读取清洗后的数据，训练轻量文本分类模型）；
3. 数据链路是「Hadoop → 大模型」的桥梁，这一步跑通后，后续模型训练、轻量化、端侧部署都是纯代码层面的操作，无环境依赖问题。

如果验证数据没问题，我可以直接给你模型训练的完整代码（读取 HDFS 清洗后的数据，训练 ERNIE Mini 轻量模型），复制就能跑，需要吗？