Doris Catalog 通过提供统一的元数据管理和数据访问能力，支持 Flink 作业直接读写 Apache Doris 数据库中的表。您可以在 Flink SQL 中直接使用三段式表名（catalog.database.table）访问 Doris 数据表，无需在每次作业中重复配置连接参数。

Doris Catalog 通过 Flink Session 集群连接 Doris FE，获取 Doris 中的数据库、表、字段类型和字段注释等元数据。同步完成后，可以在数据目录中查看 Doris 元数据，并在 Flink SQL 任务中使用三段式名称引用已同步的 Doris 表。
Doris 中新增或删除表、增加字段、修改字段类型或字段注释后，需要重新同步 Catalog，使数据目录中的元数据与 Doris 保持一致。

Doris 侧前提 #

1. 已有可用的 Doris 集群。
2. 已获取 Doris FE 的 MySQL/JDBC 查询地址和 HTTP 地址。
3. 已准备数据库账号和密码，并授予元数据读取权限；写入场景还需要目标表写权限。
4. 使用 Sink、Source 或 Lookup 能力前，已在 Doris 中创建目标数据库和表。

Flink 侧前提 #

1. Flink 引擎版本为 1.17-volcano 及以上。
2. 已有可用的资源池。
3. 已创建可用的 Session 集群，用于执行连通性测试和元数据同步。
4. Flink 资源池所在网络可以访问 Doris FE。

在流式计算 Flink 版控制台的数据目录页面创建 Doris Catalog。

1. 进入项目空间，打开数据目录（Beta）。
2. 单击创建 Catalog，然后选择 Doris 类型。
3. 填写 Doris 连接信息，选择资源池和 Session 集群。
4. 单击连通性测试。测试通过后，单击创建 Catalog完成创建。

参数说明 #

同步 Catalog #

1. 在 Catalog 列表中找到已创建的 Doris Catalog。
2. 单击同步 Catalog，触发元数据同步。
   
3. 同步成功后可展开数据目录下的数据库表名称来查看已同步的表元数据信息。
4. Doris 中新增或删除表、增加字段、修改字段类型或字段注释后，在 Catalog 列表中找到目标 Catalog，并单击同步 Catalog 重新获取元数据。

查看、编辑或删除 Catalog #

在 Catalog 列表中找到目标 Catalog，然后根据需要单击 查看、编辑 或 删除。

在 Flink SQL 任务中，可以使用 catalog.database.table 三段式名称引用已同步的 Doris 表。以下示例中的 Catalog 名称、数据库名称、表名称和字段需替换为实际值。

写入 Doris 表 #

以下示例使用 Datagen 连接器生成测试数据，并向包含 JSON 字段的 Doris Unique Key 表写入数据。目标表需要预先在 Doris 中创建并同步到数据目录。

CREATE TABLE datagen_source (
  id INTEGER,
  name STRING,
  score INTEGER,
  user_info STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.id.kind' = 'sequence',
  'fields.id.start' = '1',
  'fields.id.end' = '20',
  'fields.score.min' = '0',
  'fields.score.max' = '100',
  'fields.name.length' = '8',
  'fields.user_info.length' = '10'
);

INSERT INTO `<catalog_name>`.`<database_name>`.`<table_name>`
SELECT
  id,
  name,
  score,
  JSON_OBJECT(
    KEY 'device' VALUE JSON_OBJECT(
      KEY 'os' VALUE IF(MOD(id, 2) = 0, 'iOS', 'Android')
    ),
    KEY 'level' VALUE score,
    KEY 'metadata' VALUE JSON_OBJECT(
      KEY 'raw_str' VALUE user_info
    )
  ) AS user_info
FROM datagen_source;

读取 Doris 表 #

以下示例读取 Doris Catalog 表，将 JSON 字段解析为字符串，并通过 Print 连接器输出结果。

CREATE TABLE print_sink (
  id INTEGER,
  name STRING,
  score INTEGER,
  parsed_os_type STRING,
  raw_user_info STRING
) WITH (
  'connector' = 'print'
);

INSERT INTO print_sink
SELECT
  id,
  name,
  score,
  JSON_VALUE(user_info, '$.device.os') AS parsed_os_type,
  user_info AS raw_user_info
FROM `<catalog_name>`.`<database_name>`.`<table_name>`;

使用 Doris 表进行 Lookup Join #

以下示例将 Doris Catalog 表作为 Lookup 表，根据处理时间与实时流进行时态表 Join。

CREATE TABLE realtime_stream (
  req_id INTEGER,
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.req_id.kind' = 'sequence',
  'fields.req_id.start' = '1',
  'fields.req_id.end' = '5'
);

CREATE TABLE print_sink (
  req_id INTEGER,
  matched_name STRING,
  matched_score INTEGER,
  parsed_json_os STRING,
  raw_json STRING
) WITH (
  'connector' = 'print'
);

INSERT INTO print_sink
SELECT
  req_id,
  name AS matched_name,
  score AS matched_score,
  JSON_VALUE(user_info, '$.device.os') AS parsed_json_os,
  user_info AS raw_json
FROM realtime_stream
JOIN `<catalog_name>`.`<database_name>`.`<table_name>`
FOR SYSTEM_TIME AS OF proc_time
ON req_id = id;

创建或连通性测试失败 #

检查 JDBC 查询地址、FE HTTP 地址、用户名、密码和网络连通性。元数据访问和数据写入可能使用不同地址，需要分别检查。

元数据同步失败 #

检查 Doris FE 是否可访问、账号是否具有元数据读取权限，以及 Session 集群是否运行。

写入失败但元数据可查看 #

检查 FE HTTP 地址、目标表写权限，以及源表与目标表的字段数量、顺序和类型是否兼容。