pyflink.dataframe.DataFrame.write_json#

DataFrame.write_json(path: str, *, mode: str = 'overwrite', timestamp_format: Literal['SQL', 'ISO-8601'] = 'SQL', map_null_key_mode: Literal['FAIL', 'DROP', 'LITERAL'] = 'FAIL', map_null_key_literal: str = 'null', encode_decimal_as_plain_number: bool = False, write_null_properties: bool = True, rolling_policy_file_size: str = '128mb', rolling_policy_rollover_interval: str = '30min', rolling_policy_inactivity_interval: str = '30min', rolling_policy_check_interval: str = '1min', partition_commit_trigger: str = 'process-time', partition_commit_delay: str = '0s', partition_commit_watermark_time_zone: str = 'UTC', partition_commit_policy_kind: Optional[str] = None, partition_commit_policy_class: Optional[str] = None, partition_commit_policy_class_parameters: Optional[str] = None, partition_commit_success_file_name: str = '_SUCCESS', partition_time_extractor_kind: str = 'default', partition_time_extractor_class: Optional[str] = None, partition_time_extractor_timestamp_formatter: Optional[str] = None, partition_time_extractor_timestamp_pattern: Optional[str] = None, shuffle_by_partition_enable: bool = False, auto_compaction: bool = False, compaction_file_size: Optional[str] = None, compaction_parallelism: Optional[int] = None, sink_parallelism: Optional[int] = None, partition_default_name: str = '__DEFAULT_PARTITION__', statement_set: Optional[StatementSet] = None) → None[source]#

Write the DataFrame to JSON file(s) at the given path.

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

Parameters

path – Output path for the JSON file(s).
mode – Write mode. ‘overwrite’ to overwrite existing data (default), ‘append’ to append to existing data.
timestamp_format – Timestamp format standard. One of "SQL" or "ISO-8601". Default is "SQL".
map_null_key_mode – Handling mode for null map keys. One of "FAIL", "DROP", or "LITERAL".
map_null_key_literal – Literal used for null map keys when map_null_key_mode="LITERAL".
encode_decimal_as_plain_number – Whether to encode decimals as plain numbers instead of scientific notation.
write_null_properties – Whether to write null properties when encoding JSON. Default is True.
sink_parallelism – Custom parallelism for the sink.

Example:

>>> import pyflink.dataframe as pf
>>>
>>> orders = pf.from_records(
...     [(1, "book", 12.5), (2, "pen", 1.5)],
...     schema=["order_id", "item", "amount"],
... )
>>> orders.write_json("/path/to/orders-json")
>>>
>>> # Append with JSON format options
>>> orders.write_json(
...     "/path/to/orders-json",
...     mode="append",
...     timestamp_format="ISO-8601",
...     write_null_properties=False,
... )
>>>
>>> # Customize map null-key serialization
>>> orders.write_json(
...     "/path/to/orders-json",
...     map_null_key_mode="LITERAL",
...     map_null_key_literal="__NULL__",
... )