AI 数据湖服务提供多模态数据预处理功能，它通过解析、清洗文本数据，帮助用户从海量原始网络数据中提取高质量训练语料，将原始文本数据转化为符合大模型训练标准的清洗数据集。本文介绍文本清洗方案的优势特性与工作原理，并指导您通过控制台和代码完成文本数据清洗任务。

背景信息

在数据驱动时代，文本数据是企业洞察需求、学术探索前沿、智能系统交互的核心基础，但现实中的文本数据普遍存在 “脏数据” 问题，具体表现为：

1. 网络爬取的文本夹杂大量 HTML 标签、特殊字符。
2. 人工录入的数据存在拼写错误、格式不统一。
3. 不同来源数据在术语、编码上差异显著。

这些问题会直接导致数据分析结果偏差、大模型性能下降，阻碍数据价值挖掘。而文本清洗作为数据预处理的关键环节，能将杂乱的原始文本转化为干净、规整的数据，为后续分析与应用奠定基础，因此在大模型训练等场景中尤为重要。

应用场景

AI 数据湖服务的文本清洗方案适用于多类关键场景，覆盖从数据预处理到安全保障的全流程需求：

1. 大规模数据预处理：从 CommonCrawl 等开放数据源或第三方语料库中，批量清洗过滤大规模文本，构建高质量预训练语料，为大语言模型提供合规训练数据。
2. 文本语种识别：借助 FastText 模型实现高精度多语种识别，支持 176 种语言，满足多语言文本分类处理需求。
3. 文本质量评估：通过多维度指标（如困惑度、字符比例等）对现有文本数据集进行质量评估和筛选，剔除低质量文本。
4. 数据安全性保障：依托模型能力过滤敏感或违反法律法规的文本，提升大规模语料的纯净度和安全性。

优势特性

该方案在功能完整性、处理效率、灵活性等方面具备显著优势，具体特性如下：

• 全流程覆盖：支持从 CommonCrawl 原始数据到最终训练语料的端到端处理流程，覆盖文本提取、去重、清洗、打分等各个阶段
• 多维度质量评估：集成15+种文本质量算子，包括字符比例、重复度、困惑度、安全性等，多维度把握数据特征，全面保障语料质量。
• 智能语种识别：基于 FastText 模型的高精度多语种识别，以便针对不同语种配置独立的处理链路分支和过滤标准。
• 高效去重机制：采用 MD5 哈希实现精确内容去重，有效剔除冗余文本，节省存储和计算资源。
• 灵活过滤策略：计算与过滤解耦，支持用户按需配置过滤阈值，以便于满足不同场景需求。
• 分布式处理能力：依托 Daft 提供的多线程并行调度机制，可实现 TB 级原始数据的高性能清洗与转换。
• 中英文双链路优化：针对英文与中文语料分别设计独立处理分支，针对语言特点定制专属清洗策略和算子组合。
• 可插拔算子体系：所有数据清洗步骤基于 LAS 可组合算子实现，支持灵活调整、替换、扩展具体处理模块，也可以通过UDF 新增。

前提条件

• 已注册火山引擎账号。如果您是首次登录火山引擎控制台，请先完成实名认证。
• 已开通AI数据湖服务产品。
• 已创建通用队列、计算队列。
• 已创建开发机。

工作原理

大语言模型预训练数据清洗解决方案采用多阶段流水线架构，包含文本提取、精确去重、语种识别、文本清洗、质量评估等核心环节。

核心处理流程

案例展示

输入文件	清洗后数据
初始数据.json 未知大小

开发调试

环境说明

开发机是由 AI 数据湖服务为算法开发者量身打造的专业开发环境，具备高效便捷的特性，开发者可借助其快速开启数据处理任务的编写、调试及运行流程。开发者在 LAS 开发机 上，使用 Daft 分布式计算框架编写一个完整的文本清洗管道。
您可以在本地或者其他设备上使用终端或者 VSCode 等 IDE 远程使用 SSH 直连的方式连接开发机，快速开始数据处理任务编写、调试和运行，操作指导详见远程连接开发机。
管道逻辑包括：调用CommonCrawl内容提取算子处理复杂的WARC文件，借助md5哈希实现精确去重，调用FastText模型实现高精度多语种识别，根据语种调用多种不同的文本质量评估算子进行多维度文本清洗，进行困惑度、质量分的评估后，将过滤后的结果写入高性能的Parquet数据集。

操作步骤

1. 远程连接开发机，操作指导详见远程连接开发机。

ssh -p {端口号} root@{IP地址}

2. 创建demo文件。

# 创建demo.py文件
touch demo.py
#编辑文件
vim demo.py 
#进入后按 i切换到编辑模式，完成后按 Esc，输入 :wq保存并退出。

代码采用 Python 编写，基于 Daft 框架构建完整清洗管道，核心逻辑包含环境配置、5 个阶段处理函数及结果保存，关键代码如下：

• demo.py 代码示例

from __future__ import annotations

import glob
import logging
def configure_logging():
    logging.basicConfig(
        level=logging.INFO, 
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        datefmt='%Y-%m-%d %H:%M:%S.%s'.format("%f"))

    logging.getLogger("tracing.span").setLevel(logging.WARNING)
    logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
    logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
    logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
    logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)
    logging.getLogger("trafilatura").setLevel(logging.CRITICAL)
    logging.getLogger("lxml").setLevel(logging.CRITICAL)
    logging.getLogger("fasttext").setLevel(logging.WARNING)
    logging.getLogger("sentencepiece").setLevel(logging.WARNING)

configure_logging()
logger = logging.getLogger(__name__)

import os
from datetime import datetime
from io import StringIO

import daft
from daft import col
from daft.las.io.tos import TOSConfig
from daft.las.functions.text.commoncrawl_content_extractor import CommonCrawlContentExtractor
from daft.las.functions.text.md5_calculator import Md5Calculator
from daft.las.functions.text.language_recognition import LanguageRecognitionOperator
from daft.las.functions.text.alphanumeric_ratio_calculator import AlphanumericRatioCalculator
from daft.las.functions.text.url_ratio_calculator import UrlRatioCalculator
from daft.las.functions.text.special_characters_ratio_calculator import SpecialCharactersRatioCalculator
from daft.las.functions.text.copyright_cleaner import CopyrightCleaner
from daft.las.functions.text.repeated_lines_calculator import RepeatedLinesCalculator
from daft.las.functions.text.text_length_calculator import TextLengthCalculator
from daft.las.functions.text.maximum_word_length_calculator import MaximumWordLengthCalculator
from daft.las.functions.text.word_repetition_calculator import WordRepetitionCalculator
from daft.las.functions.text.bullet_line_ratio_calculator import BulletLineRatioCalculator
from daft.las.functions.text.whitespace_normalizer import WhitespaceNormalizer
from daft.las.functions.text.perplexity_calculator import PerplexityCalculator
from daft.las.functions.text.en_text_quality_scorer import EnTextQualityScorer
from daft.las.functions.text.multilingual_text_quality_scorer import MultilingualTextQualityScorer
from daft.las.functions.text.text_safety_scorer import TextSafetyScorer
from daft.las.functions.text.chinese_text_converter import ChineseTextConverter
from daft.las.functions.udf import las_udf

def text_extraction(tos_directory: str, output_tos_dir: str, io_config) -> daft.DataFrame:
    """阶段1：文本提取

    从WARC格式文件中提取网页内容，去除HTML标签和无关信息。
    使用parser解析WARC文件，提取纯文本内容，过滤掉导航、广告等噪声信息。
    提供trafilatura、goose3、justext这三种可选parser。

    Args:
        tos_directory: TOS WARC文件所在目录

    Returns:
        包含提取内容的DataFrame
    """
    logger.info("开始文本提取...")
    

    input_s3_dir = tos_directory.replace("tos://", "s3://", 1)
    tos_config = TOSConfig.from_env()
    io_config = daft.io.IOConfig(s3=tos_config.to_s3_config())

    df = daft.from_glob_path(
        f"{input_s3_dir}/**/*.warc.gz",
        io_config=io_config,
    )
    
    df = df.with_column(
        "warc_file",
        col("path").str.replace("s3://", "tos://")
    )
    
    df = df.exclude("path", "size", "num_rows")
    logger.info("初始df: %d 行", df.count_rows())

    df = df.with_column(
        "extracted_content",
        las_udf(
            CommonCrawlContentExtractor,
            construct_args={
                "warc_src_type": "warc_url",
                "extractor_type": "trafilatura",
                "max_records": None,
            },
            num_gpus=0,
            batch_size=1,
            concurrency=1,
        )(col("warc_file")),
    )

    df = df.explode("extracted_content") 
    df = df.with_column("url", df["extracted_content"]["url"])
    df = df.with_column("content", df["extracted_content"]["content"])
    df = df.with_column("warc_file", df["extracted_content"]["warc_file"])
    df = df.exclude("extracted_content")
    df = df.filter(col("content").not_null())
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_s3_dir = output_tos_dir.replace("tos://", "s3://", 1)
    unwashed_s3_path = f"{output_s3_dir}unwashed_parquet_{timestamp}"
    
    df.write_parquet(unwashed_s3_path, io_config=io_config)
    logger.info("内容提取结果已保存到TOS: %s", output_tos_dir + f"unwashed_parquet_{timestamp}")
    
    df = daft.read_parquet(unwashed_s3_path, io_config=io_config)

    logger.info("内容提取完成: %d 条记录", df.count_rows())
    return df

def exact_deduplication(df: daft.DataFrame) -> daft.DataFrame:
    """阶段2：精确去重

    基于MD5哈希算法进行精确去重，去除完全相同的文本内容。
    对每条文本计算MD5哈希值，通过哈希值比较快速识别并去除重复内容。
    该方法高效且准确，能够显著减少数据冗余。

    Args:
        df: 包含文本内容的DataFrame

    Returns:
        去重后的DataFrame
    """
    logger.info("开始精确去重...")

    df = df.with_column(
        "content_md5",
        las_udf(
            Md5Calculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )

    df = df.distinct("content_md5")
    

    logger.info("MD5去重后: %d 条记录", df.count_rows())
    return df

def language_identification(df: daft.DataFrame, model_path: str) -> daft.DataFrame:
    """阶段3：语种识别

    识别文本语种并且给出置信度，便于后续按语种实行不同的清洗策略。
    基于FastText模型识别文本语言并计算置信度，支持176种语言识别。

    Args:
        df: 包含文本内容的DataFrame

    Returns:
        包含语言识别结果的DataFrame
    """
    logger.info("开始语种识别...")

    df = df.with_column(
        "language_result",
        las_udf(
            LanguageRecognitionOperator,
            construct_args={
                "model_path": model_path,
                "model_name": "fasttext/lid.176.bin",
                "batch_size": 128,
            },
            num_gpus=0,
            batch_size=128,
            concurrency=1,
        )(col("content")),
    )

    df = df.with_column("language", df["language_result"]["language"])
    df = df.with_column("confidence", df["language_result"]["confidence"])
    df = df.exclude("language_result")

    df_en = df.filter((col("language") == "en") & (col("confidence") >= 0.8))
    df_zh = df.filter((col("language") == "zh") & (col("confidence") >= 0.8))

    logger.info("英文数据过滤后: %d 条记录", df_en.count_rows())
    logger.info("中文数据过滤后: %d 条记录", df_zh.count_rows())

    return df_en, df_zh

def text_cleaning_english(df_en: daft.DataFrame) -> daft.DataFrame:
    """阶段4：英文文本清洗

    针对英文数据的多维度指标评估和过滤：
    - 字母数字比例过滤
    - URL比例过滤  
    - 特殊字符比例过滤
    - 版权声明清理
    - 重复行比例过滤
    - 文本长度过滤
    - 最大单词长度过滤
    - 词重复比例过滤
    - 项目符号行比例过滤
    - 空白字符标准化

    Args:
        df_en: 英文数据DataFrame

    Returns:
        清洗后的英文数据DataFrame
    """
    logger.info("开始英文文本清洗...")

    logger.info("英文分支 - 计算字母数字字符比例")
    df_en = df_en.with_column(
        "alphanumeric_ratio",
        las_udf(
            AlphanumericRatioCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter(col("alphanumeric_ratio") >= 0.5)
    logger.info("英文分支 - 字母数字字符过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "url_ratio",
        las_udf(
            UrlRatioCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter(col("url_ratio") <= 0.3)
    logger.info("英文分支 - URL比例过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "special_char_ratio",
        las_udf(
            SpecialCharactersRatioCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter(col("special_char_ratio") <= 0.4)
    logger.info("英文分支 - 特殊字符比例过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "cleaned_content",
        las_udf(
            CopyrightCleaner,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.with_column("content", col("cleaned_content"))
    df_en = df_en.exclude("cleaned_content")
    logger.info("英文分支 - 版权声明清理后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "repeated_lines_ratio",
        las_udf(
            RepeatedLinesCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter(col("repeated_lines_ratio") <= 0.3)
    logger.info("英文分支 - 重复行比例过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "text_length",
        las_udf(
            TextLengthCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter((col("text_length") >= 100) & (col("text_length") <= 100000))
    logger.info("英文分支 - 文本长度过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "max_word_length",
        las_udf(
            MaximumWordLengthCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter(col("max_word_length") <= 20)
    logger.info("英文分支 - 最大单词长度过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "word_repetition_ratio",
        las_udf(
            WordRepetitionCalculator,
            construct_args={
                "repetition": 2,
                "lang": "en",
                "tokenization": False,
            },
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter(col("word_repetition_ratio") <= 0.5)
    logger.info("英文分支 - 2-gram词重复比例过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "bullet_line_ratio",
        las_udf(
            BulletLineRatioCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.filter(col("bullet_line_ratio") <= 0.4)
    logger.info("英文分支 - 项目符号行比例过滤后: %d 条记录", df_en.count_rows())

    df_en = df_en.with_column(
        "normalized_content",
        las_udf(
            WhitespaceNormalizer,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_en.collect()
    df_en = df_en.with_column("content", col("normalized_content"))
    df_en = df_en.exclude("normalized_content")
    logger.info("英文分支 - 空白字符标准化后: %d 条记录", df_en.count_rows())

    logger.info("英文分支处理完成: %d 条记录", df_en.count_rows())
    return df_en

def text_cleaning_chinese(df_zh: daft.DataFrame, model_path: str) -> daft.DataFrame:
    """阶段4：中文文本清洗

    针对中文数据的多维度指标评估和过滤：
    - 字母数字比例过滤
    - URL比例过滤
    - 特殊字符比例过滤
    - 版权声明清理
    - 重复行比例过滤
    - 文本长度过滤
    - 简繁体统一
    - 项目符号行比例过滤
    - 安全性评估与过滤
    - 空白字符标准化

    Args:
        df_zh: 中文数据DataFrame

    Returns:
        清洗后的中文数据DataFrame
    """
    logger.info("开始中文文本清洗...")

    df_zh = df_zh.with_column(
        "alphanumeric_ratio",
        las_udf(
            AlphanumericRatioCalculator,
            construct_args={
                "tokenization": False,
            },
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.filter(col("alphanumeric_ratio") >= 0.3)
    logger.info("中文分支 - 字母数字字符过滤后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "url_ratio",
        las_udf(
            UrlRatioCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.filter(col("url_ratio") <= 0.3)
    logger.info("中文分支 - URL比例过滤后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "special_char_ratio",
        las_udf(
            SpecialCharactersRatioCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.filter(col("special_char_ratio") <= 0.6)
    logger.info("中文分支 - 特殊字符比例过滤后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "cleaned_content",
        las_udf(
            CopyrightCleaner,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.with_column("content", col("cleaned_content"))
    df_zh = df_zh.exclude("cleaned_content")
    logger.info("中文分支 - 版权声明清理后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "repeated_lines_ratio",
        las_udf(
            RepeatedLinesCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.filter(col("repeated_lines_ratio") <= 0.3)
    logger.info("中文分支 - 重复行比例过滤后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "text_length",
        las_udf(
            TextLengthCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.filter((col("text_length") >= 50) & (col("text_length") <= 8000))
    logger.info("中文分支 - 文本长度过滤后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "converted_content",
        las_udf(
            ChineseTextConverter,
            construct_args={
                "conversion_type": "t2s",
            },
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.with_column("content", col("converted_content"))
    df_zh = df_zh.exclude("converted_content")
    logger.info("中文分支 - 简繁转换后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "bullet_line_ratio",
        las_udf(
            BulletLineRatioCalculator,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.filter(col("bullet_line_ratio") <= 0.7)
    logger.info("中文分支 - 项目符号行比例过滤后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "safety_score",
        las_udf(
            TextSafetyScorer,
            construct_args={
                "lang": "zh",
                "model_path": model_path,
                "model_name": "thu-coai/ShieldLM-6B-chatglm3",
                "batch_size": 128,
                "rank": 0,
            },
            num_gpus=1,
            batch_size=128,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    logger.info("中文分支 - 安全性评分后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.filter(col("safety_score")["safe"] >= 0.5)
    logger.info("中文分支 - 安全内容过滤后: %d 条记录", df_zh.count_rows())

    df_zh = df_zh.with_column(
        "normalized_content",
        las_udf(
            WhitespaceNormalizer,
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df_zh.collect()
    df_zh = df_zh.with_column("content", col("normalized_content"))
    df_zh = df_zh.exclude("normalized_content")
    logger.info("中文分支 - 空白字符标准化后: %d 条记录", df_zh.count_rows())

    logger.info("中文分支处理完成: %d 条记录", df_zh.count_rows())
    return df_zh

def quality_assessment(df: daft.DataFrame, lang: str, model_path: str) -> daft.DataFrame:
    """阶段5：质量评估

    使用模型评估文本的困惑度和质量分：
    - 使用kenlm/wikipedia模型计算困惑度
    - 英文使用llm-data-textbook-quality-fasttext-classifier-v2模型计算质量分
    - 中文使用multilingual-e5-small-aligned-quality模型计算质量分

    Args:
        df: 待评估的DataFrame
        lang: 语言类型，"en"或"zh"

    Returns:
        包含质量评估结果的DataFrame
    """
    logger.info("开始%s质量评估...", lang)

    df = df.with_column(
        "perplexity",
        las_udf(
            PerplexityCalculator,
            construct_args={
                "lang": lang,
                "model_path": model_path,
                "model_name": "kenlm/wikipedia",
            },
            num_gpus=0,
            batch_size=64,
            concurrency=1,
        )(col("content")),
    )
    df.collect()
    logger.info("%s分支 - 困惑度计算后: %d 条记录", lang, df.count_rows())

    if lang == "en":
        df = df.with_column(
            "quality_score",
            las_udf(
                EnTextQualityScorer,
                construct_args={
                    "model_path": model_path,
                    "model_name": "llm-data-textbook-quality-fasttext-classifier-v2/model_quantized.bin",
                    "batch_size": 64,
                    "rank": 0,
                },
                num_gpus=0,
                batch_size=64,
                concurrency=1,
            )(col("content")),
        )
    else:
        df = df.with_column(
            "quality_score",
            las_udf(
                MultilingualTextQualityScorer,
                construct_args={
                    "model_path": model_path,
                    "model_name": "multilingual-e5-small-aligned-quality",
                    "dtype": "float32",
                    "batch_size": 64,
                    "rank": 0,
                },
                num_gpus=1,
                batch_size=64,
                concurrency=1,
            )(col("content")),
        )

    df.collect()
    logger.info("%s分支 - 质量评分后: %d 条记录", lang, df.count_rows())

    return df

def save_results(df_processed: daft.DataFrame, lang: str, output_tos_dir: str, io_config) -> None:
    """保存处理结果

    Args:
        df_processed: 处理后的DataFrame
        lang: 语言类型，"en"或"zh"
        output_tos_dir: TOS输出目录
        io_config: IO配置
    """
    if df_processed.count_rows() == 0:
        logger.info("%s分支无数据，跳过保存", lang)
        return

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    output_s3_dir = output_tos_dir.replace("tos://", "s3://", 1)
    parquet_s3_path = f"{output_s3_dir}{lang}_parquet_{timestamp}"
    
    df_processed.write_parquet(parquet_s3_path, io_config=io_config)
    logger.info("%s处理结果已保存到TOS: %s", lang, output_tos_dir + f"{lang}_parquet_{timestamp}")
    logger.info("%s最终结果: %d 条记录", lang, df_processed.count_rows())
    df_processed.show(5)

def main():
    """主函数 - 按阶段执行大语言模型预训练数据清洗解决方案demo
    
    运行前请设置以下环境变量：
    export TOS_ENDPOINT="https://tos-cn-beijing.volces.com" 
    export LAS_TOS_ACCESS_KEY="your_access_key"
    export LAS_TOS_SECRET_KEY="your_secret_key"
    export INPUT_TOS_DIR="tos://your_bucket/input_data"
    export OUTPUT_TOS_DIR="tos://your_bucket/output_data"
    """
    
    # 从环境变量获取TOS路径配置
    input_tos_dir = os.getenv("INPUT_TOS_DIR")
    output_tos_dir = os.getenv("OUTPUT_TOS_DIR")
    model_path = "/opt/las/models"
    
    # 检查必需的环境变量
    if not input_tos_dir:
        raise ValueError("请设置环境变量 INPUT_TOS_DIR，例如：export INPUT_TOS_DIR='tos://your_bucket/input_data'")
    if not output_tos_dir:
        raise ValueError("请设置环境变量 OUTPUT_TOS_DIR，例如：export OUTPUT_TOS_DIR='tos://your_bucket/output_data'")
    
    logger.info("输入路径: %s", input_tos_dir)
    logger.info("输出路径: %s", output_tos_dir)
    logger.info("模型路径: %s", model_path)
    
    # 配置TOS访问
    tos_config = TOSConfig.from_env()
    io_config = daft.io.IOConfig(s3=tos_config.to_s3_config())

    logger.info("大语言模型预训练数据清洗解决方案demo开始运行...")

    # 阶段1：文本提取 - 从 WARC 文件中提取网页内容，去除 HTML 标签和无关信息
    df = text_extraction(input_tos_dir, output_tos_dir, io_config)

    # 阶段2：精确去重 - 基于 MD5 哈希算法进行精确去重，去除完全相同的文本内容
    df = exact_deduplication(df)

    # 阶段3：语种识别 - 识别文本语种并给出置信度，便于后续按语种实行不同的清洗策略
    df_en, df_zh = language_identification(df, model_path)

    # 阶段4：文本清洗 - 针对中英文数据的多维度指标评估和过滤
    df_en_cleaned = text_cleaning_english(df_en)
    df_zh_cleaned = text_cleaning_chinese(df_zh, model_path)

    # 阶段5：质量评估 - 使用模型评估文本的困惑度和质量分
    df_en_final = quality_assessment(df_en_cleaned, "en", model_path)
    df_zh_final = quality_assessment(df_zh_cleaned, "zh", model_path)

    # 保存结果到TOS
    save_results(df_en_final, "en", output_tos_dir, io_config)
    save_results(df_zh_final, "zh", output_tos_dir, io_config)

    logger.info("大语言模型预训练数据清洗解决方案demo执行完成!")

if __name__ == "__main__":
    main()

3. 环境变量设置。

用户需提前待清洗文件放入TOS路径：

sample.warc.gz

未知大小

export TOS_ENDPOINT="https://tos-cn-beijing.volces.com"
export LAS_TOS_ACCESS_KEY="your_access_key"  # 替换为实际Access Key
export LAS_TOS_SECRET_KEY="your_secret_key"  # 替换为实际Secret Key
export INPUT_TOS_DIR="tos://your_bucket/input_data"  # 输入WARC文件路径
export OUTPUT_TOS_DIR="tos://your_bucket/output_data"  # 输出清洗后数据路径 

5. 启动任务。

python demo.py

处理结果

(.venv) user@n37-034-174:~/Daft$ python ~/llm_pretrain_data_clean_pipeline_demo.py 
2025-07-30 15:25:33,020 - INFO - 大语言模型预训练数据清洗解决方案demo开始运行...
2025-07-30 15:25:33,021 - INFO - 开始文本提取...
2025-07-30 15:25:33,119 - INFO - 发现 3 个 .warc.gz 文件
2025-07-30 15:25:33,125 - INFO - 初始df: 3 行
2025-07-30 15:26:13,701 - INFO - 内容提取完成: 900 条记录
2025-07-30 15:26:13,702 - INFO - 开始精确去重...
2025-07-30 15:26:14,210 - INFO - MD5去重后: 862 条记录
2025-07-30 15:26:14,211 - INFO - 开始语种识别...
🗡️ 🐟[1/2] ✓     InMemorySource | [00:00:00]                                                                                                                         W
2025-07-30 15:26:16,396 - INFO - 英文数据过滤后: 205 条记录
2025-07-30 15:26:16,417 - INFO - 中文数据过滤后: 152 条记录
2025-07-30 15:26:16,417 - INFO - 开始英文文本清洗...
2025-07-30 15:26:16,417 - INFO - 英文分支 - 计算字母数字字符比例
2025-07-30 15:26:16,562 - INFO - 英文分支 - 字母数字字符过滤后: 205 条记录
2025-07-30 15:26:16,664 - INFO - 英文分支 - URL比例过滤后: 205 条记录
2025-07-30 15:26:16,799 - INFO - 英文分支 - 特殊字符比例过滤后: 202 条记录
2025-07-30 15:26:16,918 - INFO - 英文分支 - 版权声明清理后: 202 条记录
2025-07-30 15:26:17,048 - INFO - 英文分支 - 重复行比例过滤后: 181 条记录
2025-07-30 15:26:17,158 - INFO - 英文分支 - 文本长度过滤后: 176 条记录
2025-07-30 15:26:17,320 - INFO - 英文分支 - 最大单词长度过滤后: 175 条记录
2025-07-30 15:26:43,009 - INFO - 英文分支 - 2-gram词重复比例过滤后: 170 条记录
2025-07-30 15:26:43,149 - INFO - 英文分支 - 项目符号行比例过滤后: 154 条记录
2025-07-30 15:26:43,326 - INFO - 英文分支 - 空白字符标准化后: 154 条记录
2025-07-30 15:26:43,333 - INFO - 英文分支处理完成: 154 条记录
2025-07-30 15:26:43,333 - INFO - 开始中文文本清洗...
2025-07-30 15:26:43,484 - INFO - 中文分支 - 字母数字字符过滤后: 152 条记录
2025-07-30 15:26:43,610 - INFO - 中文分支 - URL比例过滤后: 152 条记录
2025-07-30 15:26:43,748 - INFO - 中文分支 - 特殊字符比例过滤后: 150 条记录
2025-07-30 15:26:43,897 - INFO - 中文分支 - 版权声明清理后: 150 条记录
2025-07-30 15:26:44,052 - INFO - 中文分支 - 重复行比例过滤后: 135 条记录
2025-07-30 15:26:44,181 - INFO - 中文分支 - 文本长度过滤后: 132 条记录
2025-07-30 15:26:44,412 - INFO - 中文分支 - 简繁转换后: 132 条记录
2025-07-30 15:26:44,581 - INFO - 中文分支 - 项目符号行比例过滤后: 131 条记录
2025-07-30 15:31:11,155 - INFO - 中文分支 - 安全性评分后: 131 条记录
2025-07-30 15:31:11,169 - INFO - 中文分支 - 安全内容过滤后: 65 条记录
2025-07-30 15:31:11,311 - INFO - 中文分支 - 空白字符标准化后: 65 条记录
2025-07-30 15:31:11,317 - INFO - 中文分支处理完成: 65 条记录
2025-07-30 15:31:11,318 - INFO - 开始en质量评估...
2025-07-30 15:31:40,073 - INFO - en分支 - 困惑度计算后: 154 条记录
Warning : `load_model` does not return WordVectorModel or SupervisedModel any more, but a `FastText` object which is very similar.
2025-07-30 15:32:03,362 - INFO - en分支 - 质量评分后: 154 条记录
2025-07-30 15:32:03,362 - INFO - 开始zh质量评估...
2025-07-30 15:32:24,957 - INFO - zh分支 - 困惑度计算后: 65 条记录
2025-07-30 15:32:32,259 - INFO - zh分支 - 质量评分后: 65 条记录
2025-07-30 15:32:32,344 - INFO - en处理结果已保存到目录: /home/user/test-data/en_parquet_20250730_153232
2025-07-30 15:32:32,369 - INFO - en最终结果: 154 条记录
╭──────────────────┬─────────────────┬─────────────────┬──────────┬─────────────────┬────────────┬─────────────────┬─────────────────┬────────────┬─────────────────╮
│ url              ┆ content         ┆ content_md5     ┆ language ┆ confidence      ┆      …     ┆ word_repetition ┆ bullet_line_rat ┆ perplexity ┆ quality_score   │
│ ---              ┆ ---             ┆ ---             ┆ ---      ┆ ---             ┆            ┆ _ratio          ┆ io              ┆ ---        ┆ ---             │
│ Utf8             ┆ Utf8            ┆ Utf8            ┆ Utf8     ┆ Float64         ┆ (6 hidden) ┆ ---             ┆ ---             ┆ Float64    ┆ Float64         │
│                  ┆                 ┆                 ┆          ┆                 ┆            ┆ Float64         ┆ Float64         ┆            ┆                 │
╞══════════════════╪═════════════════╪═════════════════╪══════════╪═════════════════╪════════════╪═════════════════╪═════════════════╪════════════╪═════════════════╡
│ http://artdisbe. ┆ Social          ┆ 752f6ae03c28012 ┆ en       ┆ 0.9639100432395 ┆ …          ┆ 0.0370370370370 ┆ 0               ┆ 162        ┆ 0.9771882891654 │
│ info/tag/audi…   ┆ networking      ┆ 32dfd99f51455b… ┆          ┆ 935             ┆            ┆ 37035           ┆                 ┆            ┆ 968             │
│                  ┆ sites have …    ┆                 ┆          ┆                 ┆            ┆                 ┆                 ┆            ┆                 │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://actiona.t ┆ If you see this ┆ fae6747ee81be4d ┆ en       ┆ 0.9178508520126 ┆ …          ┆ 0               ┆ 0.1666666666666 ┆ 439.5      ┆ 1.4564456939697 │
│ ools/            ┆ page, the ngi…  ┆ 8ba0c154f8029f… ┆          ┆ 343             ┆            ┆                 ┆ 6666            ┆            ┆ 266             │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://205004.xo ┆ Voltman Founder ┆ 5f5f6b042493d21 ┆ en       ┆ 0.8400994539260 ┆ …          ┆ 0.2714285714285 ┆ 0               ┆ 3753.8     ┆ 0.9884101152420 │
│ bor.com/t1933…   ┆ Member Techni…  ┆ 48690c0eefdc1a… ┆          ┆ 864             ┆            ┆ 714             ┆                 ┆            ┆ 044             │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://andyjedyn ┆ TRADEMARK FREE  ┆ d0987d68f7f704a ┆ en       ┆ 0.8312416076660 ┆ …          ┆ 0.0851063829787 ┆ 0               ┆ 859        ┆ 0.5191992409527 │
│ ak.com/__medi…   ┆ ZONE - Network… ┆ 29bdbf6e8f823b… ┆          ┆ 156             ┆            ┆ 234             ┆                 ┆            ┆ 302             │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://debtrevie ┆ _DSC0244        ┆ cabb02c4ad7faed ┆ en       ┆ 0.9041053056716 ┆ …          ┆ 0               ┆ 0               ┆ 12662      ┆ 1.0322509855031 │
│ wawards.co.za…   ┆ Published on    ┆ ae9a9cb41f4235… ┆          ┆ 919             ┆            ┆                 ┆                 ┆            ┆ 967             │
│                  ┆ March 1…        ┆                 ┆          ┆                 ┆            ┆                 ┆                 ┆            ┆                 │
╰──────────────────┴─────────────────┴─────────────────┴──────────┴─────────────────┴────────────┴─────────────────┴─────────────────┴────────────┴─────────────────╯

(Showing first 5 rows)
2025-07-30 15:32:32,427 - INFO - zh处理结果已保存到目录: /home/user/test-data/zh_parquet_20250730_153232
2025-07-30 15:32:32,442 - INFO - zh最终结果: 65 条记录
╭──────────────────┬──────────────────┬──────────────────┬──────────┬─────────────────┬────────────┬─────────────────┬─────────────────┬────────────┬───────────────╮
│ url              ┆ content          ┆ content_md5      ┆ language ┆ confidence      ┆      …     ┆ bullet_line_rat ┆ safety_score    ┆ perplexity ┆ quality_score │
│ ---              ┆ ---              ┆ ---              ┆ ---      ┆ ---             ┆            ┆ io              ┆ ---             ┆ ---        ┆ ---           │
│ Utf8             ┆ Utf8             ┆ Utf8             ┆ Utf8     ┆ Float64         ┆ (5 hidden) ┆ ---             ┆ Struct[safe:    ┆ Float64    ┆ Float32       │
│                  ┆                  ┆                  ┆          ┆                 ┆            ┆ Float64         ┆ Float64,        ┆            ┆               │
│                  ┆                  ┆                  ┆          ┆                 ┆            ┆                 ┆ unsafe:         ┆            ┆               │
│                  ┆                  ┆                  ┆          ┆                 ┆            ┆                 ┆ Float64,        ┆            ┆               │
│                  ┆                  ┆                  ┆          ┆                 ┆            ┆                 ┆ controversial:  ┆            ┆               │
│                  ┆                  ┆                  ┆          ┆                 ┆            ┆                 ┆ Float64]        ┆            ┆               │
╞══════════════════╪══════════════════╪══════════════════╪══════════╪═════════════════╪════════════╪═════════════════╪═════════════════╪════════════╪═══════════════╡
│ http://dd99msa.l ┆ 中国侨联陈红2025 ┆ 8ae64db62e97ff85 ┆ zh       ┆ 0.9891796708106 ┆ …          ┆ 0               ┆ {safe: 0.570388 ┆ 1795.6     ┆ 0.37256837    │
│ fqxw.cn/?laiz…   ┆ -6-6 17:24:29    ┆ c8e0f24e215a6…   ┆          ┆ 995             ┆            ┆                 ┆ 3171081543,     ┆            ┆               │
│                  ┆ 在如今信息…      ┆                  ┆          ┆                 ┆            ┆                 ┆ un…             ┆            ┆               │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://aisouche. ┆ 全国 全国 请选择 ┆ 8b3dabe3a8e0e181 ┆ zh       ┆ 0.9661716222763 ┆ …          ┆ 0               ┆ {safe: 0.988518 ┆ 6582.1     ┆ 0.09692196    │
│ com/goods/det…   ┆ 请选择 维保ERP   ┆ 252b8620fa2ce…   ┆          ┆ 062             ┆            ┆                 ┆ 1188583374,     ┆            ┆               │
│                  ┆ 商城首页 EPC目…  ┆                  ┆          ┆                 ┆            ┆                 ┆ un…             ┆            ┆               │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://02y3tcpv. ┆ 查看更多相关内容 ┆ 9b51987a65b1cb7a ┆ zh       ┆ 0.9952771663665 ┆ …          ┆ 0               ┆ {safe: 0.650043 ┆ 1565.5     ┆ 0.1351457     │
│ gd9.cc/?pengl…   ┆ 取消关注在如今的 ┆ 8cc52b6606d9a…   ┆          ┆ 771             ┆            ┆                 ┆ 5471534729,     ┆            ┆               │
│                  ┆ 数字时代，游戏已 ┆                  ┆          ┆                 ┆            ┆                 ┆ un…             ┆            ┆               │
│                  ┆ 经成为众…        ┆                  ┆          ┆                 ┆            ┆                 ┆                 ┆            ┆               │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://bbs.archi ┆ 建筑相关国家考试 ┆ a97382c687154b42 ┆ zh       ┆ 0.8331785202026 ┆ …          ┆ 0.0789473684210 ┆ {safe: 0.513418 ┆ 4238.2     ┆ 0.08262762    │
│ .sdnl.org/vie…   ┆ 、建筑师考试准备 ┆ a95745ef7e3d8…   ┆          ┆ 367             ┆            ┆ 5263            ┆ 972492218,      ┆            ┆               │
│                  ┆ 、心得及经验分享 ┆                  ┆          ┆                 ┆            ┆                 ┆ uns…            ┆            ┆               │
│                  ┆ 。 由 *…         ┆                  ┆          ┆                 ┆            ┆                 ┆                 ┆            ┆               │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ http://23cx9u2.d ┆ 在当今数字时代， ┆ ea140a01f2106372 ┆ zh       ┆ 0.9973435997962 ┆ …          ┆ 0               ┆ {safe: 0.908772 ┆ 1613.4     ┆ 0.20907189    │
│ lqww.cn/?shao…   ┆ 体育爱好者们寻求 ┆ 6c53a021d85e8…   ┆          ┆ 952             ┆            ┆                 ┆ 7069854736,     ┆            ┆               │
│                  ┆ 的不仅仅是比赛结 ┆                  ┆          ┆                 ┆            ┆                 ┆ un…             ┆            ┆               │
│                  ┆ 果，更多的…      ┆                  ┆          ┆                 ┆            ┆                 ┆                 ┆            ┆               │
╰──────────────────┴──────────────────┴──────────────────┴──────────┴─────────────────┴────────────┴─────────────────┴─────────────────┴────────────┴───────────────╯

(Showing first 5 rows)
2025-07-30 15:32:32,450 - INFO - 日志已保存到: /home/user/test-data/log_20250730_153232.txt
2025-07-30 15:32:32,450 - INFO - 大语言模型预训练数据清洗解决方案demo执行完成!

结果数据字段说明

全量语料构建

当开发机上的样本数据验证无误后，可按以下步骤实现 TB 级全量语料构建：

1. 将调试完成的处理管道固化为自定义镜像。
2. 在 LAS 任务管理中创建批量任务，选择上述自定义镜像。
3. 配置任务参数（如输入输出路径、计算资源等），任务将自动、高效地处理全量原始网络数据，最终生成高质量训练语料。