This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a6c87283904 [FLINK-40426][python] Add Parquet and JSON filesystem I/O
to DataFrame API (#29245)
a6c87283904 is described below
commit a6c872839042cc8b8baa8aabc80f6eaaa8b65b3d
Author: csurong <[email protected]>
AuthorDate: Thu Sep 24 15:36:19 2026 +0800
[FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API
(#29245)
---
.../docs/reference/pyflink.dataframe/io.rst | 205 +++++
flink-python/pyflink/dataframe/__init__.py | 4 +-
flink-python/pyflink/dataframe/dataframe.py | 290 ++++++-
flink-python/pyflink/dataframe/io.py | 285 ++++++-
.../pyflink/dataframe/tests/test_filesystem_io.py | 860 +++++++++++++++++++++
flink-python/pyflink/dataframe/tests/test_io.py | 1 +
6 files changed, 1635 insertions(+), 10 deletions(-)
diff --git a/flink-python/docs/reference/pyflink.dataframe/io.rst
b/flink-python/docs/reference/pyflink.dataframe/io.rst
index aab2b109f04..a5c34a926ab 100644
--- a/flink-python/docs/reference/pyflink.dataframe/io.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/io.rst
@@ -33,6 +33,8 @@ Readers
:toctree: api/
read_generic
+ read_json
+ read_parquet
Writers
-------
@@ -43,3 +45,206 @@ Writers
:toctree: api/
DataFrame.write_generic
+ DataFrame.write_json
+ DataFrame.write_parquet
+
+Filesystem sources and sinks
+----------------------------
+
+``read_json`` reads newline-delimited JSON; ``read_parquet`` reads Parquet
files.
+Both readers require an explicit schema and use the existing filesystem
connector.
+The connector and format JARs must be available to Flink. Parquet also requires
+Hadoop libraries on the classpath, which can be provided through
``HADOOP_CLASSPATH``.
+
+For example, convert JSON records to partitioned Parquet files::
+
+ import pyflink.dataframe as pf
+
+ pf.config.set("execution.runtime-mode", "batch")
+ events = pf.read_json(
+ "file:///tmp/events.json",
+ schema={"id": pf.DataType.int64(), "event": pf.DataType.string()},
+ ignore_parse_errors=True,
+ )
+ events.write_parquet(
+ "file:///tmp/events-parquet", compression="GZIP", partition_by="event"
+ )
+
+Partitioned data
+~~~~~~~~~~~~~~~~
+
+``partition_by`` accepts a column name or a non-empty list of column names.
The order
+determines the directory layout, such as ``event=login/day=2026-01-01/``. The
filesystem
+connector stores partition values in these paths and excludes the columns from
the file
+records. Readers require the partition columns in both ``schema`` and
``partition_by``
+to restore their values. Null and empty string partition values share the
default
+partition directory. Its name can be configured using
+``connector_options={"partition.default-name": "..."}``. The current
filesystem connector
+cannot read these default partitions: planning fails while reconstructing
their paths.
+Use non-null, non-empty partition values when the output needs to be read back.
+
+The writers default to ``mode="append"`` in both batch and streaming
execution, adding files
+without replacing existing data. Explicit ``mode="overwrite"`` is supported
only in batch
+execution; the filesystem connector rejects it in streaming execution. For a
non-partitioned
+sink, overwrite replaces the existing output data. For a partitioned sink,
only partitions
+present in the input are overwritten; other partitions are retained. Configure
+``execution.runtime-mode`` before creating the DataFrame environment. Writes
wait
+for completion with local or MiniCluster execution, just like
``write_generic``.
+An unbounded local write therefore continues waiting until the job terminates.
+
+Options and defaults
+~~~~~~~~~~~~~~~~~~~~
+
+All four helpers accept ``connector_options`` for filesystem options and
``format_options``
+for JSON or Parquet options. Both dictionaries require string keys and values.
Format keys
+may include or omit the current format prefix, such as ``compression`` or
+``parquet.compression``. Specifying both forms of the same key is rejected.
+``connector_options`` cannot contain ``connector``, ``path`` or ``format``;
use the dedicated
+APIs and parameters to select these. Format options belong in
``format_options`` and are
+prefixed with the format name. The bare ``connector`` key is rejected in
either dictionary.
+
+Optional connector and format parameters use ``None`` to mean unspecified.
Dictionary values
+are merged with explicitly supplied parameters. Options absent from both are
omitted, so the
+connector or format factory applies its defaults. Supplying both forms is
allowed if their
+string values agree; conflicting values raise ``ValueError``.
+Values are compared as strings: ``"1min"`` and ``"60s"`` are different
settings for this
+check. The dictionaries supplied by the caller are not modified.
+
+For example, this uses GZIP rather than the default SNAPPY::
+
+ events.write_parquet(
+ "file:///tmp/output",
+ mode="append",
+ format_options={"compression": "GZIP"},
+ connector_options={"sink.parallelism": "2"},
+ )
+
+The current format default for Parquet compression is ``"SNAPPY"``. The
default file rolling
+threshold is ``"128mb"``, rollover and inactivity thresholds are ``"30min"``,
and the
+time-based rolling check interval is ``"1min"``. Partition commit defaults to
+``"process-time"`` with a ``"0s"`` delay and no policy. JSON defaults to SQL
timestamp
+formatting, does not ignore parse errors, and does not fail on missing fields.
Parquet's
+``utc-timezone`` option defaults to ``False``, using the JVM default time zone
for timestamp
+conversion. This option is independent of the session time zone and the
partition commit time zone.
+
+``write_json`` also exposes ``ignore_null_fields`` and
``decimal_as_plain_number``. The first
+omits fields with null values from JSON objects; it does not drop the entire
record. The second
+writes DECIMAL values as plain numbers instead of scientific notation, while
retaining JSON
+numeric values rather than strings. The format default for both options is
currently ``False``.
+
+The helpers check Python argument types and conflicting options. Connector-
and format-specific
+validation is delegated to the underlying implementations. Options without a
convenience
+parameter, including partition time extractors, custom commit policies and
Parquet tuning
+options, remain available through these dictionaries.
+
+Format-specific options
+~~~~~~~~~~~~~~~~~~~~~~~
+
+Parquet writes timestamps as INT96 by default. To use INT64 with a timestamp
logical type,
+set ``write.int64.timestamp`` through ``format_options``. The
``timestamp.time.unit`` option
+then selects ``millis``, ``micros`` (the default) or ``nanos``. Milliseconds
discard precision
+below a millisecond; choose the unit required by the consuming system. For
example::
+
+ timestamps = pf.sql("SELECT TIMESTAMP '2026-01-02 03:04:05.123456' AS ts")
+ timestamps.write_parquet(
+ "file:///tmp/timestamps",
+ utc_timezone=True,
+ format_options={
+ "write.int64.timestamp": "true",
+ "timestamp.time.unit": "micros",
+ },
+ )
+
+JSON null fields and Map entries with null keys have separate settings.
``ignore_null_fields``
+omits null fields from rows; it does not control null Map keys. The
``map-null-key.mode`` format
+option defaults to ``"FAIL"``. ``"DROP"`` removes entries with null keys,
while ``"LITERAL"``
+replaces each null key with the string configured by ``map-null-key.literal``.
For example::
+
+ records = pf.sql(
+ "SELECT CAST(NULL AS STRING) AS name, "
+ "MAP[CAST(NULL AS STRING), 1, 'known', 2] AS attributes"
+ )
+ records.write_json(
+ "file:///tmp/records",
+ ignore_null_fields=True,
+ format_options={
+ "map-null-key.mode": "LITERAL",
+ "map-null-key.literal": "missing",
+ },
+ )
+
+The JSON record omits ``name`` and contains ``{"attributes": {"missing": 1,
"known": 2}}``.
+
+Continuous reads and event time
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+When ``source.monitor-interval`` is unset, readers scan the configured paths
once and produce
+a bounded source. Setting a positive interval, such as ``"60s"``, through
``monitor_interval``
+or ``connector_options`` enables continuous discovery and requires streaming
execution.
+It does not change the environment's execution mode.
+Each file is identified by its path and processed once; discovery does not
tail files
+for appended records or re-read files changed in place.
+
+For partitioned sources, Flink discovers the partition directories during
planning and
+monitors those paths. It does not discover new partition directories. If no
partitions
+exist initially, the source remains empty even when ``monitor_interval`` is
set.
+
+``path_regex_pattern`` uses Java regular expressions and matches the entire
file path,
+excluding the URI scheme and authority. It is not a substring search or a
filename-only
+match. For example, use ``r".*/events[.]json"`` to select files named
``events.json``.
+Hidden files and directories are excluded.
+
+Both readers support ``computed_columns`` and ``watermark``, using the same
schema
+declarations as ``read_generic``. A watermark needs a TIMESTAMP or
TIMESTAMP_LTZ column
+with precision from 0 to 3, such as ``pf.DataType.timestamp(3)``.
+
+Run the following example in a fresh environment. When using
``set_table_environment``,
+configure the injected environment for streaming first::
+
+ import pyflink.dataframe as pf
+
+ pf.config.set("execution.runtime-mode", "streaming")
+ incoming_events = pf.read_json(
+ "file:///tmp/incoming",
+ schema={"id": pf.DataType.int64(), "ts_millis": pf.DataType.int64()},
+ monitor_interval="10s",
+ computed_columns={"event_time": "TO_TIMESTAMP_LTZ(ts_millis, 3)"},
+ watermark=("event_time", "event_time - INTERVAL '5' SECOND"),
+ )
+
+Partition shuffle and compaction
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Both writers expose ``sink_shuffle_by_partition`` to redistribute records by
dynamic partition
+fields before writing. This can reduce the number of files but may cause data
skew for heavily
+populated partitions. The connector default is currently ``False``.
+
+For streaming writes, ``auto_compaction=True`` enables automatic compaction
after checkpoints
+complete. Files remain invisible until compaction finishes. This reduces small
files at the cost
+of additional I/O. ``compaction_file_size`` sets the target size for compacted
files; it is not a
+hard size limit. If unspecified, the connector uses the rolling policy file
size.
+Automatic compaction is disabled by default.
+
+Streaming writes
+~~~~~~~~~~~~~~~~
+
+Rolling policy parameters apply to streaming writes, not batch writes. File
size and time
+settings are rolling thresholds, not hard limits on output file size or
visibility latency.
+Continuous writes need checkpointing to finish pending files. Without
automatic compaction,
+JSON also needs a rolling condition to close the file. Parquet and writers
with automatic
+compaction enabled roll on checkpoints as well as on size/time thresholds.
+Configure ``execution.checkpointing.interval`` through the environment
configuration before
+execution; these helpers do not enable checkpointing automatically.
+
+Streaming partition commit requires ``partition_by`` and a policy such as
``"success-file"``.
+The ``"partition-time"`` trigger also needs upstream watermarks and partition
time extraction
+settings in ``connector_options``. If the watermark is declared on a
TIMESTAMP_LTZ column,
+set ``sink.partition-commit.watermark-time-zone`` to the session time zone.
For example,
+with ``table.local-time-zone="Asia/Shanghai"``, pass
+``connector_options={"sink.partition-commit.watermark-time-zone":
"Asia/Shanghai"}``.
+Leaving this option at its UTC default with a non-UTC session time zone can
shift partition
+commit times by hours. For a TIMESTAMP watermark, keep the UTC default.
+
+A success marker is a partition commit notification; it does not prevent later
records from
+being written to the partition. The ``"metastore"`` policy is specific to Hive
tables and
+cannot be used with these filesystem helpers.
diff --git a/flink-python/pyflink/dataframe/__init__.py
b/flink-python/pyflink/dataframe/__init__.py
index bce542f48c2..94697f87029 100644
--- a/flink-python/pyflink/dataframe/__init__.py
+++ b/flink-python/pyflink/dataframe/__init__.py
@@ -54,7 +54,7 @@ from pyflink.dataframe.context import (
from pyflink.dataframe.dataframe import DataFrame, GroupedDataFrame, col, lit
from pyflink.dataframe.dataframe_config import config
from pyflink.dataframe.datatype import DataType
-from pyflink.dataframe.io import read_generic
+from pyflink.dataframe.io import read_generic, read_json, read_parquet
from pyflink.dataframe.sql import sql
from pyflink.dataframe.udf import udf
from pyflink.dataframe.udtf import udtf
@@ -74,6 +74,8 @@ __all__ = [
"from_table",
"range",
"read_generic",
+ "read_json",
+ "read_parquet",
"sql",
"config",
"set_table_environment",
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
index 0341b1c8f8c..119d48b06b6 100644
--- a/flink-python/pyflink/dataframe/dataframe.py
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -1504,6 +1504,278 @@ class DataFrame:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "append",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ compression: Optional[str] = None,
+ utc_timezone: Optional[bool] = None,
+ sink_parallelism: Optional[int] = None,
+ sink_shuffle_by_partition: Optional[bool] = None,
+ auto_compaction: Optional[bool] = None,
+ compaction_file_size: Optional[str] = None,
+ rolling_policy_file_size: Optional[str] = None,
+ rolling_policy_rollover_interval: Optional[str] = None,
+ rolling_policy_inactivity_interval: Optional[str] = None,
+ rolling_policy_check_interval: Optional[str] = None,
+ partition_commit_trigger: Optional[str] = None,
+ partition_commit_delay: Optional[str] = None,
+ partition_commit_policy_kind: Optional[str] = None,
+ connector_options: Optional[Dict[str, str]] = None,
+ format_options: Optional[Dict[str, str]] = None,
+ ) -> None:
+ """
+ Write Parquet files using Flink's filesystem connector.
+
+ The filesystem connector and Parquet format must be available to
Flink. The write
+ is submitted immediately and waits for completion for local or
MiniCluster execution.
+ Sink columns are derived from this DataFrame's schema. Writes default
to append in both
+ batch and streaming execution. Explicit overwrite requires batch
execution and is rejected
+ by the filesystem connector in streaming execution. Partitioned
overwrite replaces only
+ partitions present in the input, retaining other partitions.
+
+ Optional connector and format parameters use ``None`` to leave the
option unspecified.
+ If neither a parameter nor its dictionary option is set, the connector
or format factory
+ supplies the default.
+
+ Rolling policies apply to streaming sinks. Parquet also rolls files on
checkpoints;
+ continuous writes require checkpointing to finish files. Partition
commit in streaming
+ requires ``partition_by`` and a commit policy. ``partition-time``
additionally requires
+ upstream watermarks and a partition time extractor, configured via
``connector_options``.
+ For a TIMESTAMP_LTZ watermark, set
``sink.partition-commit.watermark-time-zone`` in
+ ``connector_options`` to the session time zone; its default is UTC.
+
+ :param path: Output directory URI supported by Flink's filesystem
implementations.
+ :param mode: ``"append"`` (default) adds files; ``"overwrite"``
replaces existing data
+ in batch execution only.
+ :param partition_by: Partition column name or non-empty list of names
in directory order.
+ Values are stored in Hive-style partition paths rather than in the
Parquet records.
+ :param compression: Parquet compression codec. The format default is
currently
+ ``"SNAPPY"``.
+ :param utc_timezone: Use UTC for Parquet timestamp conversion. The
format default is
+ currently ``False``, which uses the JVM default time zone,
independently of the
+ session time zone.
+ :param sink_parallelism: Sink parallelism. The connector default is
the upstream
+ parallelism.
+ :param sink_shuffle_by_partition: Shuffle rows by dynamic partition
fields before writing.
+ This can reduce the number of files but may cause data skew. The
connector default
+ is currently ``False``.
+ :param auto_compaction: Automatically compact files in streaming
execution after
+ checkpoints complete. Files remain invisible until compaction
finishes. The connector
+ default is currently ``False``.
+ :param compaction_file_size: Target file size for automatic
compaction, for example
+ ``"128mb"``. The connector default is the rolling policy file size.
+ :param rolling_policy_file_size: Part file size threshold for rolling,
not a hard upper
+ bound. The connector default is currently ``"128mb"``.
+ :param rolling_policy_rollover_interval: Part file open-time
threshold. The connector
+ default is currently ``"30min"``.
+ :param rolling_policy_inactivity_interval: Part file inactivity
threshold. The connector
+ default is currently ``"30min"``.
+ :param rolling_policy_check_interval: Interval for checking time-based
rolling policies.
+ The connector default is currently ``"1min"``.
+ :param partition_commit_trigger: Partition commit trigger:
``"process-time"`` or
+ ``"partition-time"``. The connector default is currently
``"process-time"``.
+ :param partition_commit_delay: Delay before committing a partition.
The connector
+ default is currently ``"0s"``.
+ :param partition_commit_policy_kind: Optional comma-separated
policies, such as
+ ``"success-file"`` or ``"custom"``. The ``metastore`` policy
requires a Hive table.
+ :param connector_options: Additional filesystem options with string
keys and values.
+ The ``connector``, ``path`` and ``format`` keys are reserved.
Format options belong in
+ ``format_options``. Explicit parameter and dictionary values must
agree when both
+ are set. Unspecified options are left to the connector factory.
+ :param format_options: Parquet options with string values, with or
without the
+ ``parquet.`` prefix. Duplicate normalized keys are rejected.
``None`` parameters
+ leave dictionary values unchanged; conflicting explicit values are
rejected.
+ For INT64 timestamp encoding, use ``{"write.int64.timestamp":
"true",
+ "timestamp.time.unit": "micros"}``. The default encoding is INT96.
+ :raises TypeError: If an argument has an invalid type.
+ :raises ValueError: If the path is empty, the write mode is
unsupported, the partition
+ specification is empty or contains empty or duplicate names, or
options conflict
+ or contain reserved, empty or duplicate keys.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> _ = pf.config.set("execution.runtime-mode", "batch")
+ >>> events = pf.from_records([(1, "login")], schema=["id",
"event"])
+ >>> events.write_parquet("file:///tmp/events", compression="GZIP")
+
+ .. versionadded:: 2.4.0
+ """
+ from pyflink.dataframe.io import (
+ _boolean_option,
+ _build_filesystem_options,
+ _parallelism_option,
+ )
+
+ options = _build_filesystem_options(
+ path,
+ "parquet",
+ connector_parameters={
+ "sink.parallelism": _parallelism_option(sink_parallelism),
+ "sink.shuffle-by-partition.enable": _boolean_option(
+ sink_shuffle_by_partition, "sink_shuffle_by_partition"),
+ "auto-compaction": _boolean_option(auto_compaction,
"auto_compaction"),
+ "compaction.file-size": compaction_file_size,
+ "sink.rolling-policy.file-size": rolling_policy_file_size,
+ "sink.rolling-policy.rollover-interval":
rolling_policy_rollover_interval,
+ "sink.rolling-policy.inactivity-interval":
rolling_policy_inactivity_interval,
+ "sink.rolling-policy.check-interval":
rolling_policy_check_interval,
+ "sink.partition-commit.trigger": partition_commit_trigger,
+ "sink.partition-commit.delay": partition_commit_delay,
+ "sink.partition-commit.policy.kind":
partition_commit_policy_kind,
+ },
+ format_parameters={
+ "compression": compression,
+ "utc-timezone": _boolean_option(utc_timezone, "utc_timezone"),
+ },
+ extra_connector_options=connector_options,
+ extra_format_options=format_options,
+ )
+ self._write("filesystem", options, mode=mode,
partition_by=partition_by)
+
+ @PublicEvolving()
+ def write_json(
+ self,
+ path: str,
+ *,
+ mode: str = "append",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ timestamp_format: Optional[str] = None,
+ ignore_null_fields: Optional[bool] = None,
+ decimal_as_plain_number: Optional[bool] = None,
+ sink_parallelism: Optional[int] = None,
+ sink_shuffle_by_partition: Optional[bool] = None,
+ auto_compaction: Optional[bool] = None,
+ compaction_file_size: Optional[str] = None,
+ rolling_policy_file_size: Optional[str] = None,
+ rolling_policy_rollover_interval: Optional[str] = None,
+ rolling_policy_inactivity_interval: Optional[str] = None,
+ rolling_policy_check_interval: Optional[str] = None,
+ partition_commit_trigger: Optional[str] = None,
+ partition_commit_delay: Optional[str] = None,
+ partition_commit_policy_kind: Optional[str] = None,
+ connector_options: Optional[Dict[str, str]] = None,
+ format_options: Optional[Dict[str, str]] = None,
+ ) -> None:
+ """
+ Write newline-delimited JSON files using Flink's filesystem connector.
+
+ The filesystem connector and JSON format must be available to Flink.
The write
+ is submitted immediately and waits for completion for local or
MiniCluster execution.
+ Sink columns are derived from this DataFrame's schema. Writes default
to append in both
+ batch and streaming execution. Explicit overwrite requires batch
execution and is rejected
+ by the filesystem connector in streaming execution. Partitioned
overwrite replaces only
+ partitions present in the input, retaining other partitions.
+
+ Optional connector and format parameters use ``None`` to leave the
option unspecified.
+ If neither a parameter nor its dictionary option is set, the connector
or format factory
+ supplies the default.
+
+ Rolling policies apply to streaming sinks. Continuous writes require
both file rolling
+ and checkpointing to finish files. With automatic compaction, files
also roll on
+ checkpoints. Partition commit in streaming requires ``partition_by``
+ and a commit policy. ``partition-time`` additionally requires upstream
watermarks and a
+ partition time extractor, configured via ``connector_options``.
+ For a TIMESTAMP_LTZ watermark, set
``sink.partition-commit.watermark-time-zone`` in
+ ``connector_options`` to the session time zone; its default is UTC.
+
+ :param path: Output directory URI supported by Flink's filesystem
implementations.
+ :param mode: ``"append"`` (default) adds files; ``"overwrite"``
replaces existing data
+ in batch execution only.
+ :param partition_by: Partition column name or non-empty list of names
in directory order.
+ Values are stored in Hive-style partition paths rather than in the
JSON records.
+ :param timestamp_format: Timestamp representation, ``"SQL"`` or
``"ISO-8601"``.
+ The format default is currently ``"SQL"``.
+ :param ignore_null_fields: Omit fields with null values from JSON
objects. The format
+ default is currently ``False``. This does not control Map entries
with null keys;
+ configure ``map-null-key.mode`` through ``format_options`` for
those entries.
+ :param decimal_as_plain_number: Encode DECIMAL values as plain numbers
rather than
+ scientific notation, retaining JSON numeric values. The format
default is currently
+ ``False``.
+ :param sink_parallelism: Sink parallelism. The connector default is
the upstream
+ parallelism.
+ :param sink_shuffle_by_partition: Shuffle rows by dynamic partition
fields before writing.
+ This can reduce the number of files but may cause data skew. The
connector default
+ is currently ``False``.
+ :param auto_compaction: Automatically compact files in streaming
execution after
+ checkpoints complete. Files remain invisible until compaction
finishes. The connector
+ default is currently ``False``.
+ :param compaction_file_size: Target file size for automatic
compaction, for example
+ ``"128mb"``. The connector default is the rolling policy file size.
+ :param rolling_policy_file_size: Part file size threshold for rolling,
not a hard upper
+ bound. The connector default is currently ``"128mb"``.
+ :param rolling_policy_rollover_interval: Part file open-time
threshold. The connector
+ default is currently ``"30min"``.
+ :param rolling_policy_inactivity_interval: Part file inactivity
threshold. The connector
+ default is currently ``"30min"``.
+ :param rolling_policy_check_interval: Interval for checking time-based
rolling policies.
+ The connector default is currently ``"1min"``.
+ :param partition_commit_trigger: Partition commit trigger:
``"process-time"`` or
+ ``"partition-time"``. The connector default is currently
``"process-time"``.
+ :param partition_commit_delay: Delay before committing a partition.
The connector
+ default is currently ``"0s"``.
+ :param partition_commit_policy_kind: Optional comma-separated
policies, such as
+ ``"success-file"`` or ``"custom"``. The ``metastore`` policy
requires a Hive table.
+ :param connector_options: Additional filesystem options with string
keys and values.
+ The ``connector``, ``path`` and ``format`` keys are reserved.
Format options belong in
+ ``format_options``. Explicit parameter and dictionary values must
agree when both
+ are set. Unspecified options are left to the connector factory.
+ :param format_options: JSON format options with string values. Keys
may include or omit
+ the ``json.`` prefix, for example ``{"timestamp-format.standard":
"ISO-8601"}``.
+ ``None`` parameters leave dictionary values unchanged; conflicting
explicit values
+ are rejected.
+ :raises TypeError: If an argument has an invalid type.
+ :raises ValueError: If the path is empty, the write mode is
unsupported, the partition
+ specification is empty or contains empty or duplicate names, or
options conflict
+ or contain reserved, empty or duplicate keys.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> events = pf.from_records([(1, "login")], schema=["id",
"event"])
+ >>> events.write_json("file:///tmp/events")
+
+ .. versionadded:: 2.4.0
+ """
+ from pyflink.dataframe.io import (
+ _boolean_option,
+ _build_filesystem_options,
+ _parallelism_option,
+ )
+
+ options = _build_filesystem_options(
+ path,
+ "json",
+ connector_parameters={
+ "sink.parallelism": _parallelism_option(sink_parallelism),
+ "sink.shuffle-by-partition.enable": _boolean_option(
+ sink_shuffle_by_partition, "sink_shuffle_by_partition"),
+ "auto-compaction": _boolean_option(auto_compaction,
"auto_compaction"),
+ "compaction.file-size": compaction_file_size,
+ "sink.rolling-policy.file-size": rolling_policy_file_size,
+ "sink.rolling-policy.rollover-interval":
rolling_policy_rollover_interval,
+ "sink.rolling-policy.inactivity-interval":
rolling_policy_inactivity_interval,
+ "sink.rolling-policy.check-interval":
rolling_policy_check_interval,
+ "sink.partition-commit.trigger": partition_commit_trigger,
+ "sink.partition-commit.delay": partition_commit_delay,
+ "sink.partition-commit.policy.kind":
partition_commit_policy_kind,
+ },
+ format_parameters={
+ "timestamp-format.standard": timestamp_format,
+ "encode.ignore-null-fields": _boolean_option(
+ ignore_null_fields, "ignore_null_fields"),
+ "encode.decimal-as-plain-number": _boolean_option(
+ decimal_as_plain_number, "decimal_as_plain_number"),
+ },
+ extra_connector_options=connector_options,
+ extra_format_options=format_options,
+ )
+ self._write("filesystem", options, mode=mode,
partition_by=partition_by)
+
@PublicEvolving()
def write_generic(self, connector: str, *, options: Dict[str, str]) ->
None:
"""
@@ -1533,10 +1805,24 @@ class DataFrame:
.. versionadded:: 2.4.0
"""
+ self._write(connector, options)
+
+ def _write(
+ self,
+ connector: str,
+ options: Dict[str, str],
+ *,
+ mode: str = "append",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ ) -> None:
from pyflink.dataframe.io import _build_generic_descriptor
- descriptor = _build_generic_descriptor(connector, options)
- result = self._table.execute_insert(descriptor)
+ if not isinstance(mode, str):
+ raise TypeError("mode must be a string")
+ if mode not in ("append", "overwrite"):
+ raise ValueError("mode must be 'append' or 'overwrite'")
+ descriptor = _build_generic_descriptor(connector, options,
partition_by=partition_by)
+ result = self._table.execute_insert(descriptor, overwrite=mode ==
"overwrite")
execution_target = self._table._t_env.get_config().get(
"execution.target", None
)
diff --git a/flink-python/pyflink/dataframe/io.py
b/flink-python/pyflink/dataframe/io.py
index 05a7c0aedcd..0f629674690 100644
--- a/flink-python/pyflink/dataframe/io.py
+++ b/flink-python/pyflink/dataframe/io.py
@@ -16,15 +16,261 @@
# limitations under the License.
################################################################################
-from typing import Dict, Optional, Tuple
+from typing import Dict, List, Optional, Tuple, Union
from pyflink.dataframe.context import get_or_create_table_environment
-from pyflink.dataframe.dataframe import DataFrame
+from pyflink.dataframe.dataframe import DataFrame, _normalize_subset
from pyflink.dataframe.datatype import DataType
from pyflink.table import Schema, TableDescriptor
from pyflink.util.api_stability_decorators import PublicEvolving
-__all__ = ["read_generic"]
+__all__ = ["read_generic", "read_json", "read_parquet"]
+
+
+def _build_filesystem_options(
+ path: str,
+ file_format: str,
+ *,
+ connector_parameters: Optional[Dict[str, Optional[str]]] = None,
+ format_parameters: Optional[Dict[str, Optional[str]]] = None,
+ extra_connector_options: Optional[Dict[str, str]] = None,
+ extra_format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ if not isinstance(path, str):
+ raise TypeError("path must be a string")
+ if not path:
+ raise ValueError("path must not be empty")
+
+ result = {"path": path, "format": file_format}
+ if extra_connector_options is not None:
+ _validate_options(extra_connector_options)
+ for key in extra_connector_options:
+ if key in ("path", "format"):
+ raise ValueError(f"{key!r} must not be specified in
connector_options")
+ if key.startswith(file_format + "."):
+ raise ValueError(f"format option {key!r} must be specified in
format_options")
+ _merge_options(result, extra_connector_options)
+ if connector_parameters is not None:
+ _merge_options(result, {
+ key: value for key, value in connector_parameters.items() if value
is not None
+ })
+
+ normalized_format_options: Dict[str, str] = {}
+ if extra_format_options is not None:
+ _validate_options(extra_format_options)
+ for key, value in extra_format_options.items():
+ option = key if key.startswith(file_format + ".") else file_format
+ "." + key
+ if option in normalized_format_options:
+ raise ValueError(f"duplicate format option: {option!r}")
+ normalized_format_options[option] = value
+ if format_parameters is not None:
+ _merge_options(normalized_format_options, {
+ file_format + "." + key: value
+ for key, value in format_parameters.items() if value is not None
+ })
+ _merge_options(result, normalized_format_options)
+ return result
+
+
+def _merge_options(target: Dict[str, str], options: Dict[str, str]) -> None:
+ _validate_options(options)
+ for key, value in options.items():
+ if key in target and target[key] != value:
+ raise ValueError(f"conflicting values for option {key!r}")
+ target[key] = value
+
+
+def _boolean_option(value: Optional[bool], name: str) -> Optional[str]:
+ if value is None:
+ return None
+ if not isinstance(value, bool):
+ raise TypeError(f"{name} must be a bool or None")
+ return str(value).lower()
+
+
+def _parallelism_option(value: Optional[int]) -> Optional[str]:
+ if value is None:
+ return None
+ if isinstance(value, bool) or not isinstance(value, int):
+ raise TypeError("sink_parallelism must be an int or None")
+ return str(value)
+
+
+@PublicEvolving()
+def read_parquet(
+ path: str,
+ *,
+ schema: Dict[str, DataType],
+ partition_by: Optional[Union[str, List[str]]] = None,
+ monitor_interval: Optional[str] = None,
+ path_regex_pattern: Optional[str] = None,
+ utc_timezone: Optional[bool] = None,
+ connector_options: Optional[Dict[str, str]] = None,
+ format_options: Optional[Dict[str, str]] = None,
+ computed_columns: Optional[Dict[str, str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Read Parquet files using Flink's filesystem connector.
+
+ The filesystem connector and Parquet format must be available to Flink. By
default,
+ the source reads the existing files once. Setting ``monitor_interval``
creates an
+ unbounded source that discovers new files; it requires streaming
execution. Files are
+ identified by path and processed once, rather than tailed for appended
data.
+ With ``partition_by``, monitoring is limited to the partitions discovered
during planning;
+ new partition directories are not discovered, and an initially empty
partitioned source
+ remains empty.
+ The current filesystem connector cannot read default partitions
representing null or empty
+ partition values.
+
+ :param path: File or directory URI supported by Flink's filesystem
implementations.
+ :param schema: Non-empty mapping of physical column names to DataFrame
data types.
+ :param partition_by: Partition column name or non-empty list of names in
directory order.
+ These columns must be declared in ``schema`` and are read from the
partition paths.
+ :param monitor_interval: Positive file discovery interval, for example
``"60s"``.
+ If neither this parameter nor ``source.monitor-interval`` in
``connector_options``
+ is set, the source performs a bounded scan.
+ :param path_regex_pattern: Java regular expression matched against the
entire file path
+ (excluding the URI scheme and authority), for example
``".*[.]parquet"``. This is not
+ a substring or filename-only match. Hidden files are excluded.
+ :param utc_timezone: Use UTC for Parquet timestamp conversion. The format
default is
+ currently ``False``, which uses the JVM default time zone,
independently of the
+ session time zone.
+ :param connector_options: Additional filesystem options with string keys
and values.
+ The ``connector``, ``path`` and ``format`` keys are reserved. Format
options belong in
+ ``format_options``. Explicit parameter and dictionary values must
agree when both are set.
+ :param format_options: Parquet options with string values, with or without
the ``parquet.``
+ prefix. Duplicate normalized keys are rejected. ``None`` parameters
leave dictionary
+ values unchanged; conflicting explicit values are rejected.
Unspecified options use
+ the connector or format factory defaults.
+ :param computed_columns: Optional SQL expressions keyed by computed column
name.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
The column must
+ have a TIMESTAMP or TIMESTAMP_LTZ type with precision from 0 to 3.
+ :return: A DataFrame backed by the Parquet source.
+ :raises TypeError: If an argument has an invalid type.
+ :raises ValueError: If the path or schema is empty, the partition
specification is
+ empty or contains empty or duplicate names, or options conflict or
contain
+ reserved, empty or duplicate keys.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> events = pf.read_parquet(
+ ... "file:///tmp/events",
+ ... schema={"id": pf.DataType.int64(), "name":
pf.DataType.string()},
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ options = _build_filesystem_options(
+ path,
+ "parquet",
+ connector_parameters={
+ "source.monitor-interval": monitor_interval,
+ "source.path.regex-pattern": path_regex_pattern,
+ },
+ format_parameters={"utc-timezone": _boolean_option(utc_timezone,
"utc_timezone")},
+ extra_connector_options=connector_options,
+ extra_format_options=format_options,
+ )
+ return _read(
+ "filesystem", schema=schema, options=options,
+ computed_columns=computed_columns, watermark=watermark,
partition_by=partition_by,
+ )
+
+
+@PublicEvolving()
+def read_json(
+ path: str,
+ *,
+ schema: Dict[str, DataType],
+ partition_by: Optional[Union[str, List[str]]] = None,
+ monitor_interval: Optional[str] = None,
+ path_regex_pattern: Optional[str] = None,
+ ignore_parse_errors: Optional[bool] = None,
+ fail_on_missing_field: Optional[bool] = None,
+ timestamp_format: Optional[str] = None,
+ connector_options: Optional[Dict[str, str]] = None,
+ format_options: Optional[Dict[str, str]] = None,
+ computed_columns: Optional[Dict[str, str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Read newline-delimited JSON files using Flink's filesystem connector.
+
+ The filesystem connector and JSON format must be available to Flink. By
default,
+ the source reads the existing files once. Setting ``monitor_interval``
creates an
+ unbounded source that discovers new files; it requires streaming
execution. Files are
+ identified by path and processed once, rather than tailed for appended
data.
+ With ``partition_by``, monitoring is limited to the partitions discovered
during planning;
+ new partition directories are not discovered, and an initially empty
partitioned source
+ remains empty.
+ The current filesystem connector cannot read default partitions
representing null or empty
+ partition values.
+
+ :param path: File or directory URI supported by Flink's filesystem
implementations.
+ :param schema: Non-empty mapping of physical column names to DataFrame
data types.
+ :param partition_by: Partition column name or non-empty list of names in
directory order.
+ These columns must be declared in ``schema`` and are read from the
partition paths.
+ :param monitor_interval: Positive file discovery interval, for example
``"60s"``.
+ If neither this parameter nor ``source.monitor-interval`` in
``connector_options``
+ is set, the source performs a bounded scan.
+ :param path_regex_pattern: Java regular expression matched against the
entire file path
+ (excluding the URI scheme and authority), for example ``".*[.]json"``.
This is not
+ a substring or filename-only match. Hidden files are excluded.
+ :param ignore_parse_errors: Skip malformed fields or rows instead of
failing. Invalid fields
+ are set to null where possible. The format default is currently
``False``.
+ :param fail_on_missing_field: Fail on missing JSON fields. The format
default is currently
+ ``False``. Cannot be enabled together with ``ignore_parse_errors``.
+ :param timestamp_format: Timestamp representation, ``"SQL"`` or
``"ISO-8601"``.
+ The format default is currently ``"SQL"``.
+ :param connector_options: Additional filesystem options with string keys
and values.
+ The ``connector``, ``path`` and ``format`` keys are reserved. Format
options belong in
+ ``format_options``. Explicit parameter and dictionary values must
agree when both are set.
+ :param format_options: JSON format options with string values. Keys may
include or omit
+ the ``json.`` prefix, for example ``{"ignore-parse-errors": "true"}``.
``None`` parameters
+ leave dictionary values unchanged; conflicting explicit values are
rejected. Unspecified
+ options use the connector or format factory defaults.
+ :param computed_columns: Optional SQL expressions keyed by computed column
name.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
The column must
+ have a TIMESTAMP or TIMESTAMP_LTZ type with precision from 0 to 3.
+ :return: A DataFrame backed by the JSON source.
+ :raises TypeError: If an argument has an invalid type.
+ :raises ValueError: If the path or schema is empty, the partition
specification is
+ empty or contains empty or duplicate names, or options conflict or
contain
+ reserved, empty or duplicate keys.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> events = pf.read_json(
+ ... "file:///tmp/events.json",
+ ... schema={"id": pf.DataType.int64()},
+ ... format_options={"ignore-parse-errors": "true"},
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ options = _build_filesystem_options(
+ path,
+ "json",
+ connector_parameters={
+ "source.monitor-interval": monitor_interval,
+ "source.path.regex-pattern": path_regex_pattern,
+ },
+ format_parameters={
+ "ignore-parse-errors": _boolean_option(ignore_parse_errors,
"ignore_parse_errors"),
+ "fail-on-missing-field": _boolean_option(
+ fail_on_missing_field, "fail_on_missing_field"),
+ "timestamp-format.standard": timestamp_format,
+ },
+ extra_connector_options=connector_options,
+ extra_format_options=format_options,
+ )
+ return _read(
+ "filesystem", schema=schema, options=options,
+ computed_columns=computed_columns, watermark=watermark,
partition_by=partition_by,
+ )
def _validate_connector(connector: str) -> None:
@@ -128,18 +374,43 @@ def _build_generic_descriptor(
connector: str,
options: Dict[str, str],
schema: Optional[Schema] = None,
+ partition_by: Optional[Union[str, List[str]]] = None,
) -> TableDescriptor:
_validate_connector(connector)
_validate_options(options)
+ partition_keys = _normalize_subset(partition_by, "partition_by") or []
+ if any(not key for key in partition_keys):
+ raise ValueError("partition_by column names must not be empty")
+ if len(set(partition_keys)) != len(partition_keys):
+ raise ValueError("partition_by column names must not be duplicated")
descriptor_builder = TableDescriptor.for_connector(connector)
if schema is not None:
descriptor_builder.schema(schema)
+ if partition_keys:
+ descriptor_builder.partitioned_by(*partition_keys)
for key, value in options.items():
descriptor_builder.option(key, value)
return descriptor_builder.build()
+def _read(
+ connector: str,
+ *,
+ schema: Dict[str, DataType],
+ options: Dict[str, str],
+ computed_columns: Optional[Dict[str, str]],
+ watermark: Optional[Tuple[str, str]],
+ partition_by: Optional[Union[str, List[str]]] = None,
+) -> DataFrame:
+ source_schema = _build_source_schema(
+ schema, computed_columns=computed_columns, watermark=watermark)
+ descriptor = _build_generic_descriptor(
+ connector, options, schema=source_schema, partition_by=partition_by)
+ table_environment = get_or_create_table_environment()
+ return DataFrame(table_environment.from_descriptor(descriptor))
+
+
@PublicEvolving()
def read_generic(
connector: str,
@@ -186,7 +457,7 @@ def read_generic(
.. versionadded:: 2.4.0
"""
- source_schema = _build_source_schema(schema, computed_columns, watermark)
- descriptor = _build_generic_descriptor(connector, options, source_schema)
- table_environment = get_or_create_table_environment()
- return DataFrame(table_environment.from_descriptor(descriptor))
+ return _read(
+ connector, schema=schema, options=options,
+ computed_columns=computed_columns, watermark=watermark,
+ )
diff --git a/flink-python/pyflink/dataframe/tests/test_filesystem_io.py
b/flink-python/pyflink/dataframe/tests/test_filesystem_io.py
new file mode 100644
index 00000000000..f1012ae9cd0
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_filesystem_io.py
@@ -0,0 +1,860 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import json
+import os
+import time
+import unittest
+from datetime import datetime
+from unittest.mock import patch
+from zoneinfo import ZoneInfo
+
+from py4j.protocol import Py4JJavaError
+
+import pyflink.dataframe as pf
+from pyflink.common import JobStatus
+from pyflink.dataframe import DataType
+from pyflink.dataframe.io import _build_filesystem_options
+from pyflink.datastream import RuntimeExecutionMode
+from pyflink.java_gateway import get_gateway
+from pyflink.table import EnvironmentSettings, StreamTableEnvironment,
TableDescriptor, TableResult
+from pyflink.testing.test_case_utils import PyFlinkDataFrameUTTestCase
+
+
+class FilesystemOptionsTests(unittest.TestCase):
+ def test_format_options_are_namespaced_without_mutating_input(self):
+ format_options = {
+ "ignore-parse-errors": "true",
+ "json.timestamp-format.standard": "ISO-8601",
+ }
+ original = dict(format_options)
+ options = _build_filesystem_options(
+ "/input", "json", extra_format_options=format_options
+ )
+ self.assertEqual(options, {
+ "path": "/input",
+ "format": "json",
+ "json.ignore-parse-errors": "true",
+ "json.timestamp-format.standard": "ISO-8601",
+ })
+ self.assertEqual(format_options, original)
+
+ def test_format_options_cannot_override_filesystem_options(self):
+ options = _build_filesystem_options(
+ "/input", "json", extra_format_options={"path": "/other",
"format": "parquet"}
+ )
+ self.assertEqual(options["path"], "/input")
+ self.assertEqual(options["format"], "json")
+
+ def test_rejects_invalid_format_options(self):
+ cases = [
+ ([], TypeError, "options must be a dict"),
+ ({1: "true"}, TypeError, "option keys must be strings"),
+ ({"": "true"}, ValueError, "option keys must not be empty"),
+ ({"ignore-parse-errors": True}, TypeError, "must have a string
value"),
+ ({"connector": "blackhole"}, ValueError, "connector argument"),
+ ({"ignore-parse-errors": "true", "json.ignore-parse-errors":
"false"},
+ ValueError, "duplicate format option"),
+ ]
+ for options, error, message in cases:
+ with self.subTest(options=options):
+ with self.assertRaisesRegex(error, message):
+ _build_filesystem_options(
+ "/input", "json", extra_format_options=options)
+
+ def test_connector_options_are_validated_without_mutating_input(self):
+ options = {"source.monitor-interval": "1s",
"source.report-statistics": "NONE"}
+ original = dict(options)
+ actual = _build_filesystem_options(
+ "/input", "json", connector_parameters={"source.monitor-interval":
"1s"},
+ extra_connector_options=options,
+ )
+ self.assertEqual(actual["source.report-statistics"], "NONE")
+ self.assertEqual(options, original)
+ for invalid in ({"path": "/other"}, {"format": "parquet"},
+ {"connector": "blackhole"},
{"json.ignore-parse-errors": "true"}):
+ with self.subTest(options=invalid):
+ with self.assertRaises(ValueError):
+ _build_filesystem_options(
+ "/input", "json", extra_connector_options=invalid)
+ for invalid in ([], {1: "value"}, {"source.monitor-interval": 1}):
+ with self.subTest(options=invalid):
+ with self.assertRaises(TypeError):
+ _build_filesystem_options(
+ "/input", "json", extra_connector_options=invalid)
+
+ def test_conflicting_explicit_options_are_rejected(self):
+ with self.assertRaisesRegex(ValueError, "conflicting
values.*source.monitor-interval"):
+ _build_filesystem_options(
+ "/input", "json",
connector_parameters={"source.monitor-interval": "1s"},
+ extra_connector_options={"source.monitor-interval": "2s"},
+ )
+ for prefix in ("", "parquet."):
+ with self.subTest(prefix=prefix):
+ options = {prefix + "compression": "GZIP"}
+ with self.assertRaisesRegex(ValueError, "conflicting
values.*parquet.compression"):
+ _build_filesystem_options(
+ "/input", "parquet", extra_format_options=options,
+ format_parameters={"compression": "SNAPPY"},
+ )
+ actual = _build_filesystem_options(
+ "/input", "parquet", extra_format_options=options,
+ format_parameters={"compression": "GZIP"},
+ )
+ self.assertEqual(actual["parquet.compression"], "GZIP")
+
+ def test_readers_reject_invalid_paths_and_option_types(self):
+ cases = [
+ ({"path": None}, TypeError, "path must be a string"),
+ ({"path": ""}, ValueError, "path must not be empty"),
+ ({"monitor_interval": 60}, TypeError, "must have a string value"),
+ ({"path_regex_pattern": 1}, TypeError, "must have a string value"),
+ ]
+ for reader in (pf.read_json, pf.read_parquet):
+ for overrides, error, message in cases:
+ with self.subTest(reader=reader.__name__, overrides=overrides):
+ arguments = {"path": "/input", "schema": {"id":
DataType.int64()}}
+ arguments.update(overrides)
+ with self.assertRaisesRegex(error, message):
+ reader(**arguments)
+
+ def test_writers_reject_invalid_arguments_before_execution(self):
+ dataframe = pf.DataFrame(None)
+ cases = [
+ ({"path": None}, TypeError, "path must be a string"),
+ ({"path": ""}, ValueError, "path must not be empty"),
+ ({"mode": None}, TypeError, "mode must be a string"),
+ ({"mode": "ignore"}, ValueError, "mode must be 'append' or
'overwrite'"),
+ ({"rolling_policy_file_size": 128}, TypeError, "must have a string
value"),
+ ({"rolling_policy_inactivity_interval": 1}, TypeError, "must have
a string value"),
+ ({"rolling_policy_check_interval": 5}, TypeError, "must have a
string value"),
+ ({"partition_commit_delay": 1}, TypeError, "must have a string
value"),
+ ({"sink_parallelism": True}, TypeError, "sink_parallelism must be
an int"),
+ ({"sink_parallelism": "1"}, TypeError, "sink_parallelism must be
an int"),
+ ({"sink_shuffle_by_partition": "true"}, TypeError,
+ "sink_shuffle_by_partition must be a bool"),
+ ({"auto_compaction": 1}, TypeError, "auto_compaction must be a
bool"),
+ ({"compaction_file_size": 128}, TypeError, "must have a string
value"),
+ ({"partition_by": [1]}, TypeError, "partition_by must be a
string"),
+ ({"partition_by": []}, ValueError, "partition_by must not be
empty"),
+ ({"partition_by": ""}, ValueError, "column names must not be
empty"),
+ ({"partition_by": ["id", "id"]}, ValueError, "must not be
duplicated"),
+ ]
+ for writer in (dataframe.write_json, dataframe.write_parquet):
+ for overrides, error, message in cases:
+ with self.subTest(writer=writer.__name__, overrides=overrides):
+ arguments = {"path": "/output"}
+ arguments.update(overrides)
+ with self.assertRaisesRegex(error, message):
+ writer(**arguments)
+ with self.assertRaisesRegex(TypeError, "must have a string value"):
+ dataframe.write_parquet("/output", compression=1)
+ with self.assertRaisesRegex(TypeError, "utc_timezone must be a bool"):
+ dataframe.write_parquet("/output", utc_timezone="true")
+ with self.assertRaisesRegex(TypeError, "ignore_parse_errors must be a
bool"):
+ pf.read_json("/input", schema={"id": DataType.int64()},
ignore_parse_errors=1)
+ for parameter in ("ignore_null_fields", "decimal_as_plain_number"):
+ with self.subTest(parameter=parameter):
+ with self.assertRaisesRegex(TypeError, parameter + " must be a
bool"):
+ dataframe.write_json("/output", **{parameter: "true"})
+
+
+class FilesystemIOTests(PyFlinkDataFrameUTTestCase):
+ _SCHEMA = {"id": DataType.int64(), "name": DataType.string()}
+
+ def test_readers_declare_computed_columns_and_watermarks(self):
+ for reader in (pf.read_json, pf.read_parquet):
+ with self.subTest(reader=reader.__name__):
+ dataframe = reader(
+ "/input", schema={"ts_millis": DataType.int64()},
+ computed_columns={"ts": "TO_TIMESTAMP_LTZ(ts_millis, 3)"},
+ watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ )
+ resolved = dataframe.to_table().get_resolved_schema()
+ self.assertEqual(list(resolved.get_column_names()),
["ts_millis", "ts"])
+ self.assertEqual(len(resolved.get_watermark_specs()), 1)
+ physical = reader(
+ "/input", schema={"ts": DataType.timestamp(3)},
+ watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ )
+ self.assertEqual(len(physical.to_table().get_resolved_schema()
+ .get_watermark_specs()), 1)
+
+ def test_readers_forward_connector_and_format_parameters(self):
+ cases = [
+ (pf.read_json, {"ignore_parse_errors": True,
"fail_on_missing_field": False,
+ "timestamp_format": "ISO-8601"},
+ {"json.ignore-parse-errors": "true",
"json.fail-on-missing-field": "false",
+ "json.timestamp-format.standard": "ISO-8601"}),
+ (pf.read_parquet, {"utc_timezone": True}, {"parquet.utc-timezone":
"true"}),
+ ]
+ for reader, parameters, expected in cases:
+ with self.subTest(reader=reader.__name__):
+ with patch.object(self.t_env, "from_descriptor",
+ wraps=self.t_env.from_descriptor) as
from_descriptor:
+ reader("/input", schema=self._SCHEMA, **parameters,
+ connector_options={"source.report-statistics":
"NONE"})
+ options = dict(from_descriptor.call_args.args[0].get_options())
+ self.assertEqual(options["source.report-statistics"], "NONE")
+ for key, value in expected.items():
+ self.assertEqual(options[key], value)
+
+ def test_invalid_partition_columns_are_rejected_by_the_table_api(self):
+ for reader in (pf.read_json, pf.read_parquet):
+ with self.subTest(reader=reader.__name__):
+ with self.assertRaisesRegex(Py4JJavaError, "missing"):
+ reader("/input", schema=self._SCHEMA,
partition_by="missing")
+ dataframe = pf.from_records([(1,)], schema=["id"])
+ for writer in (dataframe.write_json, dataframe.write_parquet):
+ with self.subTest(writer=writer.__name__):
+ with self.assertRaisesRegex(Py4JJavaError, "missing"):
+ writer("/output", partition_by="missing")
+
+ def
test_writers_accept_raw_options_without_overriding_them_with_defaults(self):
+ dataframe = pf.from_records([(1,)], schema=["id"])
+ with patch.object(dataframe._table, "execute_insert") as
execute_insert:
+ dataframe.write_parquet(
+ "/output", format_options={"compression": "GZIP"},
+ connector_options={"sink.rolling-policy.file-size": "64mb"},
+ )
+ options = dict(execute_insert.call_args.args[0].get_options())
+ self.assertEqual(options["parquet.compression"], "GZIP")
+ self.assertEqual(options["sink.rolling-policy.file-size"], "64mb")
+
+ def test_readers_build_filesystem_descriptors(self):
+ for file_format, reader in [("json", pf.read_json), ("parquet",
pf.read_parquet)]:
+ for continuous in (False, True):
+ with self.subTest(file_format=file_format,
continuous=continuous):
+ arguments = {}
+ expected = {"connector": "filesystem", "path": "/input",
+ "format": file_format}
+ if continuous:
+ arguments = {"monitor_interval": "60s",
"path_regex_pattern": ".*data.*"}
+ expected.update({"source.monitor-interval": "60s",
+ "source.path.regex-pattern":
".*data.*"})
+ with patch.object(self.t_env, "from_descriptor",
+ wraps=self.t_env.from_descriptor) as
from_descriptor:
+ dataframe = reader("/input", schema=self._SCHEMA,
**arguments)
+ descriptor = from_descriptor.call_args.args[0]
+ self.assertEqual(dict(descriptor.get_options()), expected)
+ self.assert_dataframe_schema(dataframe, ["id", "name"])
+
+ def test_json_reader_forwards_format_options(self):
+ with patch.object(self.t_env, "from_descriptor",
+ wraps=self.t_env.from_descriptor) as from_descriptor:
+ pf.read_json("/input", schema=self._SCHEMA,
+ format_options={"ignore-parse-errors": "true"})
+ options = from_descriptor.call_args.args[0].get_options()
+ self.assertEqual(options.get("json.ignore-parse-errors"), "true")
+
+ def test_writers_build_descriptors_and_apply_write_mode(self):
+ dataframe = pf.from_records([(1, "a")], schema=["id", "name"])
+ for file_format in ("json", "parquet"):
+ writer = getattr(dataframe, "write_" + file_format)
+ for mode in (None, "append", "overwrite"):
+ with self.subTest(file_format=file_format, mode=mode):
+ arguments = {} if mode is None else {"mode": mode}
+ with patch.object(dataframe._table, "execute_insert") as
execute_insert:
+ self.assertIsNone(writer("/output", **arguments))
+ descriptor = execute_insert.call_args.args[0]
+ expected = {
+ "connector": "filesystem", "path": "/output",
"format": file_format,
+ }
+ self.assertEqual(dict(descriptor.get_options()), expected)
+ self.assertIsNone(descriptor.get_schema())
+ self.assertEqual(execute_insert.call_args.kwargs,
+ {"overwrite": mode == "overwrite"})
+
+ def test_writers_forward_compaction_and_shuffle_options(self):
+ dataframe = pf.from_records([(1, "a")], schema=["id", "name"])
+ for writer in (dataframe.write_json, dataframe.write_parquet):
+ for enabled in (False, True):
+ with self.subTest(writer=writer.__name__, enabled=enabled):
+ raw_options = {
+ "sink.shuffle-by-partition.enable":
str(enabled).lower(),
+ "auto-compaction": str(enabled).lower(),
+ "compaction.file-size": "64mb",
+ }
+ with patch.object(dataframe._table, "execute_insert") as
execute_insert:
+ writer("/output", partition_by="name",
+ sink_shuffle_by_partition=enabled,
auto_compaction=enabled,
+ compaction_file_size="64mb",
connector_options=raw_options)
+ options =
dict(execute_insert.call_args.args[0].get_options())
+ for key, value in raw_options.items():
+ self.assertEqual(options[key], value)
+ self.assertNotIn("sink.rolling-policy.file-size", options)
+
+ def test_json_writer_forwards_encoding_options(self):
+ dataframe = pf.from_records([(1,)], schema=["id"])
+ for enabled in (False, True):
+ with self.subTest(enabled=enabled):
+ raw_options = {
+ "encode.ignore-null-fields": str(enabled).lower(),
+ "json.encode.decimal-as-plain-number":
str(enabled).lower(),
+ }
+ with patch.object(dataframe._table, "execute_insert") as
execute_insert:
+ dataframe.write_json(
+ "/output", ignore_null_fields=enabled,
decimal_as_plain_number=enabled,
+ format_options=raw_options,
+ )
+ options = dict(execute_insert.call_args.args[0].get_options())
+ self.assertEqual(options["json.encode.ignore-null-fields"],
str(enabled).lower())
+
self.assertEqual(options["json.encode.decimal-as-plain-number"],
+ str(enabled).lower())
+
+ def
test_writers_reject_conflicting_convenience_options_before_execution(self):
+ dataframe = pf.from_records([(1,)], schema=["id"])
+ cases = [
+ {"sink_shuffle_by_partition": True,
+ "connector_options": {"sink.shuffle-by-partition.enable":
"false"}},
+ {"auto_compaction": True, "connector_options": {"auto-compaction":
"false"}},
+ {"compaction_file_size": "64mb",
+ "connector_options": {"compaction.file-size": "128mb"}},
+ ]
+ for writer in (dataframe.write_json, dataframe.write_parquet):
+ for arguments in cases:
+ with self.subTest(writer=writer.__name__, arguments=arguments):
+ with patch.object(dataframe._table, "execute_insert") as
execute_insert:
+ with self.assertRaisesRegex(ValueError, "conflicting
values"):
+ writer("/output", **arguments)
+ execute_insert.assert_not_called()
+ for parameter, option in [("ignore_null_fields",
"encode.ignore-null-fields"),
+ ("decimal_as_plain_number",
"encode.decimal-as-plain-number")]:
+ with self.subTest(parameter=parameter):
+ with patch.object(dataframe._table, "execute_insert") as
execute_insert:
+ with self.assertRaisesRegex(ValueError, "conflicting
values"):
+ dataframe.write_json(
+ "/output", **{parameter: True},
format_options={option: "false"})
+ execute_insert.assert_not_called()
+
+ def test_writers_forward_custom_sink_options_and_wait_locally(self):
+ dataframe = pf.from_records([(1,)], schema=["id"])
+ for file_format in ("json", "parquet"):
+ for target in ("local", "minicluster", "remote"):
+ with self.subTest(file_format=file_format, target=target):
+ self.t_env.get_config().set("execution.target", target)
+ arguments = ({"compression": "GZIP"} if file_format ==
"parquet" else
+ {"timestamp_format": "ISO-8601"})
+ with patch.object(dataframe._table, "execute_insert") as
execute_insert:
+ getattr(dataframe, "write_" + file_format)(
+ "/output", mode="append",
rolling_policy_file_size="64mb",
+ rolling_policy_rollover_interval="10min",
+ rolling_policy_inactivity_interval="2min",
sink_parallelism=2,
+ rolling_policy_check_interval="1min",
+ partition_commit_trigger="partition-time",
partition_commit_delay="1h",
+ partition_commit_policy_kind="success-file",
**arguments)
+ options =
dict(execute_insert.call_args.args[0].get_options())
+ self.assertEqual(options["sink.rolling-policy.file-size"],
"64mb")
+
self.assertEqual(options["sink.rolling-policy.rollover-interval"], "10min")
+
self.assertEqual(options["sink.rolling-policy.inactivity-interval"], "2min")
+ self.assertEqual(options["sink.parallelism"], "2")
+
self.assertEqual(options["sink.rolling-policy.check-interval"], "1min")
+ self.assertEqual(options["sink.partition-commit.trigger"],
"partition-time")
+ self.assertEqual(options["sink.partition-commit.delay"],
"1h")
+
self.assertEqual(options["sink.partition-commit.policy.kind"], "success-file")
+ if file_format == "parquet":
+ self.assertEqual(options["parquet.compression"],
"GZIP")
+ else:
+
self.assertEqual(options["json.timestamp-format.standard"], "ISO-8601")
+ if target in ("local", "minicluster"):
+
execute_insert.return_value.wait.assert_called_once_with()
+ else:
+ execute_insert.return_value.wait.assert_not_called()
+
+
+class FilesystemIOITTests(PyFlinkDataFrameUTTestCase):
+ _SCHEMA = {"id": DataType.int64(), "name": DataType.string()}
+
+ def setUp(self):
+ super().setUp()
+ self.env.set_runtime_mode(RuntimeExecutionMode.BATCH)
+ self.env.set_parallelism(1)
+ self.t_env = StreamTableEnvironment.create(
+ self.env, environment_settings=EnvironmentSettings.in_batch_mode()
+ )
+ pf.set_table_environment(self.t_env)
+
+ def test_json_round_trip_overwrite_and_append(self):
+ self._check_round_trip_overwrite_and_append("json")
+
+ def test_json_partitioned_round_trip_and_overwrite(self):
+ self._check_partitioned_round_trip_and_overwrite("json")
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_partitioned_round_trip_and_overwrite(self):
+ self._check_partitioned_round_trip_and_overwrite("parquet")
+
+ def _check_partitioned_round_trip_and_overwrite(self, file_format):
+ output_path = os.path.join(self.tempdir, "partitioned-" + file_format)
+ reader = getattr(pf, "read_" + file_format)
+ original = pf.from_records([(1, "a"), (2, "b")], schema=["id", "name"])
+ getattr(original, "write_" + file_format)(output_path,
partition_by="name")
+ self.assertTrue(os.path.isdir(os.path.join(output_path, "name=a")))
+ self.assertTrue(os.path.isdir(os.path.join(output_path, "name=b")))
+ rows = reader(output_path, schema=self._SCHEMA,
partition_by="name").collect()
+ self.assertEqual(sorted(tuple(row) for row in rows), [(1, "a"), (2,
"b")])
+
+ replacement = pf.from_records([(3, "a")], schema=["id", "name"])
+ getattr(replacement, "write_" + file_format)(
+ output_path, partition_by=["name"], mode="overwrite")
+ rows = reader(output_path, schema=self._SCHEMA,
partition_by=["name"]).collect()
+ self.assertEqual(sorted(tuple(row) for row in rows), [(2, "b"), (3,
"a")])
+
+ additional = pf.from_records([(4, "a/b=c")], schema=["id", "name"])
+ getattr(additional, "write_" + file_format)(
+ output_path, partition_by="name")
+ rows = reader(output_path, schema=self._SCHEMA,
partition_by="name").collect()
+ self.assertEqual(sorted(tuple(row) for row in rows),
+ [(2, "b"), (3, "a"), (4, "a/b=c")])
+
+ default_path = os.path.join(self.tempdir, "default-partition-" +
file_format)
+ default_values = pf.from_records([(5, None), (6, "")], schema=["id",
"name"])
+ getattr(default_values, "write_" + file_format)(
+ default_path, partition_by="name",
+ connector_options={"partition.default-name": "unspecified"},
+ )
+ self.assertEqual([name for name in os.listdir(default_path) if not
name.startswith(".")],
+ ["name=unspecified"])
+
+ def test_partition_column_order_and_payload(self):
+ output_path = os.path.join(self.tempdir, "partition-order")
+ dataframe = pf.from_records([(1, "a", "b")], schema=["id", "first",
"second"])
+ dataframe.write_json(output_path, partition_by=["second", "first"])
+ partition_path = os.path.join(output_path, "second=b", "first=a")
+ self.assertTrue(os.path.isdir(partition_path))
+ records = []
+ for filename in os.listdir(partition_path):
+ if filename.startswith("part-"):
+ with open(os.path.join(partition_path, filename),
encoding="utf-8") as output:
+ records.extend(json.loads(line) for line in output)
+ self.assertEqual(records, [{"id": 1}])
+
+ def test_json_continuous_discovery_and_partition_commit(self):
+ self._check_continuous_discovery_and_partition_commit("json")
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_reader_discovers_new_files(self):
+ input_path = os.path.join(self.tempdir, "parquet-source")
+ output_path = os.path.join(self.tempdir, "json-sink")
+ os.makedirs(input_path)
+ staged_files = []
+ expected_rows = [(1, "first"), (2, "second")]
+ for row in expected_rows:
+ staging_path = os.path.join(self.tempdir, "parquet-staging-" +
str(row[0]))
+ pf.from_records([row], schema=["id",
"name"]).write_parquet(staging_path)
+ files = self._finished_files(staging_path)
+ self.assertEqual(len(files), 1)
+ staged_files.append(os.path.join(staging_path, files[0]))
+
+ # Publish only finished files so discovery cannot race with Parquet
footer writes.
+ os.replace(staged_files[0], os.path.join(input_path, "first.parquet"))
+ self.env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ self.env.enable_checkpointing(100)
+ self.t_env = StreamTableEnvironment.create(self.env)
+ pf.set_table_environment(self.t_env)
+ dataframe = pf.read_parquet(input_path, schema=self._SCHEMA,
monitor_interval="50ms")
+ sink = (TableDescriptor.for_connector("filesystem")
+ .option("path", output_path)
+ .option("format", "json")
+ .option("sink.rolling-policy.rollover-interval", "100ms")
+ .option("sink.rolling-policy.check-interval", "50ms")
+ .build())
+
+ def read_finished_rows():
+ rows = []
+ for filename in self._finished_files(output_path):
+ with open(os.path.join(output_path, filename),
encoding="utf-8") as output:
+ for line in output:
+ record = json.loads(line)
+ rows.append((record["id"], record["name"]))
+ return rows
+
+ result = dataframe.to_table().execute_insert(sink)
+ job_client = result.get_job_client()
+ execution_result = job_client.get_job_execution_result()
+ try:
+ self._wait_for(lambda: expected_rows[0] in read_finished_rows(),
+ "initial Parquet file was not read")
+ self.assertFalse(execution_result.done(), "Parquet source stopped
after initial files")
+ os.replace(staged_files[1], os.path.join(input_path,
"second.parquet"))
+ self._wait_for(lambda: expected_rows[1] in read_finished_rows(),
+ "new Parquet file was not discovered")
+ self.assertFalse(execution_result.done(), "continuous Parquet
source terminated")
+ finally:
+ self._cancel_job_and_wait(job_client)
+ self.assertCountEqual(read_finished_rows(), expected_rows)
+
+ def test_json_partition_time_commit(self):
+ self._check_continuous_discovery_and_partition_commit("json",
"partition-time")
+
+ def test_json_partition_time_commit_with_local_timezone(self):
+ self._check_continuous_discovery_and_partition_commit(
+ "json", "partition-time", "Asia/Shanghai")
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_continuous_discovery_and_partition_commit(self):
+ self._check_continuous_discovery_and_partition_commit("parquet")
+
+ def test_json_streaming_compaction(self):
+ self._check_continuous_discovery_and_partition_commit("json",
auto_compaction=True)
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_streaming_compaction(self):
+ self._check_continuous_discovery_and_partition_commit("parquet",
auto_compaction=True)
+
+ def _check_continuous_discovery_and_partition_commit(
+ self, file_format, partition_commit_trigger="process-time",
local_timezone="UTC",
+ auto_compaction=None,
+ ):
+ self.env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ self.env.enable_checkpointing(100)
+ self.t_env = StreamTableEnvironment.create(self.env)
+ self.t_env.get_config().set_local_timezone(local_timezone)
+ pf.set_table_environment(self.t_env)
+ suffix = self._testMethodName
+ input_path = os.path.join(self.tempdir, "continuous-input-" + suffix)
+ output_path = os.path.join(self.tempdir, "continuous-output-" + suffix)
+ os.makedirs(input_path)
+ partition_value = "2026-01-01"
+ # Keep the watermark near the partition boundary to expose time zone
mismatches.
+ timestamp = datetime(2026, 1, 1, 0, 1, tzinfo=ZoneInfo(local_timezone))
+ record = {"id": 1, "name": partition_value,
+ "ts_millis": int(timestamp.timestamp() * 1000)}
+ self._publish_json_file(input_path, "first.json", record)
+ dataframe = pf.read_json(
+ input_path, schema={**self._SCHEMA, "ts_millis": DataType.int64()},
+ monitor_interval="50ms",
+ computed_columns={"ts": "TO_TIMESTAMP_LTZ(ts_millis, 3)"},
+ watermark=("ts", "ts - INTERVAL '1' SECOND"),
+ )
+ results = []
+ execute_insert = dataframe._table.execute_insert
+
+ def capture_result(*args, **kwargs):
+ result = execute_insert(*args, **kwargs)
+ results.append(result)
+ return result
+
+ # Keep the public writer's actual execution, but manage this unbounded
job in the test.
+ with patch.object(dataframe._table, "execute_insert",
side_effect=capture_result), \
+ patch.object(TableResult, "wait"):
+ getattr(dataframe, "write_" + file_format)(
+ output_path, partition_by="name",
+ auto_compaction=auto_compaction,
+ compaction_file_size="1mb" if auto_compaction else None,
+ sink_shuffle_by_partition=True if auto_compaction else None,
+ rolling_policy_rollover_interval="100ms",
rolling_policy_check_interval="50ms",
+ partition_commit_policy_kind="success-file",
+ partition_commit_trigger=partition_commit_trigger,
+ connector_options={
+ "sink.partition-commit.success-file.name": "_READY",
+ "partition.time-extractor.timestamp-pattern": "$name
00:00:00",
+ "sink.partition-commit.watermark-time-zone":
local_timezone,
+ },
+ )
+ job_client = results[0].get_job_client()
+ partition_path = os.path.join(output_path, "name=" + partition_value)
+ try:
+ self._wait_for(
+ lambda: os.path.exists(os.path.join(partition_path, "_READY"))
+ and len(self._finished_files(partition_path)) >= 1,
+ "first checkpoint did not finish and commit the partition",
+ )
+ self._publish_json_file(input_path, "second.json", {**record,
"id": 2})
+ self._wait_for(
+ lambda: len(self._finished_files(partition_path)) >= 2,
+ "newly discovered file was not committed by a later
checkpoint",
+ )
+ if auto_compaction:
+ self.assertTrue(all(name.startswith("compacted-")
+ for name in
self._finished_files(partition_path)))
+ self.assertEqual(job_client.get_job_status().result(),
JobStatus.RUNNING)
+ finally:
+ self._cancel_job_and_wait(job_client)
+
+ self.env.set_runtime_mode(RuntimeExecutionMode.BATCH)
+ self.t_env = StreamTableEnvironment.create(
+ self.env, environment_settings=EnvironmentSettings.in_batch_mode())
+ pf.set_table_environment(self.t_env)
+ rows = getattr(pf, "read_" + file_format)(
+ output_path, schema=self._SCHEMA, partition_by="name").collect()
+ self.assertEqual(sorted(tuple(row) for row in rows),
+ [(1, partition_value), (2, partition_value)])
+
+ @staticmethod
+ def _publish_json_file(directory, filename, record):
+ temporary_path = os.path.join(directory, ".pending")
+ with open(temporary_path, "w", encoding="utf-8") as source:
+ source.write(json.dumps(record) + "\n")
+ os.replace(temporary_path, os.path.join(directory, filename))
+
+ @staticmethod
+ def _finished_files(directory):
+ if not os.path.isdir(directory):
+ return []
+ return [name for name in os.listdir(directory)
+ if name.startswith(("part-", "compacted-part-"))]
+
+ def _cancel_job_and_wait(self, job_client):
+ execution_result = job_client.get_job_execution_result()
+ # A finished job may have already shut down its MiniCluster.
+ if not execution_result.done():
+ cancellation = job_client.cancel()
+ self._wait_for(cancellation.done, "streaming job cancellation was
not acknowledged")
+ cancellation.result()
+ # Cancellation acknowledges the request before the job has fully
terminated.
+ self._wait_for(execution_result.done, "streaming job did not
terminate")
+
+ def _wait_for(self, predicate, message):
+ deadline = time.monotonic() + 60
+ while not predicate():
+ if time.monotonic() >= deadline:
+ self.fail(message)
+ time.sleep(0.05)
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_round_trip_overwrite_and_append(self):
+ self._check_round_trip_overwrite_and_append("parquet")
+
+ def _check_round_trip_overwrite_and_append(self, file_format):
+ output_path = os.path.join(self.tempdir, file_format)
+ reader = getattr(pf, "read_" + file_format)
+ original = pf.from_records([(1, "a"), (2, None)], schema=["id",
"name"])
+ getattr(original, "write_" + file_format)(output_path)
+ if file_format == "parquet":
+ self._assert_parquet_compression(output_path, "SNAPPY")
+ rows = reader(output_path, schema=self._SCHEMA).collect()
+ self.assertEqual(sorted(tuple(row) for row in rows), [(1, "a"), (2,
None)])
+
+ replacement = pf.from_records([(3, "c")], schema=["id", "name"])
+ options = {"compression": "GZIP"} if file_format == "parquet" else {}
+ getattr(replacement, "write_" + file_format)(output_path,
mode="overwrite", **options)
+ if file_format == "parquet":
+ self._assert_parquet_compression(output_path, "GZIP")
+ rows = reader(output_path, schema=self._SCHEMA).collect()
+ self.assertEqual([tuple(row) for row in rows], [(3, "c")])
+
+ getattr(original, "write_" + file_format)(output_path)
+ rows = reader(output_path, schema=self._SCHEMA).collect()
+ self.assertEqual(sorted(tuple(row) for row in rows),
+ [(1, "a"), (2, None), (3, "c")])
+
+ def _assert_parquet_compression(self, output_path, expected_codec):
+ jvm = get_gateway().jvm
+ hadoop_conf = jvm.org.apache.hadoop.conf.Configuration()
+ codecs = set()
+ for filename in os.listdir(output_path):
+ if filename.startswith("part-"):
+ path = jvm.org.apache.hadoop.fs.Path(os.path.join(output_path,
filename))
+ footer =
jvm.org.apache.parquet.hadoop.ParquetFileReader.readFooter(
+ hadoop_conf, path
+ )
+ for block in footer.getBlocks():
+ codecs.update(column.getCodec().name() for column in
block.getColumns())
+ self.assertEqual(codecs, {expected_codec})
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_timestamp_encoding_options(self):
+ timestamp = datetime(1970, 1, 1, 0, 0, 1, 123456)
+ dataframe = pf.sql("SELECT TIMESTAMP '1970-01-01 00:00:01.123456' AS
ts")
+ jvm = get_gateway().jvm
+ hadoop_conf = jvm.org.apache.hadoop.conf.Configuration()
+ cases = [(None, None), ("millis", 1123), ("micros", 1123456),
("nanos", 1123456000)]
+ for unit, encoded_timestamp in cases:
+ with self.subTest(unit=unit):
+ output_path = os.path.join(self.tempdir, "parquet-timestamp-"
+ str(unit))
+ options = {} if unit is None else {
+ "write.int64.timestamp": "true", "timestamp.time.unit":
unit,
+ }
+ dataframe.write_parquet(output_path, utc_timezone=True,
format_options=options)
+ files = self._finished_files(output_path)
+ self.assertEqual(len(files), 1)
+ path = jvm.org.apache.hadoop.fs.Path(os.path.join(output_path,
files[0]))
+ footer =
jvm.org.apache.parquet.hadoop.ParquetFileReader.readFooter(
+ hadoop_conf, path)
+ column_type =
footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType()
+ if unit is None:
+
self.assertEqual(column_type.getPrimitiveTypeName().name(), "INT96")
+ else:
+
self.assertEqual(column_type.getPrimitiveTypeName().name(), "INT64")
+
self.assertEqual(column_type.getLogicalTypeAnnotation().getUnit().name(),
+ unit.upper())
+ reader =
(jvm.org.apache.parquet.hadoop.ParquetReader.builder(
+
jvm.org.apache.parquet.hadoop.example.GroupReadSupport(), path)
+ .withConf(hadoop_conf).build())
+ try:
+ self.assertEqual(reader.read().getLong("ts", 0),
encoded_timestamp)
+ self.assertIsNone(reader.read())
+ finally:
+ reader.close()
+ restored = pf.read_parquet(
+ output_path, schema={"ts": DataType.timestamp(6)},
utc_timezone=True)
+ expected = timestamp.replace(microsecond=123000) if unit ==
"millis" else timestamp
+ view_name = "parquet_timestamps_" + str(unit)
+ self.t_env.create_temporary_view(view_name,
restored.to_table())
+ # Compare in SQL so precision checks do not depend on Python
timestamp conversion.
+ rows = pf.sql(
+ "SELECT ts = TIMESTAMP '" + expected.isoformat(sep=" ")
+ + "' AS is_equal FROM " + view_name).collect()
+ self.assertEqual([tuple(row) for row in rows], [(True,)])
+
+ def test_json_streaming_append_and_overwrite_rejection(self):
+ self._check_streaming_append_and_overwrite_rejection("json")
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_streaming_append_and_overwrite_rejection(self):
+ self._check_streaming_append_and_overwrite_rejection("parquet")
+
+ def _check_streaming_append_and_overwrite_rejection(self, file_format):
+ self.env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ self.t_env = StreamTableEnvironment.create(self.env)
+ pf.set_table_environment(self.t_env)
+ dataframe = pf.from_records([(1, "a")], schema=["id", "name"])
+ output_path = os.path.join(self.tempdir, "stream-" + file_format)
+ writer = getattr(dataframe, "write_" + file_format)
+ with self.assertRaisesRegex(Py4JJavaError, "Streaming mode not support
overwrite"):
+ writer(output_path, mode="overwrite")
+ self.assertFalse(os.path.exists(output_path))
+ writer(output_path)
+ writer(output_path, mode="append")
+ rows = getattr(pf, "read_" + file_format)(
+ output_path, schema=self._SCHEMA
+ ).collect()
+ self.assertEqual([tuple(row) for row in rows], [(1, "a"), (1, "a")])
+
+ def test_json_encoding_options(self):
+ dataframe = pf.sql(
+ "SELECT CAST(NULL AS STRING) AS name, CAST(0.0000001 AS
DECIMAL(10, 7)) AS amount")
+ for enabled in (None, False, True):
+ with self.subTest(enabled=enabled):
+ output_path = os.path.join(self.tempdir, "json-encoding-" +
str(enabled))
+ dataframe.write_json(
+ output_path, ignore_null_fields=enabled,
decimal_as_plain_number=enabled)
+ records = []
+ for filename in os.listdir(output_path):
+ if filename.startswith("part-"):
+ with open(os.path.join(output_path, filename),
encoding="utf-8") as output:
+ for line in output:
+
self.assertIsInstance(json.loads(line)["amount"], float)
+ records.append(json.loads(line,
parse_float=str))
+ expected = {"amount": "0.0000001"} if enabled else {"name":
None, "amount": "1E-7"}
+ self.assertEqual(records, [expected])
+
+ def test_json_map_null_keys_are_independent_of_null_fields(self):
+ dataframe = pf.sql(
+ "SELECT CAST(NULL AS STRING) AS name, "
+ "MAP[CAST(NULL AS STRING), 1, 'known', 2] AS attributes")
+ cases = [
+ ({"map-null-key.mode": "DROP"}, {"attributes": {"known": 2}}),
+ ({"map-null-key.mode": "LITERAL", "map-null-key.literal":
"missing"},
+ {"attributes": {"missing": 1, "known": 2}}),
+ ]
+ for options, expected in cases:
+ with self.subTest(options=options):
+ output_path = os.path.join(self.tempdir, "json-map-" +
options["map-null-key.mode"])
+ dataframe.write_json(output_path, ignore_null_fields=True,
format_options=options)
+ records = []
+ for filename in self._finished_files(output_path):
+ with open(os.path.join(output_path, filename),
encoding="utf-8") as output:
+ records.extend(json.loads(line) for line in output)
+ self.assertEqual(records, [expected])
+
+ def test_json_format_options_and_path_filter(self):
+ input_path = os.path.join(self.tempdir, "json-input")
+ os.makedirs(input_path)
+ with open(os.path.join(input_path, "data.json"), "w",
encoding="utf-8") as source:
+
source.write('{"id":1,"name":"a"}\nnot-json\n{"id":2,"name":null}\n')
+ with open(os.path.join(input_path, "excluded.json"), "w",
encoding="utf-8") as source:
+ source.write('{"id":99,"name":"excluded"}\n')
+ rows = pf.read_json(
+ input_path, schema=self._SCHEMA,
path_regex_pattern=".*data[.]json",
+ format_options={"ignore-parse-errors": "true"},
+ ).collect()
+ self.assertEqual(sorted(tuple(row) for row in rows), [(1, "a"), (2,
None)])
+ rows = pf.read_json(
+ input_path, schema=self._SCHEMA, path_regex_pattern="data[.]json",
+ ).collect()
+ self.assertEqual(rows, [])
+
+ def test_json_continuous_source_is_rejected_in_batch_mode(self):
+ self._check_continuous_source_is_rejected_in_batch_mode(pf.read_json)
+
+ @unittest.skipIf(os.environ.get('HADOOP_CLASSPATH') is None,
+ 'Hadoop libraries are required for Parquet format tests')
+ def test_parquet_continuous_source_is_rejected_in_batch_mode(self):
+
self._check_continuous_source_is_rejected_in_batch_mode(pf.read_parquet)
+
+ def _check_continuous_source_is_rejected_in_batch_mode(self, reader):
+ input_path = os.path.join(self.tempdir, self._testMethodName)
+ os.makedirs(input_path)
+ dataframe = reader(input_path, schema=self._SCHEMA,
monitor_interval="1s")
+ with self.assertRaisesRegex(Py4JJavaError, "[Uu]nbounded"):
+ dataframe.to_table().explain()
+
+ def test_computed_watermark_supports_event_time_windows(self):
+ input_path = os.path.join(self.tempdir, "event-time.json")
+ with open(input_path, "w", encoding="utf-8") as source:
+ source.write('{"ts_millis":1000}\n{"ts_millis":2000}\n')
+ self.env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ self.t_env = StreamTableEnvironment.create(self.env)
+ pf.set_table_environment(self.t_env)
+ dataframe = pf.read_json(
+ input_path, schema={"ts_millis": DataType.int64()},
+ computed_columns={"ts": "TO_TIMESTAMP_LTZ(ts_millis, 3)"},
+ watermark=("ts", "ts - INTERVAL '1' SECOND"),
+ )
+ self.t_env.create_temporary_view("events", dataframe.to_table())
+ result = pf.sql(
+ "SELECT COUNT(*) AS cnt FROM TABLE("
+ "TUMBLE(TABLE events, DESCRIPTOR(ts), INTERVAL '5' SECOND)) "
+ "GROUP BY window_start, window_end"
+ ).collect()
+ self.assertEqual([tuple(row) for row in result], [(2,)])
+
+ def test_json_timestamp_format_options_round_trip(self):
+ output_path = os.path.join(self.tempdir, "json-timestamp")
+ timestamp = datetime(2026, 1, 2, 3, 4, 5)
+ dataframe = pf.sql("SELECT TIMESTAMP '2026-01-02 03:04:05' AS ts")
+ dataframe.write_json(
+ output_path, timestamp_format="ISO-8601",
+ format_options={"timestamp-format.standard": "ISO-8601"},
+ )
+ records = []
+ for filename in os.listdir(output_path):
+ if filename.startswith("part-"):
+ with open(os.path.join(output_path, filename),
encoding="utf-8") as output:
+ records.extend(json.loads(line) for line in output)
+ self.assertEqual(records, [{"ts": timestamp.isoformat()}])
+
+ rows = pf.read_json(
+ output_path, schema={"ts": DataType.timestamp()},
+ timestamp_format="ISO-8601",
+ format_options={"json.timestamp-format.standard": "ISO-8601"},
+ ).collect()
+ self.assertEqual([tuple(row) for row in rows], [(timestamp,)])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/flink-python/pyflink/dataframe/tests/test_io.py
b/flink-python/pyflink/dataframe/tests/test_io.py
index 708f3aa3439..ef57cfde616 100644
--- a/flink-python/pyflink/dataframe/tests/test_io.py
+++ b/flink-python/pyflink/dataframe/tests/test_io.py
@@ -221,6 +221,7 @@ class GenericIOTests(PyFlinkDataFrameUTTestCase):
self.assertIsNone(descriptor.get_schema())
self.assertEqual(descriptor.get_options().get("connector"),
"blackhole")
self.assertEqual(descriptor.get_options().get("sink.parallelism"), "1")
+ self.assertEqual(execute_insert.call_args.kwargs, {"overwrite": False})
def test_write_generic_waits_for_local_and_minicluster_execution(self):
dataframe = pf.from_records([(1,)], schema=["id"])
