pyflink.dataframe.read_json#

read_json(path: str, *, schema: Optional[Dict[str, DataType]] = None, computed_columns: Optional[Dict[str, str]] = None, watermark: Optional[Tuple[str, str]] = None, fail_on_missing_field: bool = False, ignore_parse_errors: bool = False, timestamp_format: Literal['SQL', 'ISO-8601'] = 'SQL', monitor_interval: Optional[str] = None, path_regex_pattern: Optional[str] = None) → pyflink.dataframe.dataframe.DataFrame[source]#

Read JSON file(s) into a DataFrame.

Uses Flink’s FileSystem Connector with JSON Format under the hood.

Parameters

path – Path to a JSON file or directory.
schema – Dict of {column_name: DataType} specifying the schema. This parameter is required.
computed_columns – Optional dict of computed column expressions keyed by computed column name.
watermark – Optional tuple of (rowtime_column, watermark_expression).
fail_on_missing_field – Whether to fail if a field is missing when parsing JSON. Default is False.
ignore_parse_errors – Whether to skip fields and rows with parse errors instead of failing. Default is False.
timestamp_format – Timestamp format standard. One of "SQL" or "ISO-8601". Default is "SQL".
monitor_interval – Optional interval for continuously monitoring the directory for new files (e.g., ’10s’, ‘1min’).
path_regex_pattern – Optional regex pattern to filter files.

Returns

A new DataFrame.

Example:

>>> import pyflink.dataframe as pf
>>>
>>> df = pf.read_json("/path/to/users.json", schema={
...     "id": pf.DataType.int64(),
...     "name": pf.DataType.string(),
... })
>>>
>>> # Read with JSON parse options
>>> df = pf.read_json(
...     "/path/to/users.json",
...     schema={
...         "id": pf.DataType.int64(),
...         "name": pf.DataType.string(),
...     },
...     ignore_parse_errors=True,
...     timestamp_format="ISO-8601",
... )
>>>
>>> # Add a computed event-time column and watermark
>>> events = pf.read_json(
...     "/path/to/events.json",
...     schema={
...         "id": pf.DataType.int64(),
...         "payload": pf.DataType.string(),
...         "ts_millis": pf.DataType.int64(),
...     },
...     computed_columns={
...         "event_time": "TO_TIMESTAMP_LTZ(ts_millis, 3)",
...     },
...     watermark=("event_time", "event_time - INTERVAL '5' SECOND"),
...     timestamp_format="ISO-8601",
... )