视频流（RTSP/RTMP/HLS）（公测中）

语法结构

CREATE TEMPORARY TABLE video_stream_source (
  video_source STRING NOT NULL,
  start_ts BIGINT NOT NULL,
  end_ts BIGINT NOT NULL,
  `blob` BYTES NOT NULL,
  mime_type STRING NOT NULL
) WITH (
  'connector' = 'video-stream',
  'video-stream.url-list' = '<videoStreamUrl>',
  'mode' = 'KEYFRAME_EXTRACT',
  'sample-interval' = '5s'
);

DDL 必须按照以上顺序声明固定 Schema 中的全部五列。列的 NULL 约束不影响校验，但建议与连接器输出一致，将所有列声明为 NOT NULL。

WITH 参数

以下为视频流连接器在 Flink SQL 中的 WITH 参数。connector 固定填写 video-stream。

元数据字段映射

视频流连接器使用固定 Schema。SQL DDL 必须声明以下全部列，并保持相同顺序。

使用示例

CREATE TEMPORARY TABLE video_stream_source (
  video_source STRING NOT NULL,
  start_ts BIGINT NOT NULL,
  end_ts BIGINT NOT NULL,
  `blob` BYTES NOT NULL,
  mime_type STRING NOT NULL
) WITH (
  'connector' = 'video-stream',
  'video-stream.url-list' = 'rtsp://<host>:<port>/<path>',
  'mode' = 'KEYFRAME_EXTRACT',
  'sample-interval' = '5s',
  -- 以下参数直接透传给 FFmpeg 库
  'ffmpeg.properties.rtsp_transport' = 'tcp',
  'ffmpeg-video.properties.threads' = '1'
);

INSERT INTO paimon_catalog.default.video_stream_table
SELECT video_source, start_ts, end_ts, `blob`, mime_type
FROM video_stream_source;

CREATE TEMPORARY TABLE video_clip_source (
  video_source STRING NOT NULL,
  start_ts BIGINT NOT NULL,
  end_ts BIGINT NOT NULL,
  `blob` BYTES NOT NULL,
  mime_type STRING NOT NULL
) WITH (
  'connector' = 'video-stream',
  'video-stream.url-list' = 'rtmp://<host>:<port>/<app>/<stream>',
  'mode' = 'MOVIE_CLIP',
  'sample-interval' = '10s',
  'record-queue-capacity' = '4'
);

CREATE TEMPORARY TABLE multi_video_source (
  video_source STRING NOT NULL,
  start_ts BIGINT NOT NULL,
  end_ts BIGINT NOT NULL,
  `blob` BYTES NOT NULL,
  mime_type STRING NOT NULL
) WITH (
  'connector' = 'video-stream',
  'video-stream.url-list' = 'https://<host>/live.m3u8?tracks=video,audio|rtsp://<host>/<camera>',
  'video-stream.url-list-separator' = '|',
  'mode' = 'KEYFRAME_EXTRACT',
  'sample-interval' = '3s'
);

CREATE TABLE my_catalog.my_db.video_archive (
  video_source STRING,
  start_ts BIGINT,
  end_ts BIGINT,
  `blob` BYTES,
  mime_type STRING
) WITH (
  'blob-field' = 'blob',
  'row-tracking.enabled' = 'true',
  'data-evolution.enabled' = 'true'
);

INSERT INTO paimon_catalog.default.video_stream_table
SELECT video_source, start_ts, end_ts, `blob`, mime_type
FROM video_stream_source;

语法结构

source:
  type: video-stream
  name: Video Stream Source
  video-stream.url-list: rtsp://<host>:<port>/<path>
  mode: KEYFRAME_EXTRACT
  sample-interval: 5s

transform:
  - source-table: default
    projection: video_source, start_ts, end_ts, blob, mime_type
    table-options: blob-field=blob;row-tracking.enabled=true;data-evolution.enabled=true

route:
  - source-table: default
    sink-table: <sinkDatabase>.<sinkTable>

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: <catalogEndpoint>
  catalog.properties.warehouse: <warehouseName>

未配置 url.to.table-id.matching.pattern 时，所有视频流均写入名为 default 的源表。

配置项

以下为视频流连接器在数据摄入 YAML 中的配置项。type 固定填写 video-stream。

URL 与 Table ID 映射

url.to.table-id.matching.pattern 使用 Java 正则表达式对完整 URL 进行匹配，并按照捕获组数量生成 Table ID。例如，视频地址为 rtmp://server/video/shop001/camera_01，可使用以下配置将其映射为 shop001.camera_01：

url.to.table-id.matching.pattern: ^rtmp://[^/]+/video/([^/]+)/([^/?]+)$

正则表达式必须有且仅有 1～3 个捕获组，且捕获结果不能为空。未配置正则表达式或 URL 未匹配时，连接器会使用 default 作为 Table ID。

使用示例

source:
  type: video-stream
  name: Camera Source
  video-stream.url-list: rtsp://<host>:<port>/<camera>
  mode: KEYFRAME_EXTRACT
  sample-interval: 5s
  record-queue-capacity: 8
  ffmpeg.properties.rtsp_transport: tcp
  ffmpeg.properties.timeout: 10000000

transform:
  - source-table: default
    projection: video_source, start_ts, end_ts, blob, mime_type
    table-options: blob-field=blob;row-tracking.enabled=true;data-evolution.enabled=true

route:
  - source-table: default
    sink-table: video_db.camera_frames

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: <catalogEndpoint>
  catalog.properties.warehouse: <warehouseName>

pipeline:
  name: Camera Frames Pipeline
  parallelism: 1

source:
  type: video-stream
  name: Multi-camera Source
  video-stream.url-list: rtmp://server/video/shop001/camera_01,rtmp://server/video/shop002/camera_02
  mode: MOVIE_CLIP
  sample-interval: 10s
  url.to.table-id.matching.pattern: ^rtmp://[^/]+/video/([^/]+)/([^/?]+)$

route:
  - source-table: shop001.camera_01
    sink-table: video_db.shop001_camera_01
  - source-table: shop002.camera_02
    sink-table: video_db.shop002_camera_02

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: <catalogEndpoint>
  catalog.properties.warehouse: <warehouseName>

pipeline:
  name: Multi-camera Clips Pipeline
  parallelism: 2

作业启动后没有输出数据

依次检查以下项目：

1. TaskManager 是否能够访问视频 URL。

2. FFmpeg 是否支持该协议、封装格式和编码格式。

3. 视频流是否包含可解码的视频轨道。

4. sample-interval 是否设置得过大。

5. RTSP 服务是否要求配置 ffmpeg.properties.rtsp_transport: tcp。

为什么作业恢复后没有从中断位置继续读取

由于直播流没有可稳定恢复的媒体位点，Checkpoint 只能恢复 URL Split 分配，故障期间可能存在未采集的画面，重新连接后也需要说明是否可能产生重复记录。因此，视频流连接器不提供从中断位置续读或端到端 exactly-once 保证。

如何降低 CPU 和内存消耗

• 增大 sample-interval，减少图片数量或视频切片频率。

• 只需要单帧分析时优先使用 KEYFRAME_EXTRACT。

• 调小 record-queue-capacity，限制积压媒体记录数。

• 将多路 URL 分配到适当的 Source 并行度，避免单个 Source Reader 同时承担过多视频流。

• 建议将 ffmpeg-video.properties.threads 设置为 1 以降低 CPU 占用率。