ETL 流水线
约 4350 字大约 15 分钟
2026-09-15
📌 原文地址:https://docs.spring.io/java-ai/reference/api/etl-pipeline.html
📌 原文版本:Java AI 2.0.1(Stable)
Extract(抽取)、Transform(转换)、Load(加载)框架(即 ETL 框架)是 Retrieval Augmented Generation(RAG,检索增强生成)用例中数据处理的核心骨干。
ETL 流水线编排了从原始数据源到结构化向量存储(Vector Store)的流动过程,确保数据处于最适合 AI 模型检索的格式。
RAG 用例通过从数据集合中检索相关信息来增强生成式模型的能力,从而提升生成输出的质量和相关性。
白话解读
ETL = 数据搬运工。Extract 把数据从 PDF/Word/HTML 等源头读出来,Transform 把它们切成模型能吃的"小块"(按 token 数),Load 把这些块塞进向量数据库。Java AI 把这三步抽象成三个接口(Reader/Transformer/Writer),每一步都有现成实现,可以像乐高一样拼装。
一、API 概览(API Overview)
ETL 流水线负责创建、转换和存储 Document 实例。
Document 类包含文本(text)、元数据(metadata),并且可选地包含其他媒体类型,如图像、音频和视频。
ETL 流水线有三个主要组件:
- DocumentReader:实现
Supplier<List<Document>> - DocumentTransformer:实现
Function<List<Document>, List<Document>> - DocumentWriter:实现
Consumer<List<Document>>
Document 类的内容由 DocumentReader 从 PDF、文本文件以及其他文档类型中创建。
要构建一个简单的 ETL 流水线,可以将每种类型的一个实例串联起来。
假设我们有以下三个 ETL 类型的实例:
PagePdfDocumentReader:DocumentReader的实现TokenTextSplitter:DocumentTransformer的实现VectorStore:DocumentWriter的实现
要执行将数据基本加载到向量数据库中以便与 RAG 模式配合使用,可以使用以下 Java 函数式风格的代码:
// 函数式写法:先读取 PDF,再切分,最后写入向量存储
vectorStore.accept(tokenTextSplitter.apply(pdfReader.get()));或者,也可以使用更贴合领域语义、更具表达力的方法名:
// 领域语义写法:read -> split -> write
vectorStore.write(tokenTextSplitter.split(pdfReader.read()));二、ETL 接口(ETL Interfaces)
ETL 流水线由以下接口和实现组成。详细的 ETL 类图见「ETL Class Diagram」章节。
DocumentReader
提供来自多种来源的文档源。
// 文档读取器接口,继承 Supplier<List<Document>>
public interface DocumentReader extends Supplier<List<Document>> {
// 默认方法:调用 get() 获得文档列表
default List<Document> read() {
return get();
}
}DocumentTransformer
作为处理工作流的一部分,对一批文档进行转换。
// 文档转换器接口,继承 Function<List<Document>, List<Document>>
public interface DocumentTransformer extends Function<List<Document>, List<Document>> {
// 默认方法:调用 apply() 执行转换
default List<Document> transform(List<Document> transform) {
return apply(transform);
}
}DocumentWriter
管理 ETL 流程的最后阶段,准备将文档进行存储。
// 文档写入器接口,继承 Consumer<List<Document>>
public interface DocumentWriter extends Consumer<List<Document>> {
// 默认方法:调用 accept() 执行写入
default void write(List<Document> documents) {
accept(documents);
}
}三、文档读取器(Document Readers)
⚠️ 安全性提示:下面描述的大多数
DocumentReader和DocumentWriter实现都是通过Resource或资源模式(resource pattern)构造的,内部使用DefaultResourceLoader或其变体来访问存储。在构造这类实例时应避免直接使用用户提供的 URL,因为这存在安全隐患。
3.1 JSON
JsonReader 处理 JSON 文档,将其转换为 Document 对象列表。
示例:
@Component
class MyJsonReader {
private final Resource resource;
MyJsonReader(@Value("classpath:bikes.json") Resource resource) {
this.resource = resource;
}
List<Document> loadJsonAsDocuments() {
// 使用 "description" 和 "content" 作为内容键
JsonReader jsonReader = new JsonReader(this.resource, "description", "content");
return jsonReader.get();
}
}构造函数选项:
JsonReader(Resource resource)JsonReader(Resource resource, String… jsonKeysToUse)JsonReader(Resource resource, JsonMetadataGenerator jsonMetadataGenerator, String… jsonKeysToUse)
参数:
- resource:指向 JSON 文件的 Spring
Resource对象。 - jsonKeysToUse:JSON 中应当作为最终
Document对象文本内容的键名数组。 - jsonMetadataGenerator:可选的
JsonMetadataGenerator,用于为每个Document创建元数据。
行为:
- 可以处理 JSON 数组和单个 JSON 对象。
- 对每个 JSON 对象(数组中的对象或单个对象):
- 根据指定的
jsonKeysToUse抽取内容。 - 如果未指定键,则使用整个 JSON 对象作为内容。
- 使用提供的
JsonMetadataGenerator(如果未提供则使用空实现)生成元数据。 - 使用抽取出的内容和元数据创建
Document对象。
- 根据指定的
使用 JSON Pointers
JsonReader 现已支持使用 JSON Pointer 检索 JSON 文档的特定部分——此功能允许从复杂 JSON 结构中轻松提取嵌套数据。
// 通过 JSON Pointer 获取文档列表
public List<Document> get(String pointer)参数: pointer —— 一个 JSON Pointer 字符串(按 RFC 6901 定义),用于定位 JSON 结构中的目标元素。
返回值: 返回 List<Document>,包含从 pointer 定位到的 JSON 元素解析出的文档。
行为:
- 若指向 JSON 对象:返回含单个
Document的列表。 - 若指向 JSON 数组:返回
Document列表,数组中的每个元素对应一个文档。 - 如果 pointer 无效或指向不存在的元素,则抛出
IllegalArgumentException。
示例:
// 使用 JSON Pointer 定位到 /store/books/0
JsonReader jsonReader = new JsonReader(resource, "description");
List<Document> documents = this.jsonReader.get("/store/books/0");说明:
JsonReader使用 Jackson 进行 JSON 解析。- 它通过流式处理数组来高效处理大型 JSON 文件。
- 如果
jsonKeysToUse指定了多个键,内容将是这些键对应值的拼接。
3.2 TEXT
TextReader 处理纯文本文档,将其转换为 Document 对象列表。
示例:
@Component
class MyTextReader {
private final Resource resource;
MyTextReader(@Value("classpath:text-source.txt") Resource resource) {
this.resource = resource;
}
List<Document> loadText() {
TextReader textReader = new TextReader(this.resource);
// 添加自定义元数据
textReader.getCustomMetadata().put("filename", "text-source.txt");
return textReader.read();
}
}构造函数选项:
TextReader(String resourceUrl)TextReader(Resource resource)
参数:
- resourceUrl:表示要读取的资源 URL 的字符串。
- resource:指向文本文件的 Spring
Resource对象。
配置:
setCharset(Charset charset):设置读取文本文件所用的字符集。默认 UTF-8。getCustomMetadata():返回一个可变的 map,用于为文档添加自定义元数据。
行为:
- 将文本文件的全部内容读入单个
Document对象。 - 自动添加以下元数据:
- charset:读取文件所用的字符集(默认
"UTF-8")。 - source:源文本文件的文件名。
- charset:读取文件所用的字符集(默认
- 通过
getCustomMetadata()添加的任何自定义元数据都会包含在Document中。
说明:
TextReader将整个文件内容读入内存,因此可能不适用于非常大的文件。- 如果需要将文本拆分成更小的块,可以在读取文档后使用
TokenTextSplitter:
List<Document> documents = textReader.get();
List<Document> splitDocuments = TokenTextSplitter.builder().build().apply(this.documents);3.3 HTML (Jsoup)
JsoupDocumentReader 使用 JSoup 库处理 HTML 文档。
依赖:
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-jsoup-document-reader</artifactId>
</dependency>示例:
@Component
class MyHtmlReader {
private final Resource resource;
MyHtmlReader(@Value("classpath:/my-page.html") Resource resource) {
this.resource = resource;
}
List<Document> loadHtml() {
JsoupDocumentReaderConfig config = JsoupDocumentReaderConfig.builder()
.selector("article p") // 提取 <article> 标签内的段落
.charset("ISO-8859-1") // 使用 ISO-8859-1 编码
.includeLinkUrls(true) // 在元数据中包含链接 URL
.metadataTags(List.of("author", "date")) // 提取 author 和 date 元标签
.additionalMetadata("source", "my-page.html") // 添加自定义元数据
.build();
JsoupDocumentReader reader = new JsoupDocumentReader(this.resource, config);
return reader.get();
}
}JsoupDocumentReaderConfig 字段:
- charset:HTML 文档的字符编码(默认
"UTF-8")。 - selector:JSoup CSS 选择器,指定从哪些元素抽取文本(默认
"body")。 - separator:多个被选中元素文本拼接所用的字符串(默认
"\n")。 - allElements:若为 true,忽略 selector,抽取
<body>中的全部文本(默认 false)。 - groupByElement:若为 true,为 selector 匹配到的每个元素创建单独的
Document(默认 false)。 - includeLinkUrls:若为 true,抽取绝对链接 URL 并加入元数据(默认 false)。
- metadataTags:要抽取内容的
<meta>标签名列表(默认["description", "keywords"])。 - additionalMetadata:允许为创建的所有
Document对象添加自定义元数据。
行为:
- selector 决定哪些元素用于文本抽取。
- 如果
allElements为 true,<body>中的所有文本会被抽取为单个Document。 - 如果
groupByElement为 true,每个匹配 selector 的元素会创建一个单独的Document。 - 文档标题、指定
<meta>标签的内容以及(可选的)链接 URL 会加入Document元数据。
读取器会保留选中元素的文本内容,但会移除其中的 HTML 标签。
3.4 MARKDOWN
MarkdownDocumentReader 处理 Markdown 文档。
依赖:
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-markdown-document-reader</artifactId>
</dependency>示例:
@Component
class MyMarkdownReader {
private final Resource resource;
MyMarkdownReader(@Value("classpath:code.md") Resource resource) {
this.resource = resource;
}
List<Document> loadMarkdown() {
MarkdownDocumentReaderConfig config = MarkdownDocumentReaderConfig.builder()
.withHorizontalRuleCreateDocument(true) // 分隔线会创建新文档
.withIncludeCodeBlock(false) // 代码块单独成文档
.withIncludeBlockquote(false) // 引用块单独成文档
.withAdditionalMetadata("filename", "code.md") // 添加自定义元数据
.build();
MarkdownDocumentReader reader = new MarkdownDocumentReader(this.resource, config);
return reader.get();
}
}MarkdownDocumentReaderConfig 字段:
- horizontalRuleCreateDocument:为 true 时,水平分隔线会创建新的
Document对象。 - includeCodeBlock:为 true 时,代码块与周围文本合并;为 false 时,代码块创建单独的
Document。 - includeBlockquote:为 true 时,引用块与周围文本合并;为 false 时,引用块创建单独的
Document。 - additionalMetadata:为创建的所有
Document对象添加自定义元数据。
行为:
- 标题(Headers)会成为
Document对象中的元数据。 - 段落(Paragraphs)会成为
Document对象的内容。 - 代码块可以拆分为独立的
Document对象,或与周围文本包含在一起。 - 引用块可以拆分为独立的
Document对象,或与周围文本包含在一起。 - 水平分隔线可用于将内容拆分为独立的
Document对象。
读取器会在 Document 对象的内容中保留行内代码、列表以及文本样式等格式。
3.5 PDF PAGE
PagePdfDocumentReader 使用 Apache PdfBox 库解析 PDF 文档。
依赖:
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-pdf-document-reader</artifactId>
</dependency>示例:
@Component
public class MyPagePdfDocumentReader {
List<Document> getDocsFromPdf() {
PagePdfDocumentReader pdfReader = new PagePdfDocumentReader("classpath:/sample1.pdf",
PdfDocumentReaderConfig.builder()
.withPageTopMargin(0) // 页面上边距设为 0
.withPageExtractedTextFormatter(ExtractedTextFormatter.builder()
.withNumberOfTopTextLinesToDelete(0) // 不删除任何顶部文本行
.build())
.withPagesPerDocument(1) // 每个文档包含 1 页
.build());
return pdfReader.read();
}
}3.6 PDF PARAGRAPH
ParagraphPdfDocumentReader 使用 PDF catalog(例如 TOC 目录)信息将输入 PDF 拆分为文本段落,每个段落输出一个 Document。
注意
并非所有 PDF 文档都包含 PDF catalog。如果 PDF 没有目录信息,这种 reader 就退化成"按页切"。
依赖:
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-pdf-document-reader</artifactId>
</dependency>示例:
@Component
public class MyPagePdfDocumentReader {
List<Document> getDocsFromPdfWithCatalog() {
ParagraphPdfDocumentReader pdfReader = new ParagraphPdfDocumentReader("classpath:/sample1.pdf",
PdfDocumentReaderConfig.builder()
.withPageTopMargin(0)
.withPageExtractedTextFormatter(ExtractedTextFormatter.builder()
.withNumberOfTopTextLinesToDelete(0)
.build())
.withPagesPerDocument(1)
.build());
return pdfReader.read();
}
}3.7 TIKA(DOCX、PPTX、HTML…)
TikaDocumentReader 使用 Apache Tika 从多种文档格式中提取文本,如 PDF、DOC/DOCX、PPT/PPTX 和 HTML。完整支持格式列表请参考 Tika 文档。
依赖:
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-tika-document-reader</artifactId>
</dependency>示例:
@Component
class MyTikaDocumentReader {
private final Resource resource;
MyTikaDocumentReader(@Value("classpath:/word-sample.docx")
Resource resource) {
this.resource = resource;
}
List<Document> loadText() {
TikaDocumentReader tikaDocumentReader = new TikaDocumentReader(this.resource);
return tikaDocumentReader.read();
}
}四、文档转换器(Transformers)
4.1 TextSplitter
TextSplitter 是一个抽象基类,用于帮助将文档拆分以适配 AI 模型的上下文窗口。
4.2 TokenTextSplitter
TokenTextSplitter 是 TextSplitter 的一个实现,根据 token 数量将文本拆分为块。它支持可配置的编码类型(如 CL100K_BASE、P50K_BASE、O200K_BASE),默认使用 CL100K_BASE。
使用:
@Component
class MyTokenTextSplitter {
public List<Document> splitDocuments(List<Document> documents) {
// 使用默认配置
TokenTextSplitter splitter = TokenTextSplitter.builder().build();
return splitter.apply(documents);
}
public List<Document> splitCustomized(List<Document> documents) {
// 自定义配置
TokenTextSplitter splitter = TokenTextSplitter.builder()
.withChunkSize(1000) // 每块 1000 token
.withMinChunkSizeChars(400) // 最小字符数 400
.withMinChunkLengthToEmbed(10) // 最小可嵌入长度 10
.withMaxNumChunks(5000) // 最多 5000 块
.withKeepSeparator(true) // 保留分隔符
.build();
return splitter.apply(documents);
}
}自定义编码类型:
// 使用 O200K_BASE 编码
TokenTextSplitter splitter = TokenTextSplitter.builder()
.withEncodingType(EncodingType.O200K_BASE)
.withChunkSize(1000)
.build();自定义标点符号:
@Component
class MyInternationalTextSplitter {
public List<Document> splitChineseText(List<Document> documents) {
// 使用中文标点
TokenTextSplitter splitter = TokenTextSplitter.builder()
.withChunkSize(800)
.withMinChunkSizeChars(350)
.withPunctuationMarks(List.of('。', '?', '!', ';')) // 中文标点
.build();
return splitter.apply(documents);
}
public List<Document> splitWithCustomMarks(List<Document> documents) {
// 混合中英文标点
TokenTextSplitter splitter = TokenTextSplitter.builder()
.withChunkSize(800)
.withPunctuationMarks(List.of('.', '?', '!', '\n', ';', ':', '。'))
.build();
return splitter.apply(documents);
}
}配置参数:
- encodingType:使用的 tokenizer 编码类型(默认
CL100K_BASE)。 - chunkSize:每个文本块的目标大小(token 数,默认 800)。
- minChunkSizeChars:每个文本块的最小字符数(默认 350)。
- minChunkLengthToEmbed:块被包含进输出的最小长度(默认 5)。
- maxNumChunks:从一段文本生成的最大块数(默认 10000)。
- keepSeparator:是否在块中保留分隔符(默认 true)。
- punctuationMarks:用作句子边界进行拆分的一系列字符(默认
., ?, !, \n)。
行为:
- 使用
CL100K_BASE编码将输入文本编码为 token。 - 根据
chunkSize将编码后的文本拆分为块。 - 对每个块:
- a. 将块解码回文本。
- b. 仅当总 token 数超过 chunk size 时,才尝试在
minChunkSizeChars之后使用配置的punctuationMarks找到合适的断点。 - c. 如果找到断点,则在该点截断块。
- d. 对块进行 trim,并根据
keepSeparator设置可选地移除换行符。 - e. 如果得到的块长度大于
minChunkLengthToEmbed,则加入输出。
- 该过程持续进行,直到所有 token 处理完毕或达到
maxNumChunks。 - 剩余文本如果长于
minChunkLengthToEmbed,则作为最后一块加入。
注意:基于标点的拆分仅在 token 数超过 chunk size 时适用。token 数恰好等于或小于 chunk size 的文本将作为单个块返回,不会进行基于标点的截断。
说明:
TokenTextSplitter使用 jtokkit 库的CL100K_BASE编码,与较新的 OpenAI 模型兼容。- 该拆分器尽可能在句子边界处断开,以创建语义上有意义的块。
- 原始文档的元数据会被保留并复制到由该文档派生的所有块中。
- 自定义标点符号:默认标点(
., ?, !, \n)适用于英文文本。对于其他语言或特殊内容,可通过 builder 的withPunctuationMarks()方法自定义标点。 - 性能考量:建议将标点列表保持相对较小(少于 20 个字符)以获得最佳性能。
- 小文本处理:从 2.0 版本起,小文本(token 数等于或小于 chunk size)不再按标点拆分。
4.3 ContentFormatTransformer
确保所有文档的内容格式统一。
4.4 KeywordMetadataEnricher
KeywordMetadataEnricher 是一个 DocumentTransformer,它使用生成式 AI 模型从文档内容中提取关键词并将其添加为元数据。
使用:
@Component
class MyKeywordEnricher {
private final ChatModel chatModel;
MyKeywordEnricher(ChatModel chatModel) {
this.chatModel = chatModel;
}
List<Document> enrichDocuments(List<Document> documents) {
// 使用默认模板,提取 5 个关键词
KeywordMetadataEnricher enricher = KeywordMetadataEnricher.builder(chatModel)
.keywordCount(5)
.build();
// 或使用自定义模板
KeywordMetadataEnricher enricher = KeywordMetadataEnricher.builder(chatModel)
.keywordsTemplate(YOUR_CUSTOM_TEMPLATE)
.build();
return enricher.apply(documents);
}
}构造函数选项:
KeywordMetadataEnricher(ChatModel chatModel, int keywordCount):使用默认模板并提取指定数量的关键词。KeywordMetadataEnricher(ChatModel chatModel, PromptTemplate keywordsTemplate):使用自定义模板进行关键词提取。
行为:
- 对每个输入文档,使用文档内容创建 prompt。
- 将该 prompt 发送给提供的
ChatModel以生成关键词。 - 生成的关键词会添加到文档元数据中,键为
"excerpt_keywords"。 - 返回增强后的文档。
默认模板:
{context_str}. Give %s unique keywords for this document. Format as comma separated. Keywords:说明:
KeywordMetadataEnricher需要一个可用的ChatModel来生成关键词。- 关键词数量必须大于等于 1。
- 该增强器会为每个处理的文档添加
"excerpt_keywords"元数据字段。 - 在 Builder 模式中,如果设置了
keywordsTemplate参数,keywordCount参数将被忽略。
4.5 SummaryMetadataEnricher
SummaryMetadataEnricher 是一个 DocumentTransformer,它使用生成式 AI 模型为文档创建摘要并将其添加为元数据。它可以为当前文档以及相邻文档(前一篇和后一篇)生成摘要。
使用:
@Configuration
class EnricherConfig {
@Bean
public SummaryMetadataEnricher summaryMetadata(OpenAiChatModel aiClient) {
// 为前一个、当前、后一个文档生成摘要
return new SummaryMetadataEnricher(aiClient,
List.of(SummaryType.PREVIOUS, SummaryType.CURRENT, SummaryType.NEXT));
}
}
@Component
class MySummaryEnricher {
private final SummaryMetadataEnricher enricher;
MySummaryEnricher(SummaryMetadataEnricher enricher) {
this.enricher = enricher;
}
List<Document> enrichDocuments(List<Document> documents) {
return this.enricher.apply(documents);
}
}构造函数:
SummaryMetadataEnricher(ChatModel chatModel, List<SummaryType> summaryTypes)SummaryMetadataEnricher(ChatModel chatModel, List<SummaryType> summaryTypes, String summaryTemplate, MetadataMode metadataMode)
参数:
- chatModel:用于生成摘要的 AI 模型。
- summaryTypes:
SummaryType枚举值列表(PREVIOUS、CURRENT、NEXT)。 - summaryTemplate:用于摘要生成的自定义模板(可选)。
- metadataMode:指定生成摘要时如何处理文档元数据(可选)。
行为:
- 对每个输入文档,使用文档内容和指定的摘要模板创建 prompt。
- 将该 prompt 发送给提供的
ChatModel以生成摘要。 - 根据指定的
summaryTypes,为每个文档添加以下元数据:- section_summary:当前文档的摘要。
- prev_section_summary:前一个文档的摘要(若存在且被请求)。
- next_section_summary:后一个文档的摘要(若存在且被请求)。
- 返回增强后的文档。
默认摘要模板:
"""
Here is the content of the section:
{context_str}
Summarize the key topics and entities of the section.
Summary:
"""说明:
- 该增强器可以处理任意大小的文档列表,并能正确应对第一个和最后一个文档的边界情况。
- 第一个文档获得
next_section_summary,但没有prev_section_summary;第二个文档反之。 - 第一个文档的
section_summary与第二个文档的prev_section_summary相同;第一个文档的next_section_summary与第二个文档的section_summary相同。
五、文档写入器(Writers)
5.1 File
FileDocumentWriter 是一个 DocumentWriter 实现,它将 Document 对象列表的内容写入文件。
使用:
@Component
class MyDocumentWriter {
public void writeDocuments(List<Document> documents) {
// 文件名,启用文档标记,使用全部元数据,追加模式关闭
FileDocumentWriter writer = new FileDocumentWriter("output.txt", true, MetadataMode.ALL, false);
writer.accept(documents);
}
}构造函数:
FileDocumentWriter(String fileName)FileDocumentWriter(String fileName, boolean withDocumentMarkers)FileDocumentWriter(String fileName, boolean withDocumentMarkers, MetadataMode metadataMode, boolean append)
参数:
- fileName:要写入文档的文件名。
- withDocumentMarkers:是否在输出中包含文档标记(默认 false)。
- metadataMode:指定写入文件的文档内容(默认
MetadataMode.NONE)。 - append:若为 true,数据将写入文件末尾而非开头(默认 false)。
文档标记格式:
### Doc: [index], pages:[start_page_number,end_page_number]说明:
- 该写入器使用
FileWriter,因此以操作系统默认字符编码写入文本文件。 - 如果写入过程中发生错误,会抛出一个
RuntimeException,并将原始异常作为 cause。
5.2 VectorStore
提供与各种向量存储(Vector Store)的集成。完整列表请参阅 Vector DB 文档。