拓冰网站定制 
Daft Common Crawl 实战教程从网页语料到句子级文本嵌入的完整流水线【免费下载链接】DaftHigh-performance data engine for AI and multimodal workloads. Process images, audio, video, and structured data at any scale项目地址: https://gitcode.com/GitHub_Trending/da/DaftCommon Crawl 是规模最大的开放网络数据集之一包含超过 2500 亿个网页、横跨 18 年的抓取历史自 2020 年以来已成为生成式 AI 训练数据的核心来源——GPT-3 等模型所用训练数据的绝大部分都来自 Common Crawl。本文以 Daft 仓库中的教程 docs/examples/common-crawl-daft-tutorial.md 为主体完整演示如何用daft.datasets.common_crawl加载抓取数据、理解其 WARC/WET 数据模式、用 UDF 做语言过滤与句子切分并用本地小模型批量生成文本嵌入。读完并照做后你将掌握一条从原始网页语料到可入库向量库/Parquet嵌入数据的端到端 Daft 流水线。一、环境准备安装 Daft 与依赖教程使用的所有依赖可以通过一条命令安装。daft[transformers]额外带上 Transformers 相关依赖spacy用于后续的句子切分!pip install uv uv pip install daft[transformers] spacy以下是教程用到的完整导入。注意PYTORCH_CUDA_ALLOC_CONF必须在导入 torch之前设置用于缓解 GPU 显存碎片问题import os # We need to do this _before_ importing torch os.environ[PYTORCH_CUDA_ALLOC_CONF] expandable_segments:True import json from collections.abc import Iterator from datetime import datetime from typing import TypedDict import spacy import spacy.cli import torch from transformers import AutoConfig import daft from daft import col二、用daft.datasets.common_crawl加载抓取数据Daft 通过内置函数daft.datasets.common_crawl提供对 Common Crawl 的访问参数与底层行为在 daft/datasets/common_crawl.py 中完整定义。教程中的基本用法IN_AWS False # Set this to True if youre running in the us-east-1 AWS region. # For Google Colab, this must be set to False. df daft.datasets.common_crawl( crawlCC-MAIN-2025-33, # The specific crawl to get. Crawls are listed on the # Common Crawl website. This one was crawled ~ Spring 2025. contentwarc, # Options are raw, text, metadata, warc, wet, wat. # These change the names, types, and number of columns youll get # and, importantly, what kind of data youll be able to work with! num_files1, # Optional number of files to fetch. None means get all files. # Fetching a single file is useful for rapid prototyping. in_awsIN_AWS, # Lets Daft select the optimal download location! You **MUST** # set this to False if running outside of us-east-1 in AWS. )2.1 参数详解结合 daft/datasets/common_crawl.py 的函数签名与文档字符串各参数含义如下参数类型/默认值说明crawlstr抓取批次标识如CC-MAIN-2025-332025 年春季抓取segmentstr \| None默认None指定 crawl 内的某个 segment每个 crawl 被切分为 100 个 segmentNone表示全部 segmentcontent默认raw可选raw/text/metadata/warc/wet/wat分别对应 WARC 原始记录、WET 抽取文本、WAT 元数据num_filesint \| None默认None限制要处理的文件数量None表示处理全部匹配文件设为1非常适合快速原型验证io_configIOConfig \| None存储访问配置如 S3 凭证、HuggingFace tokenin_awsbool默认False已弃用源码中标注将在 v0.9.0 移除请改用sources3sources3 / hf / http \| None数据源选择None时自动尝试 HuggingFace找不到则回退 HTTP几个值得注意的实现细节num_files有正数校验num_files 0会直接抛ValueError见 daft/datasets/common_crawl.py。in_aws弃用迁移源码中当in_awsTrue时会发出warnings.warn并自动转换为sources3若同时传了sourcein_aws优先并告警。因此新代码建议直接写sources3在 AWS us-east-1 内或sourcehfAWS 之外最省成本。2.2 数据源解析manifest 文件如何被定位从源码结构看daft.datasets.common_crawl并非直接硬编码文件列表而是先读取各数据源的manifest 清单文件再解析出实际文件路径。_get_mainfest_path函数daft/datasets/common_crawl.py按数据源拼出清单地址数据源清单路径文件前缀S3s3://commoncrawl/crawl-data/{crawl}/{file_type}.paths.gzs3://commoncrawl/HuggingFace默认首选hf://buckets/commoncrawl/commoncrawl/crawl-data/{crawl}/{file_type}.paths.gzhf://buckets/commoncrawl/commoncrawl/HTTPShttps://data.commoncrawl.org/crawl-data/{crawl}/{file_type}.paths.gzhttps://data.commoncrawl.org/_get_common_crawl_pathsdaft/datasets/common_crawl.py用daft.read_text读取清单拼接前缀得到完整 URL再按需做segment过滤contains和num_files截断limit。如果首选的 HuggingFace 源抛FileNotFoundError比如该 crawl 尚未同步到 HF bucket会自动回退到sourcehttp。content参数通过一张映射表落到具体的文件类型daft/datasets/common_crawl.pycontent_type_map {raw: warc, text: wet, metadata: wat, warc: warc, wet: wet, wat: wat}最后一步是调用read_warc(warc_paths, io_configio_config)生成 DataFrame。三、理解数据模式WARC 全记录 vs 抽取文本先看看contentwarc即 raw WARC里有什么# Show is a materializing operation in Daft. It will cause the Daft query to run. # It only computes the first 8 records and then prints them to STDOUT. df.show()WARC 文件信息量非常大每一行对应一条 WARC 记录WARC-Record-IDuuid每条记录的唯一标识WARC-Type记录类型——是元数据是发往网站的原始请求还是网站的完整响应取值如warcinfo、request、response、metadata、conversionWARC-DateTimestamp这条记录何时被抓取Content-Lengthint响应体的长度WARC-Identified-Payload-Type已识别出的响应负载类型MIMEwarc_content网站内容或响应/元数据等的原始字节warc_headers一个 JSON 对象字符串包含Content-Type、WARC-Block-Digest、WARC-Refers-To、WARC-Target-URI即被爬取的网站 URL等。这个模式与 daft/io/_warc.py 中read_warc声明的 schema 一致WARC-Record-ID为DataType.uuid()WARC-Target-URI/WARC-Type/WARC-Identified-Payload-Type/warc_headers为字符串WARC-Date为纳秒级 UTC 时间戳Content-Length为Int64warc_content为Binary。如果需要非常精细的 WARC 级细节请求头、响应状态、块摘要等就用contentwarc。但本教程只需要网页的文本不需要 HTML因此改用contenttextWET 数据——这也意味着要处理和下载的数据量更小。3.1 只取文本contenttext与try_decode下面是contenttext数据的处理与预览。这里用col(X)构建引用某列的表达式df_sample ( daft.datasets.common_crawl( crawlCC-MAIN-2025-33, contenttext, num_files1, in_awsIN_AWS, ) # we only care about website responses that were converted into # this text content we selected for in the Common Crawl data .where(col(WARC-Type) conversion) # try to decode the byte content as UTF-8 encoded text .with_column(warc_content, col(warc_content).try_decode(utf-8)) # failed decodes result in a None value -- we remove these records .drop_null(col(warc_content)) .select(WARC-Record-ID, WARC-Target-URI, WARC-Date, Content-Length, warc_content, warc_headers) ) df_sample.show()两个要点WARC-Type conversion过滤WET 文件中网页响应经过抽取转换后的记录类型为conversion其他类型warcinfo、request等不是正文文本try_decode(utf-8)容错解码warc_content是 Binary 列try_decode尝试按 UTF-8 解码为字符串解码失败的行得到None而不是抛异常随后用drop_null清掉这些记录。预览结果可以看到Common Crawl 包含多种语言的网页。为了聚焦教程只处理英文页面扩展成多语言留给读者练习。四、UDF 解析warc_headers过滤英文页面语言信息藏在warc_headers这个 JSON 字符串列的WARC-Identified-Content-Language字段里。用一个daft.func标量 UDF 解析 JSON再取值过滤WarcHeaders TypedDict( WarcHeaders, { Content-Type: str, WARC-Block-Digest: str, WARC-Identified-Content-Language: str, WARC-Refers-To: str, WARC-Target-URI: str, }, ) daft.func def json_load_warc_headers(x: str) - WarcHeaders: return json.loads(x) df_lang df_sample.with_column(warc_headers, json_load_warc_headers(col(warc_headers))).with_column( language, col(warc_headers).get(WARC-Identified-Content-Language) ) df_lang.select(warc_content, language).show()这里能观察到两个重要现象eng表示英文English同一条记录可能带多个语言标记language是列表。为简化处理只保留全部文本均为英文的记录df_lang df_lang.where(col(language) eng) df_lang.select(warc_content, language).show()五、嵌入流水线为下游任务准备 Common Crawl熟悉数据集之后进入核心目标——生成文本嵌入embedding。嵌入是对数据文本、图像、音频等的数值向量表示编码了语义信息可用于语义检索、去重、多语言应用等场景实践中常把嵌入连同标识性元数据一起存入向量数据库。要生成嵌入首先要把网页文本拆成有意义的片段chunk。文本是分层的Document → Sections → Paragraphs → Sentences → Words → Characters切分策略取决于用途句子级大多数场景通用尤其当文档结构不明确或不一致时网页正是如此所以本教程选它段落级适合 RAG检索增强生成这类需要跨句保持上下文的场景章节级适合结构清晰划分的长文档固定长度实现简单但可能在任意边界切断语义。5.1 全局配置变量教程把所有可调参数集中定义并做正数/非空校验方便复现和调参######## CONFIGURATION: Options ######## MAX_SEQ_LEN_SPACY: int 1_000 # Maximum text length for sentence splitting. NLP_MODEL_NAME: str en_core_web_sm # spaCy model for sentence detection CHUNKING_PARALLELISM: int 4 # Parallel chunking processes MAX_SEQ_LEN_SENTENCE_TRANSFORMER: int 1024 * 1 # Maximum text length for any individual embedding. EMBEDDING_MODEL_NAME: str Qwen/Qwen3-Embedding-0.6B # Text embedding model EMBEDDING_BATCH_SIZE: int 16 # Batch size for embeddings EMBEDDING_SIZE: int AutoConfig.from_pretrained(EMBEDDING_MODEL_NAME).hidden_size ######## CONFIGURATION: Validation ######## if MAX_SEQ_LEN_SPACY 0: raise ValueError(fMAX_SEQ_LEN_SPACY must be positive! {MAX_SEQ_LEN_SPACY}) if len(NLP_MODEL_NAME) 0: raise ValueError(NLP_MODEL_NAME must be specified!) if CHUNKING_PARALLELISM 0: raise ValueError(fCHUNKING_PARALLELISM must be positive! {CHUNKING_PARALLELISM}) if MAX_SEQ_LEN_SENTENCE_TRANSFORMER 0: raise ValueError(fMAX_SEQ_LEN_SENTENCE_TRANSFORMER must be positive! {MAX_SEQ_LEN_SENTENCE_TRANSFORMER}) if len(EMBEDDING_MODEL_NAME) 0: raise ValueError(EMBEDDING_MODEL_NAME must be specified!) if EMBEDDING_BATCH_SIZE 0: raise ValueError(fEMBEDDING_BATCH_SIZE must be positive! {EMBEDDING_BATCH_SIZE})各参数职责MAX_SEQ_LEN_SPACY1000 字符限制送入 spaCy 做句子检测的文本长度MAX_SEQ_LEN_SENTENCE_TRANSFORMER1024限制单个送入嵌入模型的句子长度EMBEDDING_BATCH_SIZE16是 GPU 推理批大小EMBEDDING_SIZE则直接从Qwen3-Embedding-0.6B的AutoConfig.hidden_size读取保证下游声明的嵌入维度与模型真实输出一致。5.2 下载 spaCy 模型在切分之前先下载句子检测模型。注意要在主流程之外一次性下载而不是在 UDF 内部下载——因为 Daft 可能对同一个 UDF 创建多个实例多进程并发下载会产生竞态条件try: spacy.cli.download(NLP_MODEL_NAME) except: print(fERROR: Invalid spacy model name: {NLP_MODEL_NAME}) raise5.3 句子切分 UDFdaft.cls spaCyDaft 的类式 UDF 由daft.cls装饰器定义其参数与语义见 daft/udf/init.py。教程使用max_concurrency1, use_processTrueuse_processTrue让每个类实例在独立进程中运行隔离 spaCy 的 GIL 竞争与 C 扩展max_concurrency1限制该类 UDF 的并发实例数模型加载昂贵用类式 UDF 可以在多行数据间复用一次初始化。class TextChunk(TypedDict): text: str chunk_id: int daft.cls(max_concurrency1, use_processTrue) class ChunkingUDF: Chunks text into sentences using Spacy. def __init__(self) - None: # ensure model is already present via: # python -m spacy download {NLP_MODEL_NAME} # Or via Python: # spacy.cli.download(NLP_MODEL_NAME) # We **DONT** download it here otherwise we could have a race # condition as Daft _can_ make multiple copies of our UDF. self.nlp spacy.load(NLP_MODEL_NAME) daft.method def __call__(self, text: str) - Iterator[TextChunk]: n_truncated_spacy 0 n_truncated_sentence_transformer 0 if len(text) MAX_SEQ_LEN_SPACY: n_truncated_spacy 1 text text[:MAX_SEQ_LEN_SPACY] doc self.nlp(text) for i, sentence in enumerate(doc.sents): if len(sentence.text) MAX_SEQ_LEN_SENTENCE_TRANSFORMER: s_text sentence.text[:MAX_SEQ_LEN_SENTENCE_TRANSFORMER] n_truncated_sentence_transformer 1 else: s_text sentence.text chunked_text TextChunk(texts_text, chunk_idi) yield chunked_text if n_truncated_spacy 0: print(fTruncated {n_truncated_spacy} sentences that were longer than {MAX_SEQ_LEN_SPACY} characters.) if n_truncated_sentence_transformer 0: print( fTruncated {n_truncated_sentence_transformer} sentences that were longer than {MAX_SEQ_LEN_SENTENCE_TRANSFORMER} characters. )这个方法的关键点用Iterator[TextChunk]一行产出多行一个网页会被展开为多个句子每条带自增chunk_id两级截断保护先截到MAX_SEQ_LEN_SPACY再进 spaCy切出的每个句子若超过MAX_SEQ_LEN_SENTENCE_TRANSFORMER也截断并对截断数量打印统计方便监控数据质量。在小样本上看效果chunker ChunkingUDF() df_chunk ( df_lang.with_column(chunks, chunker(col(warc_content))) # and we want to see each objects fields as their own columns .with_column(text_chunk, col(chunks).get(text)) .with_column(text_index, col(chunks).get(chunk_id)) .select(WARC-Record-ID, WARC-Target-URI, WARC-Date, text_chunk, text_index) ) df_chunk.show()切分后chunks是一个 Struct 列textchunk_id用.get(text)/.get(chunk_id)拆成独立列。5.4 批量文本嵌入 UDFsentence-transformers Qwen3有了句子级文本就可以生成嵌入了。Daft 让在数据上跑模型变得非常直接再写一个类式 UDF用本地运行的sentence-transformers模型计算嵌入。嵌入列的类型显式声明为daft.DataType.embedding(float32, EMBEDDING_SIZE)这让 Daft 在计划阶段就知道输出是固定维度的向量from sentence_transformers import SentenceTransformer daft.cls(max_concurrency1, use_processTrue) class EmbedderUDF: def __init__(self): self.device cuda if torch.cuda.is_available() else cpu self.model SentenceTransformer(EMBEDDING_MODEL_NAME).to(self.device) self.model self.model.eval() self.model.compile() daft.method.batch( return_dtypedaft.DataType.embedding(daft.DataType.float32(), EMBEDDING_SIZE), batch_sizeEMBEDDING_BATCH_SIZE, ) def embed_text(self, texts): with torch.inference_mode(): embeddings self.model.encode( texts, batch_sizeEMBEDDING_BATCH_SIZE, output_valuesentence_embedding, precisionfloat32, show_progress_barFalse, convert_to_numpyTrue, ) return embeddings要点daft.method.batch(batch_sizeEMBEDDING_BATCH_SIZE)批式方法Daft 会把batch_size行文本攒成一批交给embed_text模型按批推理GPU 利用率远高于逐行调用torch.inference_mode()关闭梯度计算self.model.compile()启用 torch.compile 加速设备自动选择 CUDA/CPUreturn_dtype声明嵌入向量维度与AutoConfig...hidden_size动态读取的EMBEDDING_SIZE呼应无需手填魔法数字。在小样本上预览嵌入列( df_chunk.with_column(embedding, EmbedderUDF().embed_text(col(text_chunk))) .select(WARC-Record-ID, WARC-Target-URI, text_chunk, text_index, embedding) .show() )六、组装完整流水线把各部分串起来就是一条从 Common Crawl 原始数据到句子嵌入的完整查询计划chunker ChunkingUDF() embedder EmbedderUDF() df ( daft.datasets.common_crawl( crawlCC-MAIN-2025-33, segmentNone, contenttext, num_files10, # INCREASE THIS NUMBER TO RUN ON MORE CRAWL FILES # OR REMOVE IT / SET IT TO None TO RUN ON ALL FILES! in_awsIN_AWS, ) # only run on actual website text .where(col(WARC-Type) conversion) # UTF-8 decode the text .with_column(text, col(warc_content).try_decode(utf-8)) .drop_null(col(text)) # extract the language filter english pages only .with_column(warc_headers, json_load_warc_headers(col(warc_headers))) .with_column(language, col(warc_headers).get(WARC-Identified-Content-Language)) .where(col(language) eng) # chunk text into sentences .into_batches(batch_sizeEMBEDDING_BATCH_SIZE * 10) .with_column(sentences, chunker(col(text))) .with_column(text, col(sentences).get(text)) .with_column(chunk_id, col(sentences).get(chunk_id)) .exclude(sentences) # perform text embedding using the GPU .into_batches(batch_sizeEMBEDDING_BATCH_SIZE) .with_column(embedding, embedder.embed_text(col(text))) # our final columns .select(WARC-Record-ID, WARC-Target-URI, WARC-Date, chunk_id, text, embedding) )其中两处into_batches是性能关键。into_batches实现见 daft/dataframe/dataframe.py按目标行数重新切分分区其启发式是攒够batch_size * 0.8行就发出一个批次以处理效率优先、批次大小近似为准。教程的编排逻辑是切分前into_batches(EMBEDDING_BATCH_SIZE * 10)——给 spaCy 切分提供更大批次减少 UDF 调用开销嵌入前into_batches(EMBEDDING_BATCH_SIZE)——由于切分把一行网页膨胀成多行句子这里重新按嵌入批大小组批正好对齐daft.method.batch的batch_sizeselect收尾只保留WARC-Record-ID、WARC-Target-URI、WARC-Date、chunk_id、text、embedding六列形成句子文本 嵌入向量 溯源元数据的最终形态。预览整体结果df.show()七、把结果落地write_parquet真实场景中必须把产出保存下来。daft.DataFrame.write_parquet接受本地路径或 S3 keyDaft 会自动写分区文件start datetime.now() output df.write_parquet(./local_chunked_cc_text_and_embeddings) end datetime.now() print(fComplete! Took {end-start} -- Wrote output partitions:\n{output})输出目录中的每个分区 Parquet 都带有嵌入列与溯源列记录 ID、URL、抓取时间可直接加载到向量数据库或后续的去重、检索流程中。八、小结与延伸阅读本教程串起了四个核心技术点均可在 Daft 仓库中溯源验证环节API源码位置数据接入daft.datasets.common_crawldaft/datasets/common_crawl.pyWARC 解析daft.read_warcdaft/io/_warc.py标量/类式 UDFdaft.func、daft.cls、daft.method.batchdaft/udf/init.py批次重分区DataFrame.into_batchesdaft/dataframe/dataframe.py注意事项与适用前提该 API 处于 beta 阶段可能随 Common Crawl 数据集演进而变化在 AWS us-east-1 内运行必须传 S3 源in_awsTrue新写法sources3以获得最优路径在 AWS 之外运行 S3 会产生出口流量费教程中IN_AWS必须设为Falsein_aws参数已弃用、将在 v0.9.0 移除新代码建议直接使用sources3 / hf / http。进一步可阅读 Common Crawl 数据源参考文档三种数据源的访问方式、认证配置与 WARC/WET/WAT 加载差异以及教程对应的 Jupyter 版本。【免费下载链接】DaftHigh-performance data engine for AI and multimodal workloads. Process images, audio, video, and structured data at any scale项目地址: https://gitcode.com/GitHub_Trending/da/Daft创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Daft Common Crawl 实战教程：从网页语料到句子级文本嵌入的完整流水线
Daft Common Crawl 实战教程从网页语料到句子级文本嵌入的完整流水线【免费下载链接】DaftHigh-performance data engine for AI and multimodal workloads. Process images, audio, video, and structured data at any scale项目地址: https://gitcode.com/GitHub_Trending/da/DaftCommon Crawl 是规模最大的开放网络数据集之一包含超过 2500 亿个网页、横跨 18 年的抓取历史自 2020 年以来已成为生成式 AI 训练数据的核心来源——GPT-3 等模型所用训练数据的绝大部分都来自 Common Crawl。本文以 Daft 仓库中的教程 docs/examples/common-crawl-daft-tutorial.md 为主体完整演示如何用daft.datasets.common_crawl加载抓取数据、理解其 WARC/WET 数据模式、用 UDF 做语言过滤与句子切分并用本地小模型批量生成文本嵌入。读完并照做后你将掌握一条从原始网页语料到可入库向量库/Parquet嵌入数据的端到端 Daft 流水线。一、环境准备安装 Daft 与依赖教程使用的所有依赖可以通过一条命令安装。daft[transformers]额外带上 Transformers 相关依赖spacy用于后续的句子切分!pip install uv uv pip install daft[transformers] spacy以下是教程用到的完整导入。注意PYTORCH_CUDA_ALLOC_CONF必须在导入 torch之前设置用于缓解 GPU 显存碎片问题import os # We need to do this _before_ importing torch os.environ[PYTORCH_CUDA_ALLOC_CONF] expandable_segments:True import json from collections.abc import Iterator from datetime import datetime from typing import TypedDict import spacy import spacy.cli import torch from transformers import AutoConfig import daft from daft import col二、用daft.datasets.common_crawl加载抓取数据Daft 通过内置函数daft.datasets.common_crawl提供对 Common Crawl 的访问参数与底层行为在 daft/datasets/common_crawl.py 中完整定义。教程中的基本用法IN_AWS False # Set this to True if youre running in the us-east-1 AWS region. # For Google Colab, this must be set to False. df daft.datasets.common_crawl( crawlCC-MAIN-2025-33, # The specific crawl to get. Crawls are listed on the # Common Crawl website. This one was crawled ~ Spring 2025. contentwarc, # Options are raw, text, metadata, warc, wet, wat. # These change the names, types, and number of columns youll get # and, importantly, what kind of data youll be able to work with! num_files1, # Optional number of files to fetch. None means get all files. # Fetching a single file is useful for rapid prototyping. in_awsIN_AWS, # Lets Daft select the optimal download location! You **MUST** # set this to False if running outside of us-east-1 in AWS. )2.1 参数详解结合 daft/datasets/common_crawl.py 的函数签名与文档字符串各参数含义如下参数类型/默认值说明crawlstr抓取批次标识如CC-MAIN-2025-332025 年春季抓取segmentstr \| None默认None指定 crawl 内的某个 segment每个 crawl 被切分为 100 个 segmentNone表示全部 segmentcontent默认raw可选raw/text/metadata/warc/wet/wat分别对应 WARC 原始记录、WET 抽取文本、WAT 元数据num_filesint \| None默认None限制要处理的文件数量None表示处理全部匹配文件设为1非常适合快速原型验证io_configIOConfig \| None存储访问配置如 S3 凭证、HuggingFace tokenin_awsbool默认False已弃用源码中标注将在 v0.9.0 移除请改用sources3sources3 / hf / http \| None数据源选择None时自动尝试 HuggingFace找不到则回退 HTTP几个值得注意的实现细节num_files有正数校验num_files 0会直接抛ValueError见 daft/datasets/common_crawl.py。in_aws弃用迁移源码中当in_awsTrue时会发出warnings.warn并自动转换为sources3若同时传了sourcein_aws优先并告警。因此新代码建议直接写sources3在 AWS us-east-1 内或sourcehfAWS 之外最省成本。2.2 数据源解析manifest 文件如何被定位从源码结构看daft.datasets.common_crawl并非直接硬编码文件列表而是先读取各数据源的manifest 清单文件再解析出实际文件路径。_get_mainfest_path函数daft/datasets/common_crawl.py按数据源拼出清单地址数据源清单路径文件前缀S3s3://commoncrawl/crawl-data/{crawl}/{file_type}.paths.gzs3://commoncrawl/HuggingFace默认首选hf://buckets/commoncrawl/commoncrawl/crawl-data/{crawl}/{file_type}.paths.gzhf://buckets/commoncrawl/commoncrawl/HTTPShttps://data.commoncrawl.org/crawl-data/{crawl}/{file_type}.paths.gzhttps://data.commoncrawl.org/_get_common_crawl_pathsdaft/datasets/common_crawl.py用daft.read_text读取清单拼接前缀得到完整 URL再按需做segment过滤contains和num_files截断limit。如果首选的 HuggingFace 源抛FileNotFoundError比如该 crawl 尚未同步到 HF bucket会自动回退到sourcehttp。content参数通过一张映射表落到具体的文件类型daft/datasets/common_crawl.pycontent_type_map {raw: warc, text: wet, metadata: wat, warc: warc, wet: wet, wat: wat}最后一步是调用read_warc(warc_paths, io_configio_config)生成 DataFrame。三、理解数据模式WARC 全记录 vs 抽取文本先看看contentwarc即 raw WARC里有什么# Show is a materializing operation in Daft. It will cause the Daft query to run. # It only computes the first 8 records and then prints them to STDOUT. df.show()WARC 文件信息量非常大每一行对应一条 WARC 记录WARC-Record-IDuuid每条记录的唯一标识WARC-Type记录类型——是元数据是发往网站的原始请求还是网站的完整响应取值如warcinfo、request、response、metadata、conversionWARC-DateTimestamp这条记录何时被抓取Content-Lengthint响应体的长度WARC-Identified-Payload-Type已识别出的响应负载类型MIMEwarc_content网站内容或响应/元数据等的原始字节warc_headers一个 JSON 对象字符串包含Content-Type、WARC-Block-Digest、WARC-Refers-To、WARC-Target-URI即被爬取的网站 URL等。这个模式与 daft/io/_warc.py 中read_warc声明的 schema 一致WARC-Record-ID为DataType.uuid()WARC-Target-URI/WARC-Type/WARC-Identified-Payload-Type/warc_headers为字符串WARC-Date为纳秒级 UTC 时间戳Content-Length为Int64warc_content为Binary。如果需要非常精细的 WARC 级细节请求头、响应状态、块摘要等就用contentwarc。但本教程只需要网页的文本不需要 HTML因此改用contenttextWET 数据——这也意味着要处理和下载的数据量更小。3.1 只取文本contenttext与try_decode下面是contenttext数据的处理与预览。这里用col(X)构建引用某列的表达式df_sample ( daft.datasets.common_crawl( crawlCC-MAIN-2025-33, contenttext, num_files1, in_awsIN_AWS, ) # we only care about website responses that were converted into # this text content we selected for in the Common Crawl data .where(col(WARC-Type) conversion) # try to decode the byte content as UTF-8 encoded text .with_column(warc_content, col(warc_content).try_decode(utf-8)) # failed decodes result in a None value -- we remove these records .drop_null(col(warc_content)) .select(WARC-Record-ID, WARC-Target-URI, WARC-Date, Content-Length, warc_content, warc_headers) ) df_sample.show()两个要点WARC-Type conversion过滤WET 文件中网页响应经过抽取转换后的记录类型为conversion其他类型warcinfo、request等不是正文文本try_decode(utf-8)容错解码warc_content是 Binary 列try_decode尝试按 UTF-8 解码为字符串解码失败的行得到None而不是抛异常随后用drop_null清掉这些记录。预览结果可以看到Common Crawl 包含多种语言的网页。为了聚焦教程只处理英文页面扩展成多语言留给读者练习。四、UDF 解析warc_headers过滤英文页面语言信息藏在warc_headers这个 JSON 字符串列的WARC-Identified-Content-Language字段里。用一个daft.func标量 UDF 解析 JSON再取值过滤WarcHeaders TypedDict( WarcHeaders, { Content-Type: str, WARC-Block-Digest: str, WARC-Identified-Content-Language: str, WARC-Refers-To: str, WARC-Target-URI: str, }, ) daft.func def json_load_warc_headers(x: str) - WarcHeaders: return json.loads(x) df_lang df_sample.with_column(warc_headers, json_load_warc_headers(col(warc_headers))).with_column( language, col(warc_headers).get(WARC-Identified-Content-Language) ) df_lang.select(warc_content, language).show()这里能观察到两个重要现象eng表示英文English同一条记录可能带多个语言标记language是列表。为简化处理只保留全部文本均为英文的记录df_lang df_lang.where(col(language) eng) df_lang.select(warc_content, language).show()五、嵌入流水线为下游任务准备 Common Crawl熟悉数据集之后进入核心目标——生成文本嵌入embedding。嵌入是对数据文本、图像、音频等的数值向量表示编码了语义信息可用于语义检索、去重、多语言应用等场景实践中常把嵌入连同标识性元数据一起存入向量数据库。要生成嵌入首先要把网页文本拆成有意义的片段chunk。文本是分层的Document → Sections → Paragraphs → Sentences → Words → Characters切分策略取决于用途句子级大多数场景通用尤其当文档结构不明确或不一致时网页正是如此所以本教程选它段落级适合 RAG检索增强生成这类需要跨句保持上下文的场景章节级适合结构清晰划分的长文档固定长度实现简单但可能在任意边界切断语义。5.1 全局配置变量教程把所有可调参数集中定义并做正数/非空校验方便复现和调参######## CONFIGURATION: Options ######## MAX_SEQ_LEN_SPACY: int 1_000 # Maximum text length for sentence splitting. NLP_MODEL_NAME: str en_core_web_sm # spaCy model for sentence detection CHUNKING_PARALLELISM: int 4 # Parallel chunking processes MAX_SEQ_LEN_SENTENCE_TRANSFORMER: int 1024 * 1 # Maximum text length for any individual embedding. EMBEDDING_MODEL_NAME: str Qwen/Qwen3-Embedding-0.6B # Text embedding model EMBEDDING_BATCH_SIZE: int 16 # Batch size for embeddings EMBEDDING_SIZE: int AutoConfig.from_pretrained(EMBEDDING_MODEL_NAME).hidden_size ######## CONFIGURATION: Validation ######## if MAX_SEQ_LEN_SPACY 0: raise ValueError(fMAX_SEQ_LEN_SPACY must be positive! {MAX_SEQ_LEN_SPACY}) if len(NLP_MODEL_NAME) 0: raise ValueError(NLP_MODEL_NAME must be specified!) if CHUNKING_PARALLELISM 0: raise ValueError(fCHUNKING_PARALLELISM must be positive! {CHUNKING_PARALLELISM}) if MAX_SEQ_LEN_SENTENCE_TRANSFORMER 0: raise ValueError(fMAX_SEQ_LEN_SENTENCE_TRANSFORMER must be positive! {MAX_SEQ_LEN_SENTENCE_TRANSFORMER}) if len(EMBEDDING_MODEL_NAME) 0: raise ValueError(EMBEDDING_MODEL_NAME must be specified!) if EMBEDDING_BATCH_SIZE 0: raise ValueError(fEMBEDDING_BATCH_SIZE must be positive! {EMBEDDING_BATCH_SIZE})各参数职责MAX_SEQ_LEN_SPACY1000 字符限制送入 spaCy 做句子检测的文本长度MAX_SEQ_LEN_SENTENCE_TRANSFORMER1024限制单个送入嵌入模型的句子长度EMBEDDING_BATCH_SIZE16是 GPU 推理批大小EMBEDDING_SIZE则直接从Qwen3-Embedding-0.6B的AutoConfig.hidden_size读取保证下游声明的嵌入维度与模型真实输出一致。5.2 下载 spaCy 模型在切分之前先下载句子检测模型。注意要在主流程之外一次性下载而不是在 UDF 内部下载——因为 Daft 可能对同一个 UDF 创建多个实例多进程并发下载会产生竞态条件try: spacy.cli.download(NLP_MODEL_NAME) except: print(fERROR: Invalid spacy model name: {NLP_MODEL_NAME}) raise5.3 句子切分 UDFdaft.cls spaCyDaft 的类式 UDF 由daft.cls装饰器定义其参数与语义见 daft/udf/init.py。教程使用max_concurrency1, use_processTrueuse_processTrue让每个类实例在独立进程中运行隔离 spaCy 的 GIL 竞争与 C 扩展max_concurrency1限制该类 UDF 的并发实例数模型加载昂贵用类式 UDF 可以在多行数据间复用一次初始化。class TextChunk(TypedDict): text: str chunk_id: int daft.cls(max_concurrency1, use_processTrue) class ChunkingUDF: Chunks text into sentences using Spacy. def __init__(self) - None: # ensure model is already present via: # python -m spacy download {NLP_MODEL_NAME} # Or via Python: # spacy.cli.download(NLP_MODEL_NAME) # We **DONT** download it here otherwise we could have a race # condition as Daft _can_ make multiple copies of our UDF. self.nlp spacy.load(NLP_MODEL_NAME) daft.method def __call__(self, text: str) - Iterator[TextChunk]: n_truncated_spacy 0 n_truncated_sentence_transformer 0 if len(text) MAX_SEQ_LEN_SPACY: n_truncated_spacy 1 text text[:MAX_SEQ_LEN_SPACY] doc self.nlp(text) for i, sentence in enumerate(doc.sents): if len(sentence.text) MAX_SEQ_LEN_SENTENCE_TRANSFORMER: s_text sentence.text[:MAX_SEQ_LEN_SENTENCE_TRANSFORMER] n_truncated_sentence_transformer 1 else: s_text sentence.text chunked_text TextChunk(texts_text, chunk_idi) yield chunked_text if n_truncated_spacy 0: print(fTruncated {n_truncated_spacy} sentences that were longer than {MAX_SEQ_LEN_SPACY} characters.) if n_truncated_sentence_transformer 0: print( fTruncated {n_truncated_sentence_transformer} sentences that were longer than {MAX_SEQ_LEN_SENTENCE_TRANSFORMER} characters. )这个方法的关键点用Iterator[TextChunk]一行产出多行一个网页会被展开为多个句子每条带自增chunk_id两级截断保护先截到MAX_SEQ_LEN_SPACY再进 spaCy切出的每个句子若超过MAX_SEQ_LEN_SENTENCE_TRANSFORMER也截断并对截断数量打印统计方便监控数据质量。在小样本上看效果chunker ChunkingUDF() df_chunk ( df_lang.with_column(chunks, chunker(col(warc_content))) # and we want to see each objects fields as their own columns .with_column(text_chunk, col(chunks).get(text)) .with_column(text_index, col(chunks).get(chunk_id)) .select(WARC-Record-ID, WARC-Target-URI, WARC-Date, text_chunk, text_index) ) df_chunk.show()切分后chunks是一个 Struct 列textchunk_id用.get(text)/.get(chunk_id)拆成独立列。5.4 批量文本嵌入 UDFsentence-transformers Qwen3有了句子级文本就可以生成嵌入了。Daft 让在数据上跑模型变得非常直接再写一个类式 UDF用本地运行的sentence-transformers模型计算嵌入。嵌入列的类型显式声明为daft.DataType.embedding(float32, EMBEDDING_SIZE)这让 Daft 在计划阶段就知道输出是固定维度的向量from sentence_transformers import SentenceTransformer daft.cls(max_concurrency1, use_processTrue) class EmbedderUDF: def __init__(self): self.device cuda if torch.cuda.is_available() else cpu self.model SentenceTransformer(EMBEDDING_MODEL_NAME).to(self.device) self.model self.model.eval() self.model.compile() daft.method.batch( return_dtypedaft.DataType.embedding(daft.DataType.float32(), EMBEDDING_SIZE), batch_sizeEMBEDDING_BATCH_SIZE, ) def embed_text(self, texts): with torch.inference_mode(): embeddings self.model.encode( texts, batch_sizeEMBEDDING_BATCH_SIZE, output_valuesentence_embedding, precisionfloat32, show_progress_barFalse, convert_to_numpyTrue, ) return embeddings要点daft.method.batch(batch_sizeEMBEDDING_BATCH_SIZE)批式方法Daft 会把batch_size行文本攒成一批交给embed_text模型按批推理GPU 利用率远高于逐行调用torch.inference_mode()关闭梯度计算self.model.compile()启用 torch.compile 加速设备自动选择 CUDA/CPUreturn_dtype声明嵌入向量维度与AutoConfig...hidden_size动态读取的EMBEDDING_SIZE呼应无需手填魔法数字。在小样本上预览嵌入列( df_chunk.with_column(embedding, EmbedderUDF().embed_text(col(text_chunk))) .select(WARC-Record-ID, WARC-Target-URI, text_chunk, text_index, embedding) .show() )六、组装完整流水线把各部分串起来就是一条从 Common Crawl 原始数据到句子嵌入的完整查询计划chunker ChunkingUDF() embedder EmbedderUDF() df ( daft.datasets.common_crawl( crawlCC-MAIN-2025-33, segmentNone, contenttext, num_files10, # INCREASE THIS NUMBER TO RUN ON MORE CRAWL FILES # OR REMOVE IT / SET IT TO None TO RUN ON ALL FILES! in_awsIN_AWS, ) # only run on actual website text .where(col(WARC-Type) conversion) # UTF-8 decode the text .with_column(text, col(warc_content).try_decode(utf-8)) .drop_null(col(text)) # extract the language filter english pages only .with_column(warc_headers, json_load_warc_headers(col(warc_headers))) .with_column(language, col(warc_headers).get(WARC-Identified-Content-Language)) .where(col(language) eng) # chunk text into sentences .into_batches(batch_sizeEMBEDDING_BATCH_SIZE * 10) .with_column(sentences, chunker(col(text))) .with_column(text, col(sentences).get(text)) .with_column(chunk_id, col(sentences).get(chunk_id)) .exclude(sentences) # perform text embedding using the GPU .into_batches(batch_sizeEMBEDDING_BATCH_SIZE) .with_column(embedding, embedder.embed_text(col(text))) # our final columns .select(WARC-Record-ID, WARC-Target-URI, WARC-Date, chunk_id, text, embedding) )其中两处into_batches是性能关键。into_batches实现见 daft/dataframe/dataframe.py按目标行数重新切分分区其启发式是攒够batch_size * 0.8行就发出一个批次以处理效率优先、批次大小近似为准。教程的编排逻辑是切分前into_batches(EMBEDDING_BATCH_SIZE * 10)——给 spaCy 切分提供更大批次减少 UDF 调用开销嵌入前into_batches(EMBEDDING_BATCH_SIZE)——由于切分把一行网页膨胀成多行句子这里重新按嵌入批大小组批正好对齐daft.method.batch的batch_sizeselect收尾只保留WARC-Record-ID、WARC-Target-URI、WARC-Date、chunk_id、text、embedding六列形成句子文本 嵌入向量 溯源元数据的最终形态。预览整体结果df.show()七、把结果落地write_parquet真实场景中必须把产出保存下来。daft.DataFrame.write_parquet接受本地路径或 S3 keyDaft 会自动写分区文件start datetime.now() output df.write_parquet(./local_chunked_cc_text_and_embeddings) end datetime.now() print(fComplete! Took {end-start} -- Wrote output partitions:\n{output})输出目录中的每个分区 Parquet 都带有嵌入列与溯源列记录 ID、URL、抓取时间可直接加载到向量数据库或后续的去重、检索流程中。八、小结与延伸阅读本教程串起了四个核心技术点均可在 Daft 仓库中溯源验证环节API源码位置数据接入daft.datasets.common_crawldaft/datasets/common_crawl.pyWARC 解析daft.read_warcdaft/io/_warc.py标量/类式 UDFdaft.func、daft.cls、daft.method.batchdaft/udf/init.py批次重分区DataFrame.into_batchesdaft/dataframe/dataframe.py注意事项与适用前提该 API 处于 beta 阶段可能随 Common Crawl 数据集演进而变化在 AWS us-east-1 内运行必须传 S3 源in_awsTrue新写法sources3以获得最优路径在 AWS 之外运行 S3 会产生出口流量费教程中IN_AWS必须设为Falsein_aws参数已弃用、将在 v0.9.0 移除新代码建议直接使用sources3 / hf / http。进一步可阅读 Common Crawl 数据源参考文档三种数据源的访问方式、认证配置与 WARC/WET/WAT 加载差异以及教程对应的 Jupyter 版本。【免费下载链接】DaftHigh-performance data engine for AI and multimodal workloads. Process images, audio, video, and structured data at any scale项目地址: https://gitcode.com/GitHub_Trending/da/Daft创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考