本文定位：这是一篇专注于 Spring AI ETL Pipeline 的深度实战指南。ETL（Extract-Transform-Load）是 RAG 系统的数据预处理核心，本文将详细讲解如何使用 Spring AI 的 ETL 组件，将各种格式的原始文档转换为可检索的向量索引，为智能问答系统奠定数据基础。

官方文档参考：

目录

1. 官方文档说了什么？——ETL 全景图

官方原文：
“The ETL (Extract, Transform, Load) pipeline is a common pattern for processing data. In the context of Spring AI, ETL is used to prepare documents for vector storage and retrieval.”

翻译：ETL（提取、转换、加载）管道是处理数据的常见模式。在 Spring AI 的上下文中，ETL 用于为向量存储和检索准备文档。

Spring AI 的 ETL Pipeline 包含三个核心抽象：

Spring AI ETL Pipeline 架构
═══════════════════════════════════════════════

┌─────────────────────────────────────────────────┐
│  DocumentReader<T>  (Extract - 提取)             │
│  ├─ TextReader                                  │
│  ├─ PagePdfDocumentReader                       │
│  ├─ ParagraphPdfDocumentReader                  │
│  ├─ TikaDocumentReader                          │
│  ├─ JsonReader                                  │
│  ├─ MarkdownDocumentReader                      │
│  └─ ... (可扩展)                                │
└─────────────────────────────────────────────────┘
                    │
                    ▼ List<Document>
┌─────────────────────────────────────────────────┐
│  DocumentTransformer  (Transform - 转换)          │
│  ├─ TokenTextSplitter                           │
│  ├─ ContentFormatTransformer                    │
│  ├─ KeywordMetadataEnricher                     │
│  ├─ SummaryMetadataEnricher                     │
│  └─ ... (可扩展)                                │
└─────────────────────────────────────────────────┘
                    │
                    ▼ List<Document> (处理后)
┌─────────────────────────────────────────────────┐
│  DocumentWriter  (Load - 加载)                   │
│  ├─ VectorStore (主要实现)                       │
│  │   ├─ SimpleVectorStore                       │
│  │   ├─ ChromaVectorStore                       │
│  │   ├─ MilvusVectorStore                       │
│  │   └─ ...                                     │
│  └─ ... (其他存储方式)                           │
└─────────────────────────────────────────────────┘

核心设计原则：

  • 接口抽象：每个阶段都有清晰的接口定义
  • 组件化：可以独立使用或组合使用
  • 可扩展：支持自定义实现
  • 类型安全：泛型设计保证编译期类型检查

2. ETL Pipeline 核心概念

2.1 为什么需要 ETL？

在 RAG 系统中，原始文档无法直接用于向量检索，需要经过处理：

问题原始文档的困境ETL 解决方案
格式多样PDF、Word、网页、JSON…DocumentReader 统一抽象
文档太大几万字的技术文档TokenTextSplitter 分块
缺少元数据没有标题、作者、分类信息Transformer 添加元数据
检索困难文本无法直接相似度计算VectorStore 自动向量化

2.2 ETL 三阶段详解

阶段英文中文输入输出核心任务
Extract提取数据提取原始文件List<Document>解析文件格式，提取文本内容
Transform转换数据转换List<Document>List<Document>分块、清理、增强元数据
Load加载数据加载List<Document>存储完成向量化并存储到向量数据库

2.3 数据流转图解

原始文档生命周期：从文件到可检索向量
══════════════════════════════════════════════════════════════

📄 knowledge.pdf (2MB, 100页)
        │ DocumentReader.get()
        ▼
   ┌─────────────────────────────────────────────┐
   │  Document {                                 │
   │    id: "doc_001",                           │
   │    text: "完整PDF内容...(10万字)",            │
   │    metadata: {                              │
   │      source: "knowledge.pdf",               │
   │      pageCount: 100                         │
   │    }                                        │
   │  }                                          │
   └─────────────────────────────────────────────┘
        │ DocumentTransformer.apply()
        ▼
   ┌─────────────────────────────────────────────┐
   │  List<Document> [                           │
   │    Document { text: "第1块内容...", ... },  │
   │    Document { text: "第2块内容...", ... },  │
   │    ...                                      │
   │    Document { text: "第200块内容...", ... } │
   │  ]                                          │
   │  (200个文档块，每块约500字)                   │
   └─────────────────────────────────────────────┘
        │ VectorStore.add()
        ▼
   ┌─────────────────────────────────────────────┐
   │  向量数据库                                  │
   │  ┌─────────────────────────────────────┐    │
   │  │ 文档块1 → [0.1, 0.3, -0.2, ...]     │    │
   │  │ 文档块2 → [0.4, -0.1, 0.7, ...]     │    │
   │  │ ...                                │    │
   │  │ 文档块200 → [-0.2, 0.5, 0.1, ...]   │    │
   │  └─────────────────────────────────────┘    │
   │  现在可以进行相似度检索了！                   │
   └─────────────────────────────────────────────┘

3. Extract（提取）——DocumentReader 文档读取器

3.1 内置 Reader 实现一览

Reader 实现支持格式Maven 依赖特点
TextReader.txt, .mdspring-ai-starter轻量级，UTF-8 编码
PagePdfDocumentReader.pdfspring-ai-pdf-document-reader按页分割PDF
ParagraphPdfDocumentReader.pdfspring-ai-pdf-document-reader按段落分割PDF
TikaDocumentReaderWord, PPT, HTML, XMLspring-ai-tika-document-reader基于Apache Tika，支持50+格式
JsonReader.jsonspring-ai-starterJSONPath 提取，适合结构化数据
MarkdownDocumentReader.mdspring-ai-starterMarkdown 特定优化

3.2 TextReader：纯文本文件读取

最简单的文档读取器，适合纯文本文件：

// 基础用法
TextReader textReader = new TextReader(
    new ClassPathResource("knowledge/spring-ai-guide.txt")
);
List<Document> documents = textReader.get();

// 自定义字符集
TextReader reader = new TextReader(
    new FileSystemResource("/path/to/file.txt"), 
    StandardCharsets.UTF_8  // 显式指定编码
);

// 批量读取目录
public List<Document> readTextFiles(String directoryPath) throws IOException {
    List<Document> allDocs = new ArrayList<>();
    
    try (Stream<Path> paths = Files.walk(Paths.get(directoryPath))) {
        List<Path> textFiles = paths
            .filter(Files::isRegularFile)
            .filter(path -> path.toString().endsWith(".txt"))
            .toList();
        
        for (Path file : textFiles) {
            TextReader reader = new TextReader(new FileSystemResource(file.toFile()));
            List<Document> docs = reader.get();
            
            // 为每个文档添加文件元数据
            docs.forEach(doc -> {
                doc.getMetadata().put("filename", file.getFileName().toString());
                doc.getMetadata().put("filepath", file.toString());
                doc.getMetadata().put("filesize", file.toFile().length());
            });
            
            allDocs.addAll(docs);
        }
    }
    
    return allDocs;
}

3.3 PdfDocumentReader：PDF 文档处理

PDF 是企业文档的主流格式，Spring AI 提供两种 PDF 读取策略：

3.3.1 按页读取（PagePdfDocumentReader）
// 基础用法
PagePdfDocumentReader pdfReader = new PagePdfDocumentReader(
    new ClassPathResource("manual.pdf")
);
List<Document> documents = pdfReader.get();

// 每个 Document 对应 PDF 的一页
documents.forEach(doc -> {
    System.out.println("页码: " + doc.getMetadata().get("page_number"));
    System.out.println("内容: " + doc.getText().substring(0, 100) + "...");
});
3.3.2 按段落读取（ParagraphPdfDocumentReader）
// 更智能的分割策略，按段落而非页面分割
ParagraphPdfDocumentReader pdfReader = new ParagraphPdfDocumentReader(
    new ClassPathResource("technical-doc.pdf")
);
List<Document> documents = pdfReader.get();

// 每个 Document 对应一个逻辑段落
// 适合内容连续性较强的文档
3.3.3 PDF 配置选项
// 自定义 PDF 读取配置
PdfDocumentReaderConfig config = PdfDocumentReaderConfig.builder()
    .pageTopMargin(50)      // 页面上边距
    .pageExtractedTextFormatter(ExtractedTextFormatter.builder()
        .maxTextNormalize(true)     // 文本规范化
        .build())
    .build();

PagePdfDocumentReader reader = new PagePdfDocumentReader(
    new FileSystemResource("large-manual.pdf"), 
    config
);

3.4 TikaDocumentReader：万能文档读取器

基于 Apache Tika，支持几乎所有常见文档格式：

// Maven 依赖
<dependency>
    <groupId>org.springframework.ai</groupId>
    <artifactId>spring-ai-tika-document-reader</artifactId>
</dependency>

// Word 文档读取
TikaDocumentReader wordReader = new TikaDocumentReader(
    new FileSystemResource("proposal.docx")
);
List<Document> wordDocs = wordReader.get();

// PowerPoint 文档读取
TikaDocumentReader pptReader = new TikaDocumentReader(
    new FileSystemResource("presentation.pptx")
);
List<Document> pptDocs = pptReader.get();

// 自动检测文件类型
public List<Document> readAnyFormat(String filePath) {
    File file = new File(filePath);
    TikaDocumentReader reader = new TikaDocumentReader(
        new FileSystemResource(file)
    );
    
    List<Document> docs = reader.get();
    
    // Tika 自动添加的元数据
    docs.forEach(doc -> {
        Map<String, Object> metadata = doc.getMetadata();
        log.info("文件类型: {}", metadata.get("Content-Type"));
        log.info("作者: {}", metadata.get("Author"));
        log.info("创建时间: {}", metadata.get("Creation-Date"));
        log.info("修改时间: {}", metadata.get("Last-Modified"));
    });
    
    return docs;
}

3.5 JsonReader：结构化数据提取

用于从 JSON 文件中提取特定字段作为文档内容：

// JSON 文件示例：articles.json
{
  "articles": [
    {
      "id": "1",
      "title": "Spring AI 入门指南",
      "content": "Spring AI 是一个...",
      "author": "张三",
      "tags": ["Spring", "AI", "教程"]
    },
    {
      "id": "2", 
      "title": "RAG 系统实战",
      "content": "检索增强生成...",
      "author": "李四",
      "tags": ["RAG", "实战"]
    }
  ]
}

// 使用 JSONPath 提取内容
JsonReader jsonReader = new JsonReader(
    new ClassPathResource("articles.json"),
    "content",           // JSONPath 表达式，提取 content 字段
    "title", "author"    // 可选：额外的元数据字段
);

List<Document> documents = jsonReader.get();

documents.forEach(doc -> {
    System.out.println("标题: " + doc.getMetadata().get("title"));
    System.out.println("作者: " + doc.getMetadata().get("author"));
    System.out.println("内容: " + doc.getText());
});

// 复杂 JSONPath 示例
JsonReader complexReader = new JsonReader(
    new ClassPathResource("complex-data.json"),
    "$.articles[*].content"  // 提取所有文章的内容
);

4. Transform（转换）——DocumentTransformer 文档转换器

4.1 文本分块器（TextSplitter）

文本分块是 ETL 的核心环节，直接影响 RAG 系统的检索质量。

4.1.1 TokenTextSplitter（推荐）

基于 Token 计数的分块器，最适合向量化场景：

// 基础配置
TokenTextSplitter splitter = new TokenTextSplitter();
// 默认配置：chunkSize=800, minChunkSizeChars=350, minChunkLengthToEmbed=200

// 自定义配置
TokenTextSplitter customSplitter = new TokenTextSplitter(
    500,    // defaultChunkSize: 每块目标大小（Token 数）
    200,    // minChunkSizeChars: 最小块字符数
    100,    // minChunkLengthToEmbed: 最小 Embedding 长度
    2000,   // maxNumChunks: 单文档最大块数
    true    // keepSeparator: 保留分隔符
);

// 应用分块
List<Document> originalDocs = reader.get();
List<Document> chunks = customSplitter.apply(originalDocs);

log.info("原文档数: {}, 分块后: {}", originalDocs.size(), chunks.size());
4.1.2 分块策略优化

不同文档类型需要不同的分块策略：

@Service
public class DocumentSplitterService {
    
    // 技术文档：保持代码块完整性
    public TokenTextSplitter getTechnicalDocSplitter() {
        return new TokenTextSplitter(
            800,   // 块稍大，保持技术内容完整性
            300,
            150,
            1500,
            true
        );
    }
    
    // FAQ 问答：小块策略
    public TokenTextSplitter getFaqSplitter() {
        return new TokenTextSplitter(
            200,   // 小块，单个问答对为单位
            50,
            30,
            5000,
            true
        );
    }
    
    // 新闻文章：标准策略
    public TokenTextSplitter getNewsSplitter() {
        return new TokenTextSplitter(
            600,   // 标准大小
            250,
            100,
            2000,
            true
        );
    }
}
4.1.3 分块质量评估
public class ChunkQualityAnalyzer {
    
    public void analyzeChunkQuality(List<Document> chunks) {
        int totalChunks = chunks.size();
        int tooSmall = 0, optimal = 0, tooLarge = 0;
        
        for (Document chunk : chunks) {
            int length = chunk.getText().length();
            
            if (length < 100) tooSmall++;
            else if (length > 2000) tooLarge++;
            else optimal++;
        }
        
        log.info("分块质量分析:");
        log.info("总块数: {}", totalChunks);
        log.info("太小块数 (<100字符): {} ({:.1f}%)", tooSmall, (tooSmall * 100.0 / totalChunks));
        log.info("最优块数 (100-2000字符): {} ({:.1f}%)", optimal, (optimal * 100.0 / totalChunks));
        log.info("过大块数 (>2000字符): {} ({:.1f}%)", tooLarge, (tooLarge * 100.0 / totalChunks));
    }
}

4.2 内容格式化（ContentFormatTransformer）

为文档块添加格式化的上下文信息：

// 基础用法
ContentFormatTransformer formatter = new ContentFormatTransformer(
    "文档标题: {title}\n文件来源: {source}\n内容: {text}",
    true  // 包含元数据
);

List<Document> formattedDocs = formatter.apply(originalDocs);

// 自定义格式化模板
ContentFormatTransformer customFormatter = new ContentFormatTransformer(
    """
    ## 文档信息
    - 标题: {title}
    - 作者: {author}
    - 类别: {category}
    - 创建时间: {created_date}
    
    ## 内容
    {text}
    """,
    true
);

4.3 元数据增强器

使用 LLM 自动为文档块生成额外的元数据。

4.3.1 关键词提取器
// 需要注入 ChatModel
@Service
public class DocumentEnrichmentService {
    
    private final ChatModel chatModel;
    
    public List<Document> enrichWithKeywords(List<Document> documents) {
        KeywordMetadataEnricher enricher = new KeywordMetadataEnricher(
            chatModel, 
            5  // 提取5个关键词
        );
        
        return enricher.apply(documents);
    }
}
4.3.2 摘要生成器
public List<Document> enrichWithSummary(List<Document> documents) {
    SummaryMetadataEnricher enricher = new SummaryMetadataEnricher(
        chatModel,
        100  // 摘要长度限制
    );
    
    return enricher.apply(documents);
}

4.4 自定义 Transformer

实现自己的文档转换逻辑：

@Component
public class CustomDocumentTransformer implements DocumentTransformer {
    
    @Override
    public List<Document> apply(List<Document> documents) {
        return documents.stream()
            .map(this::transformDocument)
            .toList();
    }
    
    private Document transformDocument(Document doc) {
        String originalText = doc.getText();
        
        // 1. 清理文本（去除特殊字符、多余空行）
        String cleanedText = originalText
            .replaceAll("\\s+", " ")           // 多个空白字符合并为一个空格
            .replaceAll("[\\x00-\\x08\\x0B\\x0C\\x0E-\\x1F\\x7F]", "")  // 去除控制字符
            .trim();
        
        // 2. 提取章节标题
        String title = extractTitle(cleanedText);
        
        // 3. 计算文档特征
        Map<String, Object> newMetadata = new HashMap<>(doc.getMetadata());
        newMetadata.put("word_count", cleanedText.split("\\s+").length);
        newMetadata.put("char_count", cleanedText.length());
        newMetadata.put("extracted_title", title);
        newMetadata.put("language", detectLanguage(cleanedText));
        
        return new Document(doc.getId(), cleanedText, newMetadata);
    }
    
    private String extractTitle(String text) {
        // 简单的标题提取逻辑
        String[] lines = text.split("\n");
        for (String line : lines) {
            if (line.trim().length() > 5 && line.trim().length() < 100) {
                return line.trim();
            }
        }
        return "无标题";
    }
    
    private String detectLanguage(String text) {
        // 简单的语言检测
        return text.matches(".*[\\u4e00-\\u9fa5].*") ? "中文" : "英文";
    }
}

5. Load（加载）——DocumentWriter 文档写入器

5.1 VectorStore 作为 DocumentWriter

在 Spring AI 中，所有 VectorStore 实现都继承了 DocumentWriter 接口：

public interface DocumentWriter {
    void accept(List<Document> documents);
}

public interface VectorStore extends DocumentWriter {
    // VectorStore 特有方法
    void add(List<Document> documents);        // 等同于 accept()
    void delete(List<String> ids);
    List<Document> similaritySearch(SearchRequest request);
    // ...
}

5.2 批量写入策略

@Service
public class BatchDocumentLoader {
    
    private final VectorStore vectorStore;
    private final int batchSize;
    
    public BatchDocumentLoader(VectorStore vectorStore, 
                              @Value("${etl.batch.size:100}") int batchSize) {
        this.vectorStore = vectorStore;
        this.batchSize = batchSize;
    }
    
    /**
     * 批量加载文档，避免内存溢出
     */
    public void loadDocuments(List<Document> documents) {
        log.info("开始加载 {} 个文档，批次大小: {}", documents.size(), batchSize);
        
        for (int i = 0; i < documents.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, documents.size());
            List<Document> batch = documents.subList(i, endIndex);
            
            log.info("加载批次 {}/{}: {} 个文档", 
                (i / batchSize + 1), 
                (documents.size() + batchSize - 1) / batchSize, 
                batch.size());
            
            vectorStore.add(batch);
            
            // 避免频繁调用导致的限流
            if (endIndex < documents.size()) {
                try {
                    Thread.sleep(1000);  // 1秒间隔
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("加载被中断", e);
                }
            }
        }
        
        log.info("文档加载完成");
    }
}

5.3 写入性能优化

@Component
public class OptimizedDocumentLoader {
    
    private final VectorStore vectorStore;
    private final ExecutorService executorService;
    
    public OptimizedDocumentLoader(VectorStore vectorStore) {
        this.vectorStore = vectorStore;
        // 使用线程池并行处理
        this.executorService = Executors.newFixedThreadPool(4);
    }
    
    /**
     * 并行批量加载（仅适用于支持并发的 VectorStore）
     */
    public CompletableFuture<Void> loadDocumentsAsync(List<Document> documents) {
        int batchSize = 50;
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        
        for (int i = 0; i < documents.size(); i += batchSize) {
            int startIndex = i;
            int endIndex = Math.min(i + batchSize, documents.size());
            
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                List<Document> batch = documents.subList(startIndex, endIndex);
                vectorStore.add(batch);
                log.info("批次 {} 加载完成", startIndex / batchSize + 1);
            }, executorService);
            
            futures.add(future);
        }
        
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
    }
    
    @PreDestroy
    public void shutdown() {
        executorService.shutdown();
    }
}

6. 完整 ETL Pipeline 实战

6.1 项目依赖配置

<!-- pom.xml -->
<dependencies>
    <!-- Spring AI 核心 -->
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-starter-model-openai</artifactId>
    </dependency>
    
    <!-- PDF 支持 -->
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-pdf-document-reader</artifactId>
    </dependency>
    
    <!-- Tika 支持（Word, PPT等） -->
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-tika-document-reader</artifactId>
    </dependency>
    
    <!-- Vector Store -->
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-chroma-store</artifactId>
    </dependency>
</dependencies>

# application.yml
spring:
  ai:
    openai:
      api-key: ${OPENAI_API_KEY}
      embedding:
        model: text-embedding-3-small
    vectorstore:
      chroma:
        client:
          host: localhost
          port: 8000
        collection-name: knowledge_base

# ETL 配置
etl:
  batch-size: 100
  chunk-size: 600
  chunk-overlap: 100
  supported-formats:
    - pdf
    - txt
    - docx
    - md

6.2 基础 ETL Service

@Service
@Slf4j
public class DocumentETLService {
    
    private final VectorStore vectorStore;
    private final EmbeddingModel embeddingModel;
    
    @Value("${etl.chunk-size:600}")
    private int chunkSize;
    
    @Value("${etl.chunk-overlap:100}")
    private int chunkOverlap;
    
    public DocumentETLService(VectorStore vectorStore, EmbeddingModel embeddingModel) {
        this.vectorStore = vectorStore;
        this.embeddingModel = embeddingModel;
    }
    
    /**
     * 完整的 ETL 流程
     */
    public ETLResult processDocument(Resource resource) {
        try {
            long startTime = System.currentTimeMillis();
            
            // 1. Extract - 文档提取
            log.info("开始提取文档: {}", resource.getFilename());
            List<Document> rawDocuments = extractDocuments(resource);
            log.info("提取完成，原始文档数: {}", rawDocuments.size());
            
            // 2. Transform - 文档转换
            log.info("开始转换文档");
            List<Document> transformedDocuments = transformDocuments(rawDocuments);
            log.info("转换完成，文档块数: {}", transformedDocuments.size());
            
            // 3. Load - 文档加载
            log.info("开始加载到向量存储");
            loadDocuments(transformedDocuments);
            log.info("加载完成");
            
            long endTime = System.currentTimeMillis();
            
            return ETLResult.builder()
                .filename(resource.getFilename())
                .rawDocumentCount(rawDocuments.size())
                .processedDocumentCount(transformedDocuments.size())
                .processingTimeMs(endTime - startTime)
                .success(true)
                .build();
                
        } catch (Exception e) {
            log.error("ETL 处理失败: {}", resource.getFilename(), e);
            return ETLResult.builder()
                .filename(resource.getFilename())
                .success(false)
                .errorMessage(e.getMessage())
                .build();
        }
    }
    
    /**
     * Extract 阶段
     */
    private List<Document> extractDocuments(Resource resource) {
        String filename = resource.getFilename();
        if (filename == null) {
            throw new IllegalArgumentException("无法获取文件名");
        }
        
        String extension = getFileExtension(filename).toLowerCase();
        
        return switch (extension) {
            case "pdf" -> new PagePdfDocumentReader(resource).get();
            case "txt", "md" -> new TextReader(resource).get();
            case "docx", "doc", "pptx" -> new TikaDocumentReader(resource).get();
            case "json" -> new JsonReader(resource, "content").get();
            default -> throw new UnsupportedOperationException("不支持的文件格式: " + extension);
        };
    }
    
    /**
     * Transform 阶段
     */
    private List<Document> transformDocuments(List<Document> documents) {
        // 1. 文本分块
        TokenTextSplitter splitter = new TokenTextSplitter(
            chunkSize, 
            chunkOverlap, 
            50,    // 最小 embedding 长度
            2000,  // 最大块数
            true   // 保留分隔符
        );
        
        List<Document> chunks = splitter.apply(documents);
        
        // 2. 内容格式化（可选）
        ContentFormatTransformer formatter = new ContentFormatTransformer(
            "文档来源: {source}\n内容: {text}",
            true
        );
        
        return formatter.apply(chunks);
    }
    
    /**
     * Load 阶段
     */
    private void loadDocuments(List<Document> documents) {
        int batchSize = 50;
        
        for (int i = 0; i < documents.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, documents.size());
            List<Document> batch = documents.subList(i, endIndex);
            
            vectorStore.add(batch);
            
            log.info("已加载 {}/{} 个文档块", endIndex, documents.size());
        }
    }
    
    private String getFileExtension(String filename) {
        int lastDotIndex = filename.lastIndexOf('.');
        return lastDotIndex > 0 ? filename.substring(lastDotIndex + 1) : "";
    }
}

/**
 * ETL 结果对象
 */
@Data
@Builder
public class ETLResult {
    private String filename;
    private int rawDocumentCount;
    private int processedDocumentCount;
    private long processingTimeMs;
    private boolean success;
    private String errorMessage;
}

6.3 多格式文档批量处理

@Service
public class BatchETLService {
    
    private final DocumentETLService etlService;
    
    public BatchETLService(DocumentETLService etlService) {
        this.etlService = etlService;
    }
    
    /**
     * 批量处理目录下的所有支持格式文档
     */
    public BatchETLResult processDirectory(String directoryPath) {
        List<ETLResult> results = new ArrayList<>();
        
        try (Stream<Path> paths = Files.walk(Paths.get(directoryPath))) {
            List<Path> supportedFiles = paths
                .filter(Files::isRegularFile)
                .filter(this::isSupportedFormat)
                .toList();
            
            log.info("发现 {} 个支持的文件", supportedFiles.size());
            
            for (Path file : supportedFiles) {
                Resource resource = new FileSystemResource(file.toFile());
                ETLResult result = etlService.processDocument(resource);
                results.add(result);
            }
            
        } catch (IOException e) {
            log.error("目录遍历失败", e);
        }
        
        return aggregateResults(results);
    }
    
    /**
     * 并行处理（适合大批量文档）
     */
    public BatchETLResult processDirectoryParallel(String directoryPath) {
        List<ETLResult> results = Collections.synchronizedList(new ArrayList<>());
        
        try (Stream<Path> paths = Files.walk(Paths.get(directoryPath))) {
            List<Path> supportedFiles = paths
                .filter(Files::isRegularFile)
                .filter(this::isSupportedFormat)
                .toList();
            
            log.info("开始并行处理 {} 个文件", supportedFiles.size());
            
            supportedFiles.parallelStream().forEach(file -> {
                try {
                    Resource resource = new FileSystemResource(file.toFile());
                    ETLResult result = etlService.processDocument(resource);
                    results.add(result);
                } catch (Exception e) {
                    log.error("处理文件失败: {}", file, e);
                    results.add(ETLResult.builder()
                        .filename(file.getFileName().toString())
                        .success(false)
                        .errorMessage(e.getMessage())
                        .build());
                }
            });
            
        } catch (IOException e) {
            log.error("目录遍历失败", e);
        }
        
        return aggregateResults(results);
    }
    
    private boolean isSupportedFormat(Path file) {
        String filename = file.getFileName().toString().toLowerCase();
        return filename.endsWith(".pdf") || 
               filename.endsWith(".txt") ||
               filename.endsWith(".md") ||
               filename.endsWith(".docx") ||
               filename.endsWith(".json");
    }
    
    private BatchETLResult aggregateResults(List<ETLResult> results) {
        int successCount = (int) results.stream().mapToLong(r -> r.isSuccess() ? 1 : 0).sum();
        int failCount = results.size() - successCount;
        int totalDocuments = results.stream().mapToInt(ETLResult::getProcessedDocumentCount).sum();
        long totalTime = results.stream().mapToLong(ETLResult::getProcessingTimeMs).sum();
        
        return BatchETLResult.builder()
            .totalFiles(results.size())
            .successCount(successCount)
            .failCount(failCount)
            .totalProcessedDocuments(totalDocuments)
            .totalProcessingTimeMs(totalTime)
            .results(results)
            .build();
    }
}

@Data
@Builder
public class BatchETLResult {
    private int totalFiles;
    private int successCount;
    private int failCount;
    private int totalProcessedDocuments;
    private long totalProcessingTimeMs;
    private List<ETLResult> results;
}

6.4 ETL 任务调度与监控

@RestController
@RequestMapping("/api/etl")
public class ETLController {
    
    private final BatchETLService batchETLService;
    private final DocumentETLService etlService;
    
    public ETLController(BatchETLService batchETLService, 
                        DocumentETLService etlService) {
        this.batchETLService = batchETLService;
        this.etlService = etlService;
    }
    
    /**
     * 单文件处理
     */
    @PostMapping("/process-file")
    public ResponseEntity<ETLResult> processFile(@RequestParam("file") MultipartFile file) {
        try {
            Resource resource = file.getResource();
            ETLResult result = etlService.processDocument(resource);
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            return ResponseEntity.badRequest().build();
        }
    }
    
    /**
     * 目录批量处理
     */
    @PostMapping("/process-directory")
    public ResponseEntity<BatchETLResult> processDirectory(@RequestParam String path) {
        BatchETLResult result = batchETLService.processDirectory(path);
        return ResponseEntity.ok(result);
    }
    
    /**
     * ETL 状态监控
     */
    @GetMapping("/status")
    public ResponseEntity<ETLStatus> getStatus() {
        // 这里可以实现 ETL 任务状态监控
        // 例如：正在处理的文件数、队列长度、成功/失败统计等
        return ResponseEntity.ok(ETLStatus.builder()
            .isProcessing(false)
            .queueSize(0)
            .build());
    }
}

@Data
@Builder
public class ETLStatus {
    private boolean isProcessing;
    private int queueSize;
    private int totalProcessed;
    private int totalFailed;
}

7. Document 对象深度解析

7.1 Document 数据结构

public class Document {
    private String id;                           // 文档唯一标识
    private String text;                         // 文档文本内容
    private Map<String, Object> metadata;       // 元数据
    private float[] embedding;                   // 向量（通常由 VectorStore 管理）
    
    // 构造方法
    public Document(String text) {
        this(UUID.randomUUID().toString(), text, new HashMap<>());
    }
    
    public Document(String id, String text, Map<String, Object> metadata) {
        this.id = id;
        this.text = text;
        this.metadata = metadata != null ? metadata : new HashMap<>();
    }
}

7.2 元数据（Metadata）最佳实践

元数据类别字段名类型用途
文件信息sourceString文件路径或来源
filenameString文件名
file_sizeLong文件大小（字节）
file_typeString文件格式
内容信息titleString文档标题
authorString作者
created_dateString创建日期
page_numberInteger页码（PDF）
word_countInteger字数
处理信息chunk_indexInteger分块序号
processing_dateString处理时间
versionString文档版本
业务信息categoryString分类
departmentString部门
tagsList标签
languageString语言
// 元数据使用示例
public Document createDocumentWithMetadata(String content, Path filePath) {
    Map<String, Object> metadata = new HashMap<>();
    
    // 文件信息
    metadata.put("source", filePath.toString());
    metadata.put("filename", filePath.getFileName().toString());
    metadata.put("file_size", filePath.toFile().length());
    metadata.put("file_type", getFileExtension(filePath.getFileName().toString()));
    
    // 内容信息
    metadata.put("word_count", content.split("\\s+").length);
    metadata.put("char_count", content.length());
    metadata.put("language", detectLanguage(content));
    
    // 处理信息
    metadata.put("processing_date", LocalDateTime.now().toString());
    metadata.put("processor_version", "v1.0.0");
    
    // 业务信息（可选）
    metadata.put("category", inferCategory(filePath));
    
    return new Document(content, metadata);
}

7.3 文档 ID 策略

不同的 ID 策略适合不同场景：

public class DocumentIdGenerator {
    
    /**
     * 基于内容哈希的 ID（去重）
     */
    public String generateContentBasedId(String content) {
        return DigestUtils.md5DigestAsHex(content.getBytes(StandardCharsets.UTF_8));
    }
    
    /**
     * 基于文件路径的 ID（可追溯）
     */
    public String generatePathBasedId(Path filePath, int chunkIndex) {
        String pathHash = DigestUtils.md5DigestAsHex(filePath.toString().getBytes());
        return String.format("%s_chunk_%d", pathHash, chunkIndex);
    }
    
    /**
     * 时间序列 ID（顺序性）
     */
    public String generateTimeSeriesId() {
        return LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss_")) 
               + UUID.randomUUID().toString().substring(0, 8);
    }
    
    /**
     * 业务逻辑 ID（语义性）
     */
    public String generateBusinessId(String category, String department, String title) {
        return String.format("%s_%s_%s_%s", 
            category, 
            department, 
            sanitize(title), 
            System.currentTimeMillis());
    }
    
    private String sanitize(String input) {
        return input.replaceAll("[^a-zA-Z0-9\\u4e00-\\u9fa5]", "_").toLowerCase();
    }
}

8. ETL 性能优化与最佳实践

8.1 分块策略选择

文档类型推荐块大小重叠策略理由
技术文档600-1000 tokens15-20%保持代码块和概念完整性
产品手册400-600 tokens10-15%平衡检索精度和内容完整性
新闻文章300-500 tokens10%段落相对独立
法律条文200-400 tokens20-25%条款间关联性强
FAQ问答100-200 tokens5%单个问答对为单位
对话记录150-300 tokens10%保持对话上下文

8.2 内存管理与流式处理

@Service
public class StreamingETLService {
    
    private final VectorStore vectorStore;
    private final int maxMemoryDocuments = 1000;  // 内存中最大文档数
    
    /**
     * 流式处理大文件，避免内存溢出
     */
    public void processLargeDirectory(String directoryPath) throws IOException {
        try (Stream<Path> fileStream = Files.walk(Paths.get(directoryPath))) {
            
            // 使用流式处理 + 缓冲区
            List<Document> buffer = new ArrayList<>();
            
            fileStream.filter(Files::isRegularFile)
                     .filter(this::isSupportedFile)
                     .forEach(file -> {
                         try {
                             List<Document> docs = processFile(file);
                             buffer.addAll(docs);
                             
                             // 当缓冲区达到限制时，写入向量存储并清空
                             if (buffer.size() >= maxMemoryDocuments) {
                                 vectorStore.add(new ArrayList<>(buffer));
                                 buffer.clear();
                                 
                                 // 强制垃圾回收（在大批量处理时有用）
                                 System.gc();
                             }
                             
                         } catch (Exception e) {
                             log.error("处理文件失败: {}", file, e);
                         }
                     });
            
            // 处理剩余文档
            if (!buffer.isEmpty()) {
                vectorStore.add(buffer);
            }
        }
    }
    
    /**
     * 使用 WeakReference 避免内存泄漏
     */
    private final Map<String, WeakReference<List<Document>>> documentCache = 
        new ConcurrentHashMap<>();
    
    public List<Document> getCachedDocuments(String key) {
        WeakReference<List<Document>> ref = documentCache.get(key);
        if (ref != null) {
            List<Document> docs = ref.get();
            if (docs != null) {
                return docs;
            } else {
                documentCache.remove(key);  // 清除过期引用
            }
        }
        return null;
    }
}

8.3 并行处理

@Configuration
@EnableAsync
public class ETLAsyncConfig {
    
    @Bean(name = "etlExecutor")
    public Executor etlExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);                    // 核心线程数
        executor.setMaxPoolSize(8);                     // 最大线程数
        executor.setQueueCapacity(100);                 // 队列容量
        executor.setThreadNamePrefix("ETL-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

@Service
public class ParallelETLService {
    
    private final DocumentETLService etlService;
    
    public ParallelETLService(DocumentETLService etlService) {
        this.etlService = etlService;
    }
    
    /**
     * 并行处理多个文件
     */
    @Async("etlExecutor")
    public CompletableFuture<List<ETLResult>> processFilesAsync(List<Path> files) {
        
        // 将文件列表分组，每组在一个线程中处理
        List<List<Path>> groups = partitionList(files, 4);
        
        List<CompletableFuture<List<ETLResult>>> futures = groups.stream()
            .map(this::processFileGroup)
            .toList();
        
        // 等待所有组完成
        CompletableFuture<Void> allGroups = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );
        
        return allGroups.thenApply(v -> 
            futures.stream()
                   .flatMap(f -> f.join().stream())
                   .toList()
        );
    }
    
    @Async("etlExecutor")
    public CompletableFuture<List<ETLResult>> processFileGroup(List<Path> files) {
        return CompletableFuture.supplyAsync(() -> 
            files.stream()
                 .map(file -> etlService.processDocument(new FileSystemResource(file.toFile())))
                 .toList()
        );
    }
    
    private <T> List<List<T>> partitionList(List<T> list, int partitionSize) {
        List<List<T>> partitions = new ArrayList<>();
        for (int i = 0; i < list.size(); i += partitionSize) {
            partitions.add(list.subList(i, Math.min(i + partitionSize, list.size())));
        }
        return partitions;
    }
}

8.4 增量更新

@Service
public class IncrementalETLService {
    
    private final VectorStore vectorStore;
    private final RedisTemplate<String, String> redisTemplate;
    
    /**
     * 检查文件是否需要重新处理
     */
    public boolean needsUpdate(Path file) {
        try {
            String fileKey = "etl:" + file.toString();
            String lastModified = redisTemplate.opsForValue().get(fileKey);
            
            if (lastModified == null) {
                return true;  // 首次处理
            }
            
            long lastProcessTime = Long.parseLong(lastModified);
            long currentModTime = Files.getLastModifiedTime(file).toMillis();
            
            return currentModTime > lastProcessTime;
            
        } catch (Exception e) {
            log.warn("检查文件更新状态失败: {}", file, e);
            return true;  // 出错时重新处理
        }
    }
    
    /**
     * 增量更新处理
     */
    public void processIncrementalUpdates(String directoryPath) throws IOException {
        try (Stream<Path> fileStream = Files.walk(Paths.get(directoryPath))) {
            
            List<Path> modifiedFiles = fileStream
                .filter(Files::isRegularFile)
                .filter(this::isSupportedFile)
                .filter(this::needsUpdate)
                .toList();
            
            log.info("发现 {} 个需要更新的文件", modifiedFiles.size());
            
            for (Path file : modifiedFiles) {
                // 1. 删除旧文档（根据文件路径）
                deleteDocumentsBySource(file.toString());
                
                // 2. 处理新文档
                ETLResult result = etlService.processDocument(new FileSystemResource(file.toFile()));
                
                if (result.isSuccess()) {
                    // 3. 更新处理时间戳
                    String fileKey = "etl:" + file.toString();
                    redisTemplate.opsForValue().set(fileKey, 
                        String.valueOf(Files.getLastModifiedTime(file).toMillis()));
                    
                    log.info("文件增量更新完成: {}", file);
                } else {
                    log.error("文件增量更新失败: {}", file);
                }
            }
        }
    }
    
    private void deleteDocumentsBySource(String source) {
        // 这里需要根据具体 VectorStore 实现
        // 大部分 VectorStore 支持根据元数据过滤删除
        try {
            SearchRequest searchRequest = SearchRequest.builder()
                .query("")  // 空查询
                .filterExpression("source == '" + source + "'")
                .topK(1000)  // 足够大的数字
                .build();
            
            List<Document> existingDocs = vectorStore.similaritySearch(searchRequest);
            List<String> idsToDelete = existingDocs.stream()
                .map(Document::getId)
                .toList();
            
            if (!idsToDelete.isEmpty()) {
                vectorStore.delete(idsToDelete);
                log.info("删除了 {} 个旧文档，来源: {}", idsToDelete.size(), source);
            }
            
        } catch (Exception e) {
            log.error("删除旧文档失败，来源: {}", source, e);
        }
    }
}

9. 常见场景与解决方案

9.1 企业文档库迁移

企业常见需求：将现有的文档管理系统迁移到 RAG 系统。

@Service
public class EnterpriseDocumentMigrationService {
    
    private final DocumentETLService etlService;
    private final BatchETLService batchETLService;
    
    /**
     * 企业文档库迁移方案
     */
    public MigrationResult migrateDocumentLibrary(String sourceDirectory, 
                                                 MigrationConfig config) {
        log.info("开始企业文档库迁移，源目录: {}", sourceDirectory);
        
        MigrationResult result = new MigrationResult();
        result.setStartTime(LocalDateTime.now());
        
        try {
            // 1. 扫描并分类文档
            DocumentScanResult scanResult = scanDocuments(sourceDirectory, config);
            result.setScanResult(scanResult);
            
            // 2. 按优先级处理文档
            for (DocumentCategory category : config.getProcessingOrder()) {
                List<Path> categoryFiles = scanResult.getFilesByCategory().get(category);
                if (categoryFiles != null && !categoryFiles.isEmpty()) {
                    processCategoryFiles(categoryFiles, category, result);
                }
            }
            
            // 3. 生成迁移报告
            generateMigrationReport(result);
            
        } catch (Exception e) {
            log.error("文档库迁移失败", e);
            result.setSuccess(false);
            result.setErrorMessage(e.getMessage());
        }
        
        result.setEndTime(LocalDateTime.now());
        return result;
    }
    
    private DocumentScanResult scanDocuments(String sourceDirectory, 
                                           MigrationConfig config) throws IOException {
        
        DocumentScanResult scanResult = new DocumentScanResult();
        Map<DocumentCategory, List<Path>> filesByCategory = new HashMap<>();
        
        try (Stream<Path> fileStream = Files.walk(Paths.get(sourceDirectory))) {
            fileStream.filter(Files::isRegularFile)
                     .forEach(file -> {
                         DocumentCategory category = classifyDocument(file, config);
                         filesByCategory.computeIfAbsent(category, k -> new ArrayList<>())
                                       .add(file);
                     });
        }
        
        scanResult.setFilesByCategory(filesByCategory);
        
        // 生成扫描统计
        int totalFiles = filesByCategory.values().stream()
            .mapToInt(List::size)
            .sum();
        scanResult.setTotalFiles(totalFiles);
        
        log.info("文档扫描完成，总计 {} 个文件，按类别分布: {}", 
            totalFiles, 
            filesByCategory.entrySet().stream()
                .collect(Collectors.toMap(
                    Map.Entry::getKey, 
                    e -> e.getValue().size()))
        );
        
        return scanResult;
    }
    
    private DocumentCategory classifyDocument(Path file, MigrationConfig config) {
        String filename = file.getFileName().toString().toLowerCase();
        String parent = file.getParent().getFileName().toString().toLowerCase();
        
        // 根据文件路径和名称推断类别
        if (parent.contains("manual") || parent.contains("guide")) {
            return DocumentCategory.MANUAL;
        } else if (parent.contains("policy") || parent.contains("regulation")) {
            return DocumentCategory.POLICY;
        } else if (parent.contains("faq") || parent.contains("qa")) {
            return DocumentCategory.FAQ;
        } else if (filename.contains("spec") || filename.contains("specification")) {
            return DocumentCategory.SPECIFICATION;
        }
        
        return DocumentCategory.GENERAL;
    }
    
    private void processCategoryFiles(List<Path> files, 
                                    DocumentCategory category,
                                    MigrationResult result) {
        log.info("开始处理 {} 类文档，共 {} 个文件", category, files.size());
        
        for (Path file : files) {
            try {
                ETLResult etlResult = etlService.processDocument(
                    new FileSystemResource(file.toFile())
                );
                
                result.addETLResult(category, etlResult);
                
                if (etlResult.isSuccess()) {
                    log.info("成功处理文档: {}", file.getFileName());
                } else {
                    log.error("处理文档失败: {}, 错误: {}", 
                        file.getFileName(), etlResult.getErrorMessage());
                }
                
            } catch (Exception e) {
                log.error("处理文档异常: {}", file.getFileName(), e);
                result.addETLResult(category, ETLResult.builder()
                    .filename(file.getFileName().toString())
                    .success(false)
                    .errorMessage(e.getMessage())
                    .build());
            }
        }
    }
}

// 支持类
public enum DocumentCategory {
    MANUAL, POLICY, FAQ, SPECIFICATION, GENERAL;
}

@Data
public class MigrationConfig {
    private List<DocumentCategory> processingOrder = Arrays.asList(
        DocumentCategory.MANUAL,
        DocumentCategory.SPECIFICATION, 
        DocumentCategory.POLICY,
        DocumentCategory.FAQ,
        DocumentCategory.GENERAL
    );
    
    private Set<String> excludeDirectories = Set.of("temp", "backup", ".git");
    private long maxFileSizeBytes = 50 * 1024 * 1024; // 50MB
}

@Data
public class MigrationResult {
    private LocalDateTime startTime;
    private LocalDateTime endTime;
    private boolean success = true;
    private String errorMessage;
    private DocumentScanResult scanResult;
    private Map<DocumentCategory, List<ETLResult>> etlResults = new HashMap<>();
    
    public void addETLResult(DocumentCategory category, ETLResult result) {
        etlResults.computeIfAbsent(category, k -> new ArrayList<>()).add(result);
    }
}

9.2 网页爬取与处理

@Service
public class WebCrawlETLService {
    
    private final DocumentETLService etlService;
    private final WebClient webClient;
    
    public WebCrawlETLService(DocumentETLService etlService) {
        this.etlService = etlService;
        this.webClient = WebClient.builder()
            .defaultHeader(HttpHeaders.USER_AGENT, 
                "Mozilla/5.0 (compatible; SpringAI-ETL/1.0)")
            .build();
    }
    
    /**
     * 网页内容抓取并处理
     */
    public ETLResult processWebPage(String url) {
        try {
            // 1. 抓取网页内容
            String htmlContent = webClient.get()
                .uri(url)
                .retrieve()
                .bodyToMono(String.class)
                .block();
            
            // 2. 提取文本内容（去除 HTML 标签）
            String textContent = extractTextFromHtml(htmlContent);
            
            // 3. 创建文档对象
            Document document = new Document(textContent);
            document.getMetadata().put("source", url);
            document.getMetadata().put("source_type", "web");
            document.getMetadata().put("crawl_time", LocalDateTime.now().toString());
            
            // 4. 使用标准 ETL 流程处理
            return etlService.processDocument(new InMemoryResource(textContent, url));
            
        } catch (Exception e) {
            log.error("网页处理失败: {}", url, e);
            return ETLResult.builder()
                .filename(url)
                .success(false)
                .errorMessage(e.getMessage())
                .build();
        }
    }
    
    private String extractTextFromHtml(String html) {
        // 使用 Jsoup 提取纯文本
        return Jsoup.parse(html).text();
    }
    
    /**
     * 内存资源实现，用于处理非文件数据
     */
    private static class InMemoryResource implements Resource {
        private final String content;
        private final String description;
        
        public InMemoryResource(String content, String description) {
            this.content = content;
            this.description = description;
        }
        
        @Override
        public InputStream getInputStream() throws IOException {
            return new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8));
        }
        
        @Override
        public boolean exists() { return true; }
        
        @Override
        public String getDescription() { return description; }
        
        @Override
        public String getFilename() { 
            return description.replaceAll("[^a-zA-Z0-9]", "_") + ".txt"; 
        }
        
        // 其他必要的方法实现...
    }
}

9.3 多语言文档处理

@Service
public class MultiLanguageETLService {
    
    private final DocumentETLService etlService;
    
    /**
     * 多语言文档处理
     */
    public ETLResult processMultiLanguageDocument(Resource resource) {
        try {
            // 1. 检测文档语言
            String content = readResourceContent(resource);
            String detectedLanguage = detectLanguage(content);
            
            // 2. 根据语言选择合适的处理策略
            DocumentProcessingStrategy strategy = getProcessingStrategy(detectedLanguage);
            
            // 3. 应用语言特定的预处理
            String preprocessedContent = strategy.preprocess(content);
            
            // 4. 选择合适的分块器
            TokenTextSplitter splitter = strategy.getTextSplitter();
            
            // 5. 处理文档
            Document document = new Document(preprocessedContent);
            document.getMetadata().put("detected_language", detectedLanguage);
            document.getMetadata().put("processing_strategy", strategy.getName());
            
            return processDocumentWithStrategy(document, strategy);
            
        } catch (Exception e) {
            log.error("多语言文档处理失败", e);
            return ETLResult.builder()
                .filename(resource.getFilename())
                .success(false)
                .errorMessage(e.getMessage())
                .build();
        }
    }
    
    private String detectLanguage(String content) {
        // 简单的语言检测逻辑
        if (content.matches(".*[\\u4e00-\\u9fa5].*")) {
            return "zh";  // 中文
        } else if (content.matches(".*[ひらがなカタカナ].*")) {
            return "ja";  // 日文
        } else if (content.matches(".*[가-힣].*")) {
            return "ko";  // 韩文
        } else {
            return "en";  // 默认英文
        }
    }
    
    private DocumentProcessingStrategy getProcessingStrategy(String language) {
        return switch (language) {
            case "zh" -> new ChineseProcessingStrategy();
            case "ja" -> new JapaneseProcessingStrategy();
            case "ko" -> new KoreanProcessingStrategy();
            default -> new EnglishProcessingStrategy();
        };
    }
}

// 处理策略接口
interface DocumentProcessingStrategy {
    String getName();
    String preprocess(String content);
    TokenTextSplitter getTextSplitter();
}

// 中文处理策略
class ChineseProcessingStrategy implements DocumentProcessingStrategy {
    
    @Override
    public String getName() { return "Chinese"; }
    
    @Override
    public String preprocess(String content) {
        // 中文特定的预处理
        return content
            .replaceAll("\\s+", "")           // 去除多余空格
            .replaceAll("[，。！？；：]", "$0 "); // 在标点后添加空格，便于分句
    }
    
    @Override
    public TokenTextSplitter getTextSplitter() {
        // 中文文档通常密度更高，块可以稍小一些
        return new TokenTextSplitter(400, 100, 50, 2000, true);
    }
}

// 英文处理策略
class EnglishProcessingStrategy implements DocumentProcessingStrategy {
    
    @Override
    public String getName() { return "English"; }
    
    @Override
    public String preprocess(String content) {
        // 英文预处理
        return content
            .replaceAll("\\s+", " ")          // 规范化空格
            .replaceAll("([.!?])\\s*", "$1 "); // 确保句号后有空格
    }
    
    @Override
    public TokenTextSplitter getTextSplitter() {
        // 英文标准分块策略
        return new TokenTextSplitter(600, 150, 80, 2000, true);
    }
}

10. 故障排查与调试

常见问题与解决方案

问题类型症状可能原因解决方案
文档读取失败DocumentReader.get() 抛异常文件格式不支持/损坏检查文件完整性，使用 TikaDocumentReader
分块结果异常块太大/太小/数量异常分块参数不当调整 TokenTextSplitter 参数
向量化失败VectorStore.add() 超时EmbeddingModel 配置错误检查 API 密钥和网络连接
内存溢出OutOfMemoryError文档批次太大减小批次大小，使用流式处理
检索结果差相关文档检索不到分块策略不当优化分块大小和重叠策略

调试工具

@Component
public class ETLDebugHelper {
    
    /**
     * 分析文档分块质量
     */
    public void analyzeChunking(List<Document> documents) {
        log.info("=== 文档分块质量分析 ===");
        
        IntSummaryStatistics stats = documents.stream()
            .mapToInt(doc -> doc.getText().length())
            .summaryStatistics();
        
        log.info("总块数: {}", documents.size());
        log.info("平均长度: {:.0f} 字符", stats.getAverage());
        log.info("最短块: {} 字符", stats.getMin());
        log.info("最长块: {} 字符", stats.getMax());
        
        // 长度分布
        Map<String, Long> lengthDistribution = documents.stream()
            .collect(Collectors.groupingBy(
                doc -> {
                    int length = doc.getText().length();
                    if (length < 100) return "<100";
                    else if (length < 500) return "100-500";
                    else if (length < 1000) return "500-1000";
                    else if (length < 2000) return "1000-2000";
                    else return ">2000";
                },
                Collectors.counting()
            ));
        
        log.info("长度分布: {}", lengthDistribution);
    }
    
    /**
     * 检查元数据完整性
     */
    public void analyzeMetadata(List<Document> documents) {
        log.info("=== 元数据完整性分析 ===");
        
        Set<String> allKeys = documents.stream()
            .flatMap(doc -> doc.getMetadata().keySet().stream())
            .collect(Collectors.toSet());
        
        log.info("发现的元数据字段: {}", allKeys);
        
        for (String key : allKeys) {
            long count = documents.stream()
                .mapToLong(doc -> doc.getMetadata().containsKey(key) ? 1 : 0)
                .sum();
            
            double coverage = (count * 100.0) / documents.size();
            log.info("字段 '{}' 覆盖率: {:.1f}% ({}/{})", 
                key, coverage, count, documents.size());
        }
    }
    
    /**
     * 模拟向量检索测试
     */
    public void testVectorRetrieval(VectorStore vectorStore, List<String> testQueries) {
        log.info("=== 向量检索测试 ===");
        
        for (String query : testQueries) {
            log.info("测试查询: {}", query);
            
            List<Document> results = vectorStore.similaritySearch(
                SearchRequest.builder()
                    .query(query)
                    .topK(3)
                    .build()
            );
            
            for (int i = 0; i < results.size(); i++) {
                Document doc = results.get(i);
                log.info("  结果 {}: 相似度={}, 内容={}...", 
                    i+1,
                    doc.getMetadata().get("score"),
                    doc.getText().substring(0, Math.min(100, doc.getText().length())));
            }
        }
    }
}

11. 总结

本文深入讲解了 Spring AI ETL Pipeline 的完整实现：

模块核心组件关键点
ExtractDocumentReader支持多种文档格式，统一抽象
TransformDocumentTransformer分块策略是质量关键，支持元数据增强
LoadVectorStore/DocumentWriter批量写入，性能优化

核心认知：

  1. ETL 是 RAG 的数据基础，文档质量直接影响检索效果
  2. 分块策略是成功关键，需要根据文档类型和业务场景调优
  3. 性能优化不可忽视，大规模文档库需要流式处理和并行策略
  4. 元数据设计要前瞻，为后续的过滤和分类奠定基础

最佳实践：

  • 使用 TokenTextSplitter 进行分块，根据文档类型调整参数
  • 合理设计元数据结构，便于后续检索和管理
  • 大批量处理时使用流式处理和批量写入
  • 建立增量更新机制，避免重复处理
  • 完善监控和调试工具，持续优化 ETL 质量

12. 参考资料

Spring AI 官方文档

相关依赖项目

前置知识

其他资源

# spring # 人工智能 # etl
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

开源鸿蒙跨平台直播｜napigen 开源鸿蒙跨端实践：打通 Flutter 与 ArkTS

avatar AtomGit开源社区
cover

玩转 AtomCode！AtomGit「码动四季・开源同行」夏季征稿获奖名单出炉！

avatar AtomGit开源社区
cover

榜单出炉！AtomGit「源启盛夏」暑期开源成长计划正式收官

avatar AtomGit开源社区