基于 Tantivy 的 DLF Paimon 表全文检索-数据湖构建(DLF)-阿里云帮助中心

DLF 支持基于 Tantivy 引擎为 Paimon Append-only 表的文本列创建全文索引。您可以通过 PyPaimon 或 Spark SQL 执行全文检索。

功能说明

全文检索基于 Tantivy 搜索引擎，为 Paimon Append-only 表的 STRING 类型列提供基于倒排索引的检索能力。适用于日志搜索、文档检索、内容过滤等场景。

特性	说明
索引引擎	Tantivy（Rust 实现，通过 JNI/Python 绑定调用）
支持的列类型	STRING（VARCHAR / CHAR）
搜索类型	关键词匹配、布尔查询（AND/OR）

前提条件

根据您的使用方式，确保满足以下环境要求：

PyPaimon：已安装 pypaimon 和 tantivy 依赖。平台要求为 Linux x86_64（glibc >= 2.34，如 Ubuntu 22.04+）。
1. 下载并安装pypaimon-1.5.dev20260704.tar.gz：
```
pip install pypaimon-1.5.dev20260608.tar.gz
```
2. 下载并安装tantivy-0.25.1-manylinux_x86_64.tar.gz：
```
tar xzf tantivy-0.25.1-manylinux_x86_64.tar.gz
pip install tantivy --no-index --find-links .
```
3. （可选）如需 DuckDB 输出格式，还需安装：
```
pip install duckdb
```

Spark SQL：下载并配置自定义 JARpaimon-ali-emr-spark-3.5-fulltext-sample.jar。

spark.emr.serverless.excludedModules                paimon
spark.emr.serverless.user.defined.jars              oss://<your-bucket>/paimon-ali-emr-spark-3.5-fulltext-sample.jar

创建表并开启全文检索

通过 Flink SQL 创建

CREATE TABLE articles (
    id INT,
    title STRING,
    content STRING
) WITH (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'morax.tantivy-fulltext-index.enabled' = 'true',
    'global-index.tantivy-fulltext.index-column' = 'content'
);

通过 Spark SQL 创建

CREATE TABLE articles (
    id INT,
    title STRING,
    content STRING
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'morax.tantivy-fulltext-index.enabled' = 'true',
    'global-index.tantivy-fulltext.index-column' = 'content'
);

建表参数说明

参数	说明
`row-tracking.enabled = true`	行级跟踪，用于关联索引与数据行。
`data-evolution.enabled = true`	数据演化，支持索引的增量构建和长期维护。
`morax.tantivy-fulltext-index.enabled = true`	开启全文索引自动调度。
`global-index.tantivy-fulltext.index-column`	索引列名，必须为 STRING 类型。

写入数据并触发索引构建

以下 INSERT 语句适用于 Flink SQL 和 Spark SQL：

INSERT INTO articles VALUES
    (1, 'lake storage', 'apache paimon is a lake storage format for big data'),
    (2, 'stream engine', 'flink is a stream processing engine for real time analytics'),
    (3, 'data evolution', 'paimon supports data evolution and row tracking features'),
    (4, 'query engine', 'spark sql can query paimon tables directly with high performance'),
    (5, 'distributed', 'ray data provides distributed data processing capabilities');

写入数据后：

DLF 自动调度全文索引构建。
索引构建为异步任务，构建完成后查询自动使用索引加速。
您可在 DLF 控制台查看全局索引构建进度：进入目标 Catalog > 目标表 > 文件列表页签，索引构建完成后可看到全文索引文件（如 tantivy-global-index-<UUID>.index）。

全文搜索

方式一：通过 PyPaimon 搜索

通过 PyPaimon 执行全文搜索，无需启动 Flink/Spark 引擎。

初始化 Catalog

执行全文搜索前需要初始化 Catalog 连接 DLF。完整参数说明请参见 PyPaimon与Ray Data。

from pypaimon import CatalogFactory

CATALOG_OPTIONS = {
    "metastore": "rest",
    "uri": "http://<DLF-ENDPOINT>",  # VPC 环境使用 http://<REGION>-vpc.dlf.aliyuncs.com
    "warehouse": "<YOUR-CATALOG>",
    "token.provider": "dlf",
    "dlf.region": "<REGION-ID>",
    "dlf.access-key-id": "<ACCESS-KEY-ID>",
    "dlf.access-key-secret": "<ACCESS-KEY-SECRET>",
    "dlf.oss-endpoint": "<OSS-ENDPOINT>",
}

catalog = CatalogFactory.create(CATALOG_OPTIONS)

基本用法

执行全文搜索，获取匹配行 ID：

table = catalog.get_table('default.articles')

builder = table.new_full_text_search_builder()
builder.with_text_column('content')
builder.with_query_text('paimon')
builder.with_limit(3)
result = builder.execute_local()

根据搜索结果读取实际行数据：

read_builder = table.new_read_builder()
read_builder = read_builder.with_projection(['id', 'title', 'content'])
scan = read_builder.new_scan().with_global_index_result(result)
splits = scan.plan().splits()
table_read = read_builder.new_read()
df = table_read.to_pandas(splits)
print(df)

输出示例：

id	content
1	apache paimon is a lake storage format for big data
3	paimon supports data evolution and row tracking features
4	spark sql can query paimon tables directly with high performance

获取相关性分数

# 接续上方的 builder 和 result 变量
score_fn = result.score_getter()
for row_id in result.results():
    print(f"row_id={row_id}, score={score_fn(row_id)}")

输出示例：

row_id=0, score=1.0508
row_id=2, score=1.1567
row_id=3, score=1.0508

布尔查询

AND 模式（所有关键词都必须出现）：

builder = table.new_full_text_search_builder()
builder.with_text_column('content')
builder.with_query_operator('and')
builder.with_query_text('data processing')
builder.with_limit(10)
result = builder.execute_local()

输出示例：

id	content
5	ray data provides distributed data processing capabilities

OR 模式（默认，任一关键词出现即匹配）：

builder = table.new_full_text_search_builder()
builder.with_text_column('content')
builder.with_query_operator('or')
builder.with_query_text('spark flink')
builder.with_limit(10)
result = builder.execute_local()

输出格式

PyPaimon 支持多种输出格式：

# PyArrow Table
arrow_table = table_read.to_arrow(splits)
print(arrow_table)

# DuckDB
conn = table_read.to_duckdb(splits, 'articles')
print(conn.execute('SELECT * FROM articles').fetchdf())

方式二：通过 Spark SQL 搜索

-- 搜索包含"paimon"的文章，返回 Top-3
SELECT * FROM full_text_search('articles', 'content', 'paimon', 3);

输出示例：

id	content
1	apache paimon is a lake storage format for big data
3	paimon supports data evolution and row tracking features
4	spark sql can query paimon tables directly with high performance

结合投影和过滤：

SELECT id, title FROM full_text_search('default.articles', 'content', 'data', 10)
  WHERE id > 1;

full_text_search 函数参数说明：

参数	类型	说明
table_name	STRING	表名，支持 `db.table` 或 `catalog.db.table` 格式。
column_name	STRING	文本列名，类型必须为 STRING。
query_text	STRING	搜索关键词。
limit	INT	返回结果数量。

使用限制

限制项	说明
表类型	仅支持 Append-only 表。
列类型	仅支持 STRING（VARCHAR / CHAR）列，不支持数值、数组等类型。
索引列数量	每个表只能对一个 STRING 列建全文索引。
索引构建	索引通过 Flink 批作业异步构建，写入数据后需等待构建完成才能搜索。
搜索精度	基于倒排索引的关键词匹配，非语义搜索。如需语义搜索，请参见构建 Global Index 实现查询加速和向量搜索。
NULL 值	若文本列值为 NULL，该行不参与索引构建和全文搜索。
分词器	当前版本使用 default 分词器（英文优化），暂不支持中文分词（jieba）及其他分词器。