使用datatrove的LambdaFilter过滤URL列表

在数据处理流程中，我们经常需要对包含URL的文档进行过滤。datatrove项目提供了一个灵活的LambdaFilter组件，可以方便地实现这一功能。

核心思路

datatrove的LambdaFilter允许我们通过自定义函数来实现复杂的过滤逻辑。对于URL过滤场景，我们可以：

1. 从文本文件中读取需要过滤的URL列表
2. 使用tldextract库解析URL的各个组成部分
3. 将文档中的URL与过滤列表进行比对

实现方案

from datatrove.data import Document
from datatrove.executor.local import LocalPipelineExecutor
from datatrove.pipeline.filters.lambda_filter import LambdaFilter
from tldextract import TLDExtract

def create_url_filter(url_list_file):
    """创建基于URL列表的过滤器
    
    Args:
        url_list_file: 包含需要过滤URL的文本文件路径
    """
    with open(url_list_file, "r") as f:
        urls = f.read().splitlines()
        tldextractor = TLDExtract()

    def filter_func(doc: Document):
        """实际的过滤函数"""
        url = doc.metadata.get("url")
        if not url:
            raise ValueError("文档元数据中缺少URL信息")
            
        url_info = tldextractor(url)
        return url_info.fqdn not in urls

    return filter_func

使用示例

# 示例文档列表
sample_docs = [
    Document(
        metadata={"text": "示例1", "url": "https://example.com"},
        text="示例内容1",
        id="1"
    ),
    Document(
        metadata={"text": "示例2", "url": "https://test.org"}, 
        text="示例内容2",
        id="2"
    )
]

# 创建并执行管道
LocalPipelineExecutor([
    lambda x,y,z: sample_docs,  # 数据输入
    LambdaFilter(create_url_filter("filter_urls.txt"))  # URL过滤
]).run()

技术要点

1. tldextract库：用于准确解析URL的各个部分（域名、后缀等），比简单的字符串匹配更可靠

2. 闭包技巧：使用函数嵌套来缓存URL列表和解析器实例，避免重复加载

3. 错误处理：检查文档是否包含必要的URL元数据，避免运行时错误

4. 灵活性：可以轻松调整过滤逻辑，例如支持通配符或正则表达式匹配

扩展应用

这种方法不仅适用于URL过滤，稍作修改后可以用于：

• 基于域名的内容分类
• 特定网站的内容提取
• 垃圾内容过滤
• 数据来源分析

通过datatrove的管道机制，这种过滤器可以方便地与其他数据处理步骤组合，构建完整的数据处理流程。