功能概述 #

Lance Flink SQL Connector 用于在火山引擎流式计算 Flink 版中读取或写入存储在对象存储 TOS 上的 Lance 数据集。Connector 同时提供 Source 和 Sink 能力，适用于向量、文本和其他 AI/机器学习数据的批式处理与流式接入。

版本兼容性 #

特色功能 #

• 支持以批模式或流模式向 TOS 写入 Lance 数据集。
• 支持批量读取已有 Lance 数据集，也支持按固定间隔发现新增数据。
• 支持通过 write.large-string-columns 将指定 STRING 字段写为 Arrow LargeUtf8，适合较大的文本或结构化字符串字段。
• 支持通过文件行数和批量大小控制 Sink 输出文件规模与吞吐。

前提条件 #

1. 已在 TOS 创建用于存储 Lance 数据集的 FNS Bucket；HNS Bucket 不受支持。
2. Flink 作业账号对目标 Bucket 具有读取、写入和 List 权限。
3. 已从 TOS 控制台获取 Bucket 的 S3 Endpoint。
4. 已准备访问 TOS 所需的 AK 和 SK，并在流式计算 Flink 版中创建加密变量。
5. Flink 资源池能够访问目标 TOS Bucket 的 S3 Endpoint。
6. 使用 Streaming Sink 时已启用 Checkpoint；Connector 在 Checkpoint 时提交数据，以保证 Exactly-Once。

使用限制 #

• 仅支持 TOS FNS Bucket，不支持 HNS Bucket。
• path 必须使用 tos://<bucket>/<path> 格式，并指向有权限访问的 Bucket 和目录。
• write.large-string-columns 中的字段名按 Sink 表 Schema 精确匹配并区分大小写，多个字段使用英文逗号分隔。
• Streaming Source 启动时先读取数据集中已有的 Fragment，然后按照发现间隔读取新增 Fragment；只支持追加数据集。
• Sink 仅接受 INSERT。上游产生 UPDATE 或 DELETE 时不受支持，应在写入前转换为仅追加数据。
• Streaming Sink 依赖 Checkpoint 提交数据并保证 Exactly-Once；生产作业必须启用 Checkpoint。
• 本文只介绍 Lance Connector 的通用读写能力，不包含读取 Lance 后写入特定下游 Connector 的组合方案。

语法结构 #

Sink #

CREATE TABLE lance_sink ( 
    id BIGINT,
    content STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'lance',
    'path' = 'tos://<bucket>/<path>',
    's3-access-key' = '${secret_values.tos-access-key}',
    's3-secret-key' = '${secret_values.tos-secret-key}',
    's3-endpoint' = '<tos_s3_endpoint>'
);

Source #

CREATE TABLE lance_source ( 
    id BIGINT,
    content STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'lance',
    'path' = 'tos://<bucket>/<path>',
    's3-access-key' = '${secret_values.tos-access-key}',
    's3-secret-key' = '${secret_values.tos-secret-key}',
    's3-endpoint' = '<tos_s3_endpoint>',
    'source.mode' = 'batch'
);

参数说明 #

通用参数 #

Sink 独有参数 #

Source 独有参数 #

使用示例 #

批模式写入 #

CREATE TABLE src (
    id BIGINT, 
    content STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'datagen',
    'number-of-rows' = '200',
    'rows-per-second' = '1',
    'fields.id.kind' = 'sequence',
    'fields.id.start' = '1',
    'fields.id.end' = '200'
);

CREATE TABLE lance_sink (
    id BIGINT,
    content STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'lance',
    'path' = 'tos://<bucket>/<dataset_path>',
    's3-access-key' = '${secret_values.tos-access-key}',
    's3-secret-key' = '${secret_values.tos-secret-key}',
    's3-endpoint' = '<tos_s3_endpoint>',
    'write.batch-size' = '1024'
);

INSERT INTO lance_sink SELECT * FROM src;

预期结果。 作业完成后，目标 TOS 路径中生成 Lance 数据集及数据 Fragment。

流模式写入 #

将 datagen Source 改为无界数据源，并以 Streaming 方式提交相同的 INSERT INTO 语句，即可持续向目标 Lance 数据集写入数据。上游只能产生 INSERT。Connector 在 Checkpoint 时提交数据，因此必须为作业启用 Checkpoint。

批模式读取 #

CREATE TABLE lance_source (
    id BIGINT,
    content STRING, 
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'lance',
    'path' = 'tos://<bucket>/<dataset_path>',
    's3-access-key' = '${secret_values.tos-access-key}',
    's3-secret-key' = '${secret_values.tos-secret-key}',
    's3-endpoint' = '<tos_s3_endpoint>',
    'source.mode' = 'batch'
);

CREATE TABLE print_sink (
    id BIGINT,
    content STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'print'
);

INSERT INTO print_sink SELECT * FROM lance_source;

预期结果。 作业读取当前 Lance 数据集的记录，并将结果输出到 Print Sink。

流模式读取 #

CREATE TABLE lance_streaming_source (
    id BIGINT,
    content STRING,
    embedding ARRAY<FLOAT> 
) WITH (
    'connector' = 'lance',
    'path' = 'tos://<bucket>/<dataset_path>',
    's3-access-key' = '${secret_values.tos-access-key}',
    's3-secret-key' = '${secret_values.tos-secret-key}',
    's3-endpoint' = '<tos_s3_endpoint>',
    'source.mode' = 'streaming',
    'source.discovery-interval' = '10000'
);

Streaming Source 启动时先读取当前数据集已有的 Fragment，然后按照配置间隔发现并读取新增 Fragment。该模式只支持追加数据集。

运行机制 #

Sink 仅处理 INSERT，按 write.batch-size 累积记录，并通过 write.max-rows-per-file 控制单个 Lance Fragment 的最大行数。Streaming Sink 在 Checkpoint 时提交数据，从而保证 Exactly-Once；作业恢复后从最近一次成功 Checkpoint 继续处理。
Source 在 Batch 模式读取当前数据集后结束。Streaming Source 先读取启动时已存在的 Fragment，再按照 source.discovery-interval 定期发现并读取新增 Fragment；数据集必须保持追加写入。

验证与排障 #

验证方法 #

1. 提交批写作业，确认作业成功结束，且目标 TOS 路径中生成 Lance 数据集文件。
2. 使用相同 Schema 提交批读作业，核对读取记录数和样例字段值。
3. 提交流读作业后向目标数据集追加数据，确认新数据在发现间隔后被下游读取。
4. 提交流写作业，等待至少一次 Checkpoint 成功完成，再检查目标数据集；执行受控重启后确认数据没有重复或丢失。
5. 检查作业日志中是否存在 TOS 权限、Endpoint、Schema 或路径错误。

常见问题 #

注意事项 #

• AK 和 SK 必须通过 Flink 加密变量引用，不要明文写入 SQL。
• 生产环境建议使用与 Flink 资源池同地域的 TOS 内网 S3 Endpoint。
• 写入前确认目标路径不会与其他不兼容 Schema 的 Lance 数据集冲突。
• Streaming Sink 必须启用 Checkpoint，且上游只能产生 INSERT。
• Streaming Source 只适用于持续追加 Fragment 的 Lance 数据集。

相关文档 #

• Lance 官方网站
• Lance 开源项目
• 火山引擎对象存储 TOS
• 火山引擎流式计算 Flink 版：变量管理
• 火山引擎流式计算 Flink 版：上线 Flink 任务