The AI Data Lake Service provides multimodal data preprocessing capabilities. By parsing and cleaning text data, it helps users extract high-quality training corpora from massive amounts of raw web data, transforming raw text data into cleaned datasets that meet large model training standards. This article introduces the advantages and working principles of the text cleaning solution, and guides you through completing text data cleaning tasks using both the console and code.

Background information #

In the data-driven era, text data is the core foundation for enterprises to gain insights into requirements, for academic exploration of cutting-edge topics, and for intelligent system interactions. However, in reality, text data commonly suffers from the problem of "dirty data," which is manifested as follows:

1. Text obtained through web crawling contains a large number of HTML tags and special characters.
2. Manually entered data contains spelling errors and inconsistent formatting.
3. Data from different sources shows significant differences in terminology and encoding.

These issues can directly lead to biased data analysis results, degraded large model performance, and hinder the extraction of data value. As a key step in data preprocessing, text cleaning can transform messy raw text into clean and well-structured data, laying the foundation for subsequent analysis and applications. Therefore, it is especially important in scenarios such as large model training.

Application scenarios #

The text cleaning solution of the AI Data Lake Service is suitable for various critical scenarios, covering the entire process from data preprocessing to security assurance:

1. Large-scale data preprocessing: Batch clean and filter large-scale text from open data sources such as CommonCrawl or third-party corpora to build high-quality pre-training corpora and provide compliant training data for large language models.
2. Text language identification: Achieve high-precision multilingual identification using the FastText model, supporting 176 languages to meet the needs of multilingual text classification processing.
3. Text quality assessment: Assess the quality of existing text datasets using multidimensional metrics (such as perplexity, character ratio, and more), and filter out low-quality text.
4. Data security assurance: Leverage model capabilities to filter sensitive or non-compliant text according to laws and regulations, improving the purity and security of large-scale corpora.

Key advantages #

This solution offers significant advantages in terms of functional completeness, processing efficiency, and flexibility. The specific features are as follows:

• End-to-end process coverage: Supports an end-to-end processing workflow from CommonCrawl raw data to the final training corpus, covering all stages such as text extraction, deduplication, cleaning, and scoring.
• Multidimensional quality assessment: Integrates more than 15 text quality operators, including character ratio, duplication rate, perplexity, security, and more, to comprehensively capture data characteristics and ensure corpus quality.
• Intelligent language identification: Utilizes high-precision multilingual identification based on the FastText model, enabling independent processing branches and filtering criteria for different languages.
• Efficient deduplication mechanism: Implements precise content deduplication using MD5 hashing, effectively removing redundant text and saving storage and computing resources.
• Flexible filtering strategy: Decouples computation from filtering, allowing users to configure filtering thresholds as needed to meet different scenario requirements.
• Distributed processing capability: Based on the multithreaded parallel scheduling mechanism provided by Daft, enables high-performance cleaning and transformation of terabyte-scale raw data.
• Optimized dual pipeline for Chinese and English: Designs independent processing branches for English and Chinese corpora, with dedicated cleaning strategies and operator combinations tailored to language characteristics.
• Pluggable operator system: All data cleaning steps are implemented using LAS composable operators, supporting flexible adjustment, replacement, and extension of specific processing modules, and allowing additions via UDF.

Prerequisites #

• You have registered a and a . If this is your first time logging in to the , please complete first.
• has been enabled.
• have been created.
• has been created.

Working principle #

The solution for cleaning pre-training data for large language models adopts a multi-stage pipeline architecture, which includes core steps such as text extraction, precise deduplication, language identification, text cleaning, and quality evaluation.

Core processing flow #

Case demonstration #

Input file	Cleaned data
初始数据.json Unknown size

Development debugging #

Environment description #

The development machine is a professional development environment tailored by the AI Data Lake Service for algorithm developers. It features high efficiency and convenience, enabling developers to quickly start writing, debugging, and running data processing tasks. Developers use the Daft distributed computing framework on the LAS development machine to write a complete text cleaning pipeline.
You can use a terminal or IDE such as VSCode on your local or other devices to connect to the development machine remotely via SSH direct connection, and quickly start writing, debugging, and running data processing tasks. For operation guidance, see Remote connection to development machine.
The pipeline logic includes: using the CommonCrawl content extraction operator to process complex WARC files, achieving precise deduplication with md5 hashing, using the FastText model for high-precision multilingual identification, performing multidimensional text cleaning by invoking various text quality evaluation operators based on language, evaluating perplexity and quality scores, and writing the filtered results into high-performance Parquet datasets.

Operation steps #

1. Remotely connect to the development machine. For operation guidance, see Remote connection to development machine.

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

2. Create a demo file.

# Create demo.py file
touch demo.py
# Edit file
vim demo.py 
# After entering, press i to switch to edit mode. When finished, press Esc, enter :wq to save and exit.

The code is written in Python and builds a complete cleaning pipeline based on the Daft framework. The core logic includes environment configuration, five-stage processing functions, and result saving. Key code is as follows:

• demo.py code example

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"
    """
    
    # Obtain TOS path configuration from environment variables
    input_tos_dir = os.getenv("INPUT_TOS_DIR")
    output_tos_dir = os.getenv("OUTPUT_TOS_DIR")
    model_path = "/opt/las/models"
    
    # Check required environment variables
    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)
    
    # Configure TOS access
    tos_config = TOSConfig.from_env()
    io_config = daft.io.IOConfig(s3=tos_config.to_s3_config())

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

    # Stage 1: Text extraction - Extract web page content from WARC files, remove HTML tags and irrelevant information
    df = text_extraction(input_tos_dir, output_tos_dir, io_config)

    # Stage 2: Exact deduplication - Perform exact deduplication based on MD5 hash algorithm, remove completely identical text content
    df = exact_deduplication(df)

    # Stage 3: Language identification - Identify text language and provide confidence score, facilitating subsequent cleaning strategies by language
    df_en, df_zh = language_identification(df, model_path)

    # Stage 4: Text cleaning - Multi-dimensional metric evaluation and filtering for Chinese and English data
    df_en_cleaned = text_cleaning_english(df_en)
    df_zh_cleaned = text_cleaning_chinese(df_zh, model_path)

    # Step 5: Quality evaluation - Use models to evaluate the perplexity and quality score of the text
    df_en_final = quality_assessment(df_en_cleaned, "en", model_path)
    df_zh_final = quality_assessment(df_zh_cleaned, "zh", model_path)

    # Save results to 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. Environment variable settings.

Users must place the files to be cleaned in the TOS path in advance:

sample.warc.gz

Unknown size

export TOS_ENDPOINT="https://tos-cn-beijing.volces.com"
export LAS_TOS_ACCESS_KEY="your_access_key"  # Replace with the actual Access Key
export LAS_TOS_SECRET_KEY="your_secret_key"  # Replace with the actual Secret Key
export INPUT_TOS_DIR="tos://your_bucket/input_data"  # Enter the WARC file path
export OUTPUT_TOS_DIR="tos://your_bucket/output_data"  # Output path for cleaned data

4. Start the task.

python demo.py

Processing results #

(.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执行完成!

Description of result data fields

Full corpus construction #

After the sample data on the development machine has been successfully validated, follow these steps to perform TB-level full corpus construction:

1. Save the debugged processing pipeline as a custom image.
2. Create a batch task in LAS task management and select the above custom image.
3. Configure task parameters (such as input and output paths, computing resources, and so on). The task will automatically and efficiently process the entire raw web data and ultimately generate high-quality training corpora.