飞书知识库 CDC（公测中）

变更到 changelog 的映射

连接器按主键 node_id 输出 upsert 语义的 changelog，包含 INSERT 和 UPDATE_AFTER，不输出 UPDATE_BEFORE。

当前实现不感知文档删除：文档被删除后不再出现在节点列表中，连接器不会输出 DELETE 事件，下游中该行会保留为最后一次同步的状态。

已删除文档的清理任务建议独立进行，直接调用飞书 OpenAPI 完成比对，低频、定期执行，可选择周末或夜间等业务低峰时段，避免与同步作业的轮询叠加争抢 OpenAPI 限流。比对方式如下：

1. 完整拉取一次知识空间节点列表（GET /open-apis/wiki/v2/spaces/{space_id}/nodes，须全部分页成功），与下游已存储的 node_id 比对，找出「下游有、列表无」的节点。

2. 对这类节点逐个调用GET /open-apis/wiki/v2/spaces/get_node二次确认：返回节点不存在类错误的，判定为真实删除；返回权限拒绝类错误的，说明文档仍存在但应用暂时失去访问权限，不得删除；限流、网络等瞬时错误则跳过本轮，下一周期再确认。

3. 确认为真实删除后执行清理，建议软删除，行数据保留用于审计和恢复；对误删零容忍的场景，可连续多个周期均确认删除后才执行清理。

权限点列表

上述接口对应的具体权限点（scope）名称以飞书开放平台文档为准。授权时需要覆盖以下两类能力：

语法结构

CREATE TEMPORARY TABLE <yourTableName> (
  node_id STRING NOT NULL,
  title STRING,
  url STRING,
  `type` STRING,
  content STRING,
  updated_time TIMESTAMP_LTZ(3),
  PRIMARY KEY (node_id) NOT ENFORCED
) WITH (
  'connector' = 'feishu-cdc',
  'object-type' = 'document',
  'app-id' = '<yourAppId>',
  'app-secret' = '<yourAppSecret>',
  'wiki.space-id' = '<yourWikiSpaceId>'
);

Flink SQL DDL 必须声明下表中全部 6 个物理字段，字段顺序和名称需保持一致。

源表必须声明 PRIMARY KEY (node_id) NOT ENFORCED。

此外，DDL 可以通过 METADATA 声明以下可读元数据。元数据列不改变上述物理 Schema。

WITH 参数

基础参数

读取范围参数

启动和轮询参数

请求和限流参数

代码示例

全量加增量同步

示例将指定知识空间的文档写入 Paimon 表，声明了 blocks_json 元数据列。

CREATE TEMPORARY TABLE feishu_wiki_source (
  node_id STRING NOT NULL,
  title STRING,
  url STRING,
  `type` STRING,
  content STRING,
  updated_time TIMESTAMP_LTZ(3),
  blocks_json STRING METADATA VIRTUAL,
  PRIMARY KEY (node_id) NOT ENFORCED
) WITH (
  'connector' = 'feishu-cdc',
  'object-type' = 'document',
  'app-id' = '<yourAppId>',
  'app-secret' = '<yourAppSecret>',
  'wiki.space-id' = '<yourWikiSpaceId>',
  'scan.startup.mode' = 'initial'
);

CREATE TABLE IF NOT EXISTS paimon.feishu_wiki.documents (
  node_id STRING,
  title STRING,
  url STRING,
  `type` STRING,
  content STRING,
  blocks_json STRING,
  updated_time TIMESTAMP_LTZ(3),
  PRIMARY KEY (node_id) NOT ENFORCED
);

INSERT INTO paimon.feishu_wiki.documents
SELECT node_id, title, url, `type`, content, blocks_json, updated_time
FROM feishu_wiki_source;

SNAPSHOT 全量扫描

SNAPSHOT 只做一次全量读取，适合内容迁移或一次性导出。

CREATE TEMPORARY TABLE feishu_wiki_snapshot (
  node_id STRING NOT NULL,
  title STRING,
  url STRING,
  `type` STRING,
  content STRING,
  updated_time TIMESTAMP_LTZ(3),
  PRIMARY KEY (node_id) NOT ENFORCED
) WITH (
  'connector' = 'feishu-cdc',
  'object-type' = 'document',
  'app-id' = '<yourAppId>',
  'app-secret' = '<yourAppSecret>',
  'wiki.space-id' = '<yourWikiSpaceId>',
  'scan.startup.mode' = 'snapshot'
);

如需只同步空间内的一篇文档，在 WITH 参数中追加 'wiki.document-url' = '<文档链接>' 或 'wiki.node-token' = '<节点Token>'。

语法结构

source:
  type: feishu-cdc
  name: <yourSourceName>
  app-id: <yourAppId>
  app-secret: <yourAppSecret>
  wiki.space-id: <yourWikiSpaceId>
  scan.startup.mode: initial

配置项

基础参数

同步范围参数

启动和轮询参数

请求和限流参数

固定 Schema

固定 Schema 的主键为 node_id。与 Flink SQL 相比，YAML Schema 增加了 space_id 物理列，blocks_json 也作为物理列直接输出。

空间到表的路由

未配置 wiki.space-id 时，连接器自动发现应用有权读取的全部知识空间，每个空间生成对应的 <知识空间ID>.documents 表。配置 wiki.space-id（支持逗号分隔多个值）后，只同步指定的空间。

配置文档选择器 wiki.document-url 或 wiki.node-token（均支持逗号分隔多个值）后，选择器与 wiki.space-id 默认取并集：配置的空间仍整体同步，选择器不做收窄；未配置的空间如果包含所选文档，也会被自动发现并同步这些文档。wiki.restrict-documents-to-selected-spaces 控制是否把选择器严格限制在 wiki.space-id 配置的空间内。

假设 wiki.space-id 配置为空间 A，文档选择器包含属于 A 的文档 X 和属于 B 的文档 Y：

如想让 wiki.space-id 作为严格同步边界，建议设置为 true，避免选择器静默突破边界、同步到预期外的空间。

代码示例

以下示例把指定知识空间中的文档全量读取后持续同步增量变化，写入 Paimon。

source:
  type: feishu-cdc
  name: Feishu Wiki Source
  app-id: <yourAppId>
  app-secret: <yourAppSecret>
  wiki.space-id: <yourWikiSpaceId>
  scan.startup.mode: initial

route:
  - source-table: <yourWikiSpaceId>.documents
    sink-table: feishu_wiki.documents

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: <yourDlfCatalogEndpoint>
  catalog.properties.warehouse: <yourWarehouse>