使用Flink批任务导出：

-- 创建Source，链接StarRocks的表
CREATE TABLE IF NOT EXISTS `source_mcp_metric` (
`logtime` STRING NOT NULL,
`log_id` STRING NOT NULL,
`user` STRING NOT NULL,
`department_id` STRING NOT NULL,
`mcp_name` STRING NULL,
`tool_name` STRING NULL,
`tool_id` STRING NULL,
`toolkit_id` STRING NULL,
`request_time` STRING NULL,
`request` STRING NULL,
`response` STRING NULL,
`cost_time` BIGINT NULL,
`status` BIGINT NULL,
`info` STRING NULL,
`env` STRING NULL,
`app_name` STRING NULL,
`label` BIGINT NULL,
`type` BIGINT NULL,
`trace_id` STRING NULL,
`pt` date NULL,
`version` STRING NULL,
  PRIMARY KEY (pt) NOT ENFORCED
) WITH (
  'connector' = 'starrocks',
  'jdbc-url' = 'jdbc:mysql:loadbalance://fe-query-lvs:9030',
  'scan-url'='fe-lvs:8030',
  'database-name' = 'starrocks',
  'table-name' = 'mcp_metric',
  'username' = 'root',
  'password' = '12345'
);

-- 创建iceberg catalog
CREATE CATALOG iceberg_hive_catalog WITH (
  'type' = 'iceberg',
  'default-database' = 'default',
  'catalog-type' = 'hive',
  'uri' = 'thrift://PRO:9083',
  'clients' = '5',
  'property-version' = '1',
  'hive-conf-dir' = '/app/hive/conf'
);
USE CATALOG iceberg_hive_catalog;

insert into  iceberg_hive_catalog.app_dev.sink_mcp_metric
select 
`logtime`,
`log_id`,
`user`,
`department_id`,
`mcp_name`,
`tool_name`,
`tool_id`,
`toolkit_id`,
`request_time`,
`request`,
`response`,
`cost_time`,
`status`,
`info`,
`env`,
`app_name`,
`label`,
`type`,
`trace_id`,
`version`,
REPLACE(cast(pt as string), '-', '') 
from
default_catalog.default_database.source_mcp_metric
where pt >= TO_DATE('20260101', 'yyyyMMdd') 
and pt < TO_DATE('20260201', 'yyyyMMdd');