pyflink.dataframe.read_mns#

read_mns(endpoint: str, *, schema: Optional[Dict[str, DataType]] = None, computed_columns: Optional[Dict[str, str]] = None, watermark: Optional[Tuple[str, str]] = None, region: str, queue_name: str, access_key_id: str, access_key_secret: str, format: str = 'json', format_options: Optional[Dict[str, Any]] = None, batch_size: int = 1, polling_wait_time: str = '10s', message_type: Literal['RAW', 'OSS'] = 'RAW', delete_max_retries: int = 3, start_time_ms: Optional[int] = None, rate_limit_records_per_second: Optional[float] = None, extra_options: Optional[Dict[str, Any]] = None) → pyflink.dataframe.dataframe.DataFrame[source]#

Read an MNS (Alibaba Cloud Message Service) queue into a DataFrame.

Reading from MNS requires checkpointing to be enabled.

Parameters

endpoint – MNS queue endpoint.
schema – Dict of {column_name: DataType} specifying the source schema. Required.
computed_columns – Optional dict of computed column expressions keyed by computed column name.
watermark – Optional tuple of (rowtime_column, watermark_expression).
region – Alibaba Cloud region, for example "cn-hangzhou".
queue_name – MNS queue name.
access_key_id – Alibaba Cloud AccessKey ID.
access_key_secret – Alibaba Cloud AccessKey Secret.
format – Message payload format. Defaults to "json".
format_options – Additional options for format. Keys are unprefixed, for example {"ignore-parse-errors": True} produces json.ignore-parse-errors = true.
batch_size – Number of messages fetched per poll. Must be in [1, 16].
polling_wait_time – Long-poll wait time. Must be a non-empty duration string. Format and range are validated by the connector.
message_type – "RAW" for normal queue messages or "OSS" for OSS message references. OSS messages require format="json".
delete_max_retries – Maximum delete retries after a message is read. Must be non-negative.
start_time_ms – Optional start time in epoch milliseconds. Must be non-negative when set.
rate_limit_records_per_second – Optional source-side rate limit. Must be positive when set.
extra_options – Additional connector options forwarded through to the underlying MNS connector. This is for options not exposed as named parameters of read_mns. Keys matching options exposed as named parameters are rejected; use the named parameters instead. Parser options for format should be passed through format_options. "connector" is reserved and must not be supplied.

Returns

A DataFrame backed by the MNS source.

Raises

ValueError – If schema is missing, an MNS option is out of range, message_type is invalid, message_type="OSS" is combined with a non-JSON format, or extra_options contains "connector", a named option key, or a format option key.
TypeError – If schema, format_options, extra_options, or numeric MNS options have invalid types.

Examples

Basic JSON messages:

import pyflink.dataframe as pf

# MNS reads require checkpointing.
pf.config.set("execution.checkpointing.interval", "30s")

df = pf.read_mns(
    "https://1234567890.mns.cn-hangzhou.aliyuncs.com",
    region="cn-hangzhou",
    queue_name="orders",
    access_key_id=ALIYUN_AK_ID,
    access_key_secret=ALIYUN_AK_SECRET,
    schema={
        "order_id": pf.DataType.string(),
        "amount": pf.DataType.float64(),
        "ts_ms": pf.DataType.int64(),
    },
)

OSS message references:

df = pf.read_mns(
    "https://1234567890.mns.cn-hangzhou.aliyuncs.com",
    region="cn-hangzhou",
    queue_name="oss-events",
    access_key_id=ALIYUN_AK_ID,
    access_key_secret=ALIYUN_AK_SECRET,
    schema={
        "eventTime": pf.DataType.string(),
        "region": pf.DataType.string(),
        "ossBucketName": pf.DataType.string(),
        "ossObjectKey": pf.DataType.string(),
        "ossObjectSize": pf.DataType.int64(),
    },
    message_type="OSS",
    format="json",
)

JSON parser options:

df = pf.read_mns(
    "https://1234567890.mns.cn-hangzhou.aliyuncs.com",
    region="cn-hangzhou",
    queue_name="orders",
    access_key_id=ALIYUN_AK_ID,
    access_key_secret=ALIYUN_AK_SECRET,
    schema={"payload": pf.DataType.string()},
    format_options={
        "ignore-parse-errors": True,
        "timestamp-format.standard": "ISO-8601",
    },
)

Computed event time and watermark:

events = pf.read_mns(
    "https://1234567890.mns.cn-hangzhou.aliyuncs.com",
    region="cn-hangzhou",
    queue_name="orders",
    access_key_id=ALIYUN_AK_ID,
    access_key_secret=ALIYUN_AK_SECRET,
    schema={
        "order_id": pf.DataType.string(),
        "ts_ms": pf.DataType.int64(),
    },
    computed_columns={
        "event_time": "TO_TIMESTAMP_LTZ(ts_ms, 3)",
    },
    watermark=("event_time", "event_time - INTERVAL '5' SECOND"),
)

Replay with source throttling:

replay = pf.read_mns(
    "https://1234567890.mns.cn-hangzhou.aliyuncs.com",
    region="cn-hangzhou",
    queue_name="orders",
    access_key_id=ALIYUN_AK_ID,
    access_key_secret=ALIYUN_AK_SECRET,
    schema={"order_id": pf.DataType.string()},
    start_time_ms=1767225600000,
    rate_limit_records_per_second=1000,
)