Video stream (RTSP/RTMP/HLS) (Beta)

Syntax

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'
);

The DDL must declare all five columns of the fixed schema in the order shown above. The NULL constraints of the columns do not affect validation, but we recommend that you declare all columns as NOT NULL to match the connector output.

WITH parameters

The following table lists the WITH parameters of the video stream connector in Flink SQL. Set connector to video-stream.

Metadata field mapping

The video stream connector uses a fixed schema. The SQL DDL must declare all of the following columns in the same order.

Examples

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',
  -- The following parameters are passed through directly to the FFmpeg library
  '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;

Syntax

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>

If url.to.table-id.matching.pattern is not configured, all video streams are written to a source table named default.

Configurations

The following table lists the configurations of the video stream connector in data ingestion YAML. Set type to video-stream.

URL to Table ID mapping

url.to.table-id.matching.pattern matches the full URL by using a Java regular expression and generates a table ID based on the number of capturing groups. For example, if the video URL is rtmp://server/video/shop001/camera_01, you can use the following configuration to map it to shop001.camera_01:

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

The regular expression must contain exactly 1 to 3 capturing groups, and the captured results cannot be empty. If no regular expression is configured or a URL does not match, the connector uses default as the table ID.

Examples

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

No data is output after the job starts

Check the following items in order:

1. Whether the TaskManager can access the video URL.

2. Whether FFmpeg supports the protocol, container format, and codec.

3. Whether the video stream contains a decodable video track.

4. Whether sample-interval is set too large.

5. Whether the RTSP service requires ffmpeg.properties.rtsp_transport: tcp.

Why does the job not resume reading from the interruption point after a recovery?

Live streams have no stable, recoverable media position. A checkpoint can only restore the assignment of URL splits. Frames during a failure may not be captured, and duplicate records may be generated after reconnection. Therefore, the video stream connector does not support resuming from the interruption point and does not provide end-to-end exactly-once guarantees.

How do I reduce CPU and memory consumption?

• Increase sample-interval to reduce the number of images or the video clipping frequency.

• Use KEYFRAME_EXTRACT when only single-frame analysis is required.

• Decrease record-queue-capacity to limit the number of buffered media records.

• Distribute multiple URLs across an appropriate source parallelism so that a single source reader does not handle too many video streams at the same time.

• We recommend that you set ffmpeg-video.properties.threads to 1 to reduce CPU usage.