You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "csurong (via GitHub)" <gi...@apache.org> on 2026/09/20 04:07:14 UTC
[PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
csurong opened a new pull request, #29245:
URL: https://github.com/apache/flink/pull/29245
## What is the purpose of the change
Add Parquet and JSON filesystem I/O to the PyFlink DataFrame API, as described in [FLINK-40426](https://issues.apache.org/jira/browse/FLINK-40426) and FLIP-591.
## Brief change log
- Add `read_parquet`, `read_json`, `write_parquet`, and `write_json`, using the existing filesystem connector.
- Support format options and append/overwrite writes while preserving `write_generic` behavior.
- Add tests and API documentation with a usage example.
## Verifying this change
- [Flink CI (Java 17)](https://github.com/csurong/flink/actions/runs/35483836376) passed, including basic QA, compilation, packaging/licensing, tests, and end-to-end tests.
- All 21 tests in `test_filesystem_io.py` and `test_io.py` passed, including real JSON/Parquet reads and writes, write modes, format options, and argument validation.
- Flake8, targeted mypy checks, strict Sphinx build, and a clean build of `flink-python` with its dependencies passed (Java tests skipped in that build).
- Full `clean verify` was attempted, but four OpenTelemetry tests failed because no Docker environment was available locally.
## Does this pull request potentially affect one of the following parts:
- Dependencies: no
- Public API: yes, the Python helpers specified in FLIP-591
- Serializers: no
- Runtime per-record code paths: no
- Deployment or recovery: no
- S3 file system connector: no
## Documentation
- New feature: yes
- Documented in the Python API docstrings and DataFrame I/O reference.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: OpenAI Codex (GPT-6)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4069749143
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ compression: Optional[str] = None,
+ utc_timezone: Optional[bool] = None,
+ sink_parallelism: Optional[int] = 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,
Review Comment:
Could we also add the following options `sink.shuffle-by-partition.enable`, `auto-compaction`, `compaction.file-size` to the parameters of both write_parquet/write_json which I think may be frequently used?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4059101358
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,125 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ compression: str = "SNAPPY",
+ rolling_policy_file_size: str = "128mb",
+ rolling_policy_rollover_interval: str = "30min",
+ rolling_policy_check_interval: Optional[str] = None,
+ partition_commit_trigger: str = "process-time",
Review Comment:
Thanks for catching this. Added `partition_by` to both readers and writers in b104961a064, with the partition keys passed to `TableDescriptor.partitioned_by(...)`. I also added partitioned round-trip/overwrite tests and streaming partition-commit tests, including a non-UTC case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4090273733
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
Review Comment:
Thanks, both writers now default to append in batch and streaming. I've updated the docs and added tests for append and streaming overwrite rejection.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4059102688
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,125 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
Review Comment:
Agreed. I went through the connector and format options and added parameters for sink parallelism, inactivity rolling, JSON parsing/timestamps, and Parquet UTC conversion. All four helpers now also accept `connector_options` and `format_options` for options without a dedicated parameter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4088915411
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
Review Comment:
Makes sense to me~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4059106066
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -24,7 +24,155 @@
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,
+ options: Dict[str, Optional[str]],
+ 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}
+ result.update({key: value for key, value in options.items() if value is not None})
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in format_options.items():
+ option = key if key.startswith(file_format + ".") else file_format + "." + key
+ if option in result:
+ raise ValueError(f"duplicate format option: {option!r}")
+ result[option] = value
+ _validate_options(result)
+ return result
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ rolling_policy_file_size: str,
+ rolling_policy_rollover_interval: str,
+ rolling_policy_check_interval: Optional[str],
+ partition_commit_trigger: str,
+ partition_commit_delay: str,
+ partition_commit_policy_kind: Optional[str],
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ options = {
+ "sink.rolling-policy.file-size": rolling_policy_file_size,
+ "sink.rolling-policy.rollover-interval": rolling_policy_rollover_interval,
+ "sink.partition-commit.trigger": partition_commit_trigger,
+ "sink.partition-commit.delay": partition_commit_delay,
+ }
+ _validate_options(options)
+ return _build_filesystem_options(
+ path,
+ file_format,
+ {
+ **options,
+ "sink.rolling-policy.check-interval": rolling_policy_check_interval,
+ "sink.partition-commit.policy.kind": partition_commit_policy_kind,
+ },
+ format_options,
+ )
+
+
+@PublicEvolving()
+def read_parquet(
+ path: str,
+ *,
+ schema: Dict[str, DataType],
+ monitor_interval: Optional[str] = None,
+ path_regex_pattern: Optional[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 a
+ continuous source that discovers new files.
+
+ :param path: File or directory URI supported by Flink's filesystem implementations.
+ :param schema: Mapping of column names to DataFrame data types.
+ :param monitor_interval: Optional file discovery interval, for example ``"60s"``.
Review Comment:
How about defaulting to append for both batch and streaming, consistent with the non-overwrite behavior of Java’s Table.executeInsert() and write_generic()? This would also avoid checking the execution mode in the Python wrapper.
Explicit overwrite would still work in batch and be rejected by the filesystem connector in streaming. I’ll make that clear in the API docs.
This would change the batch default from overwrite in FLIP-591 too. Would that be OK with you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4059106066
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -24,7 +24,155 @@
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,
+ options: Dict[str, Optional[str]],
+ 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}
+ result.update({key: value for key, value in options.items() if value is not None})
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in format_options.items():
+ option = key if key.startswith(file_format + ".") else file_format + "." + key
+ if option in result:
+ raise ValueError(f"duplicate format option: {option!r}")
+ result[option] = value
+ _validate_options(result)
+ return result
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ rolling_policy_file_size: str,
+ rolling_policy_rollover_interval: str,
+ rolling_policy_check_interval: Optional[str],
+ partition_commit_trigger: str,
+ partition_commit_delay: str,
+ partition_commit_policy_kind: Optional[str],
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ options = {
+ "sink.rolling-policy.file-size": rolling_policy_file_size,
+ "sink.rolling-policy.rollover-interval": rolling_policy_rollover_interval,
+ "sink.partition-commit.trigger": partition_commit_trigger,
+ "sink.partition-commit.delay": partition_commit_delay,
+ }
+ _validate_options(options)
+ return _build_filesystem_options(
+ path,
+ file_format,
+ {
+ **options,
+ "sink.rolling-policy.check-interval": rolling_policy_check_interval,
+ "sink.partition-commit.policy.kind": partition_commit_policy_kind,
+ },
+ format_options,
+ )
+
+
+@PublicEvolving()
+def read_parquet(
+ path: str,
+ *,
+ schema: Dict[str, DataType],
+ monitor_interval: Optional[str] = None,
+ path_regex_pattern: Optional[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 a
+ continuous source that discovers new files.
+
+ :param path: File or directory URI supported by Flink's filesystem implementations.
+ :param schema: Mapping of column names to DataFrame data types.
+ :param monitor_interval: Optional file discovery interval, for example ``"60s"``.
Review Comment:
How about defaulting to append for both batch and streaming, consistent with the non-overwrite behavior of Java’s Table.executeInsert() and write_generic()? This would also avoid checking the execution mode in the Python wrapper.
Explicit overwrite would still work in batch and be rejected by the filesystem connector in streaming. I’ll make that clear in the API docs.
This would change the batch default from overwrite in FLIP-591 too. Would that be OK with you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu merged PR #29245:
URL: https://github.com/apache/flink/pull/29245
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4069430565
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
Review Comment:
Do you think if it makes sense to set the `mode` to `append` for streaming jobs as this is the only supported mode for streaming jobs? It's necessary to document this behavior clearly in the API doc.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ compression: Optional[str] = None,
+ utc_timezone: Optional[bool] = None,
+ sink_parallelism: Optional[int] = 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,
Review Comment:
Could you also add the following options `sink.shuffle-by-partition.enable`, `auto-compaction`, `compaction.file-size` to the parameters of both write_parquet/write_json which I think may be frequently used?
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -16,15 +16,269 @@
# 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(
Review Comment:
The four option arguments represent two independent dimensions: connector vs. format options, and named convenience parameters vs. user-provided escape-hatch dictionaries.
The current names (`options`, `format_parameters`, `connector_options`, and `format_options`) do not make this distinction clear.
What about renaming them consistently, for example to `connector_parameters`, `format_parameters`, `extra_connector_options`, and `extra_format_options`, make them keyword-only.
For example:
```
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]:
```
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -16,15 +16,269 @@
# 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,
+ options: Dict[str, Optional[str]],
+ format_options: Optional[Dict[str, str]] = None,
+ *,
+ connector_options: Optional[Dict[str, str]] = None,
+ format_parameters: Optional[Dict[str, Optional[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 connector_options is not None:
+ _validate_options(connector_options)
+ for key in 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, connector_options)
+ _merge_options(result, {key: value for key, value in options.items() if value is not None})
+
+ normalized_format_options: Dict[str, str] = {}
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in 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)
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ options: Dict[str, Optional[str]],
+ format_parameters: Dict[str, Optional[str]],
+ connector_options: Optional[Dict[str, str]] = None,
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ result = _build_filesystem_options(
+ path, file_format, options, format_options,
+ connector_options=connector_options, format_parameters=format_parameters,
+ )
+ # Apply defaults after merging explicit settings from either API entry point.
+ defaults = {
Review Comment:
This helper re-materializes only a subset of the upstream defaults, while other documented defaults are left to the connector factory. What about using one consistent model? Preferably keep `None` as “unspecified”, omit the option, and let the connector/format factory apply its default. The API docs can state the current effective default explicitly as following: `If None, the connector default is used (currently "SNAPPY")`.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ compression: Optional[str] = None,
+ utc_timezone: Optional[bool] = None,
+ sink_parallelism: Optional[int] = 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. Overwrite requires batch
+ execution; use ``mode="append"`` for streaming execution. Partitioned overwrite replaces
+ only partitions present in the input, retaining other partitions.
+
+ 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: ``"overwrite"`` (default) replaces existing data; ``"append"`` adds files.
+ :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, defaulting to ``"SNAPPY"``.
+ :param utc_timezone: Use UTC for Parquet timestamp conversion. Defaults to ``False``,
+ which uses the JVM default time zone, independently of the session time zone.
+ :param sink_parallelism: Sink parallelism; defaults to the upstream parallelism.
+ :param rolling_policy_file_size: Part file size threshold for rolling, default ``"128mb"``.
+ This is not a hard upper bound on file size.
+ :param rolling_policy_rollover_interval: Part file open-time threshold, default ``"30min"``.
+ :param rolling_policy_inactivity_interval: Part file inactivity threshold, default
+ ``"30min"``.
+ :param rolling_policy_check_interval: Interval for checking time-based rolling policies,
+ default ``"1min"``.
+ :param partition_commit_trigger: Partition commit trigger: ``"process-time"`` (default) or
+ ``"partition-time"``.
+ :param partition_commit_delay: Delay before committing a partition, default ``"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. Defaults are applied only after merging explicit settings.
+ :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.
+ :raises TypeError: If an argument has an invalid type.
+ :raises ValueError: If the path or partition keys are invalid, the write mode is
+ unsupported, or options conflict.
+
+ 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_sink_options,
+ _parallelism_option,
+ )
+
+ options = _build_filesystem_sink_options(
+ path,
+ "parquet",
+ {
+ "sink.parallelism": _parallelism_option(sink_parallelism),
+ "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,
+ },
+ {
+ "compression": compression,
+ "utc-timezone": _boolean_option(utc_timezone, "utc_timezone"),
+ },
+ connector_options=connector_options,
+ format_options=format_options,
+ )
+ self._write("filesystem", options, mode, partition_by)
+
+ @PublicEvolving()
+ def write_json(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ timestamp_format: Optional[str] = None,
+ sink_parallelism: Optional[int] = 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,
Review Comment:
Could we also expose the following parameters `encode.ignore-null-fields` and `encode.decimal-as-plain-number` for write_json?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4059104545
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,125 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ compression: str = "SNAPPY",
Review Comment:
Parquet now accepts `format_options` too. I used `connector_options` for advanced sink options to keep the naming consistent across readers and writers. Defaults are applied after user options, so `format_options={"compression": "GZIP"}` is preserved rather than overwritten by the SNAPPY default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4090290925
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ compression: Optional[str] = None,
+ utc_timezone: Optional[bool] = None,
+ sink_parallelism: Optional[int] = 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. Overwrite requires batch
+ execution; use ``mode="append"`` for streaming execution. Partitioned overwrite replaces
+ only partitions present in the input, retaining other partitions.
+
+ 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: ``"overwrite"`` (default) replaces existing data; ``"append"`` adds files.
+ :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, defaulting to ``"SNAPPY"``.
+ :param utc_timezone: Use UTC for Parquet timestamp conversion. Defaults to ``False``,
+ which uses the JVM default time zone, independently of the session time zone.
+ :param sink_parallelism: Sink parallelism; defaults to the upstream parallelism.
+ :param rolling_policy_file_size: Part file size threshold for rolling, default ``"128mb"``.
+ This is not a hard upper bound on file size.
+ :param rolling_policy_rollover_interval: Part file open-time threshold, default ``"30min"``.
+ :param rolling_policy_inactivity_interval: Part file inactivity threshold, default
+ ``"30min"``.
+ :param rolling_policy_check_interval: Interval for checking time-based rolling policies,
+ default ``"1min"``.
+ :param partition_commit_trigger: Partition commit trigger: ``"process-time"`` (default) or
+ ``"partition-time"``.
+ :param partition_commit_delay: Delay before committing a partition, default ``"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. Defaults are applied only after merging explicit settings.
+ :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.
+ :raises TypeError: If an argument has an invalid type.
+ :raises ValueError: If the path or partition keys are invalid, the write mode is
+ unsupported, or options conflict.
+
+ 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_sink_options,
+ _parallelism_option,
+ )
+
+ options = _build_filesystem_sink_options(
+ path,
+ "parquet",
+ {
+ "sink.parallelism": _parallelism_option(sink_parallelism),
+ "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,
+ },
+ {
+ "compression": compression,
+ "utc-timezone": _boolean_option(utc_timezone, "utc_timezone"),
+ },
+ connector_options=connector_options,
+ format_options=format_options,
+ )
+ self._write("filesystem", options, mode, partition_by)
+
+ @PublicEvolving()
+ def write_json(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ timestamp_format: Optional[str] = None,
+ sink_parallelism: Optional[int] = 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,
Review Comment:
Added ignore_null_fields and decimal_as_plain_number to write_json, with tests checking the actual JSON output.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4078523296
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
Review Comment:
How about defaulting to append for both batch and streaming, consistent with the non-overwrite behavior of Java’s Table.executeInsert() and write_generic()? This would also avoid checking the execution mode in the Python wrapper.
Explicit overwrite would still work in batch and be rejected by the filesystem connector in streaming. I’ll make that clear in the API docs.
This would change the batch default from overwrite in FLIP-591 too. Would that be OK with you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29245:
URL: https://github.com/apache/flink/pull/29245#issuecomment-5747561911
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "edf26e20b564f2dd4da107c3cc4ad6197f57705c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "edf26e20b564f2dd4da107c3cc4ad6197f57705c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* edf26e20b564f2dd4da107c3cc4ad6197f57705c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4059106066
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -24,7 +24,155 @@
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,
+ options: Dict[str, Optional[str]],
+ 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}
+ result.update({key: value for key, value in options.items() if value is not None})
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in format_options.items():
+ option = key if key.startswith(file_format + ".") else file_format + "." + key
+ if option in result:
+ raise ValueError(f"duplicate format option: {option!r}")
+ result[option] = value
+ _validate_options(result)
+ return result
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ rolling_policy_file_size: str,
+ rolling_policy_rollover_interval: str,
+ rolling_policy_check_interval: Optional[str],
+ partition_commit_trigger: str,
+ partition_commit_delay: str,
+ partition_commit_policy_kind: Optional[str],
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ options = {
+ "sink.rolling-policy.file-size": rolling_policy_file_size,
+ "sink.rolling-policy.rollover-interval": rolling_policy_rollover_interval,
+ "sink.partition-commit.trigger": partition_commit_trigger,
+ "sink.partition-commit.delay": partition_commit_delay,
+ }
+ _validate_options(options)
+ return _build_filesystem_options(
+ path,
+ file_format,
+ {
+ **options,
+ "sink.rolling-policy.check-interval": rolling_policy_check_interval,
+ "sink.partition-commit.policy.kind": partition_commit_policy_kind,
+ },
+ format_options,
+ )
+
+
+@PublicEvolving()
+def read_parquet(
+ path: str,
+ *,
+ schema: Dict[str, DataType],
+ monitor_interval: Optional[str] = None,
+ path_regex_pattern: Optional[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 a
+ continuous source that discovers new files.
+
+ :param path: File or directory URI supported by Flink's filesystem implementations.
+ :param schema: Mapping of column names to DataFrame data types.
+ :param monitor_interval: Optional file discovery interval, for example ``"60s"``.
Review Comment:
Updated both reader docstrings and the I/O docs to spell this out: `monitor_interval` enables unbounded file discovery, while the regex matches the entire path, excluding the scheme and authority. The docs also explain that files aren't tailed and new partition directories aren't discovered. Added tests for full-path matching and rejection of unbounded reads in batch mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4090285459
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -16,15 +16,269 @@
# 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(
Review Comment:
Renamed the four arguments as suggested and made them keyword-only. Readers and writers now use the same helper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4056388706
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,125 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ compression: str = "SNAPPY",
+ rolling_policy_file_size: str = "128mb",
+ rolling_policy_rollover_interval: str = "30min",
+ rolling_policy_check_interval: Optional[str] = None,
+ partition_commit_trigger: str = "process-time",
Review Comment:
Partition commit only works for partitioned sinks, but the writer APIs do not expose partition columns and the descriptor never calls `partitioned_by(...)`. Please add complete `partition_by` support.
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -24,7 +24,155 @@
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,
+ options: Dict[str, Optional[str]],
+ 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}
+ result.update({key: value for key, value in options.items() if value is not None})
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in format_options.items():
+ option = key if key.startswith(file_format + ".") else file_format + "." + key
+ if option in result:
+ raise ValueError(f"duplicate format option: {option!r}")
+ result[option] = value
+ _validate_options(result)
+ return result
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ rolling_policy_file_size: str,
+ rolling_policy_rollover_interval: str,
+ rolling_policy_check_interval: Optional[str],
+ partition_commit_trigger: str,
+ partition_commit_delay: str,
+ partition_commit_policy_kind: Optional[str],
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ options = {
+ "sink.rolling-policy.file-size": rolling_policy_file_size,
+ "sink.rolling-policy.rollover-interval": rolling_policy_rollover_interval,
+ "sink.partition-commit.trigger": partition_commit_trigger,
+ "sink.partition-commit.delay": partition_commit_delay,
+ }
+ _validate_options(options)
+ return _build_filesystem_options(
+ path,
+ file_format,
+ {
+ **options,
+ "sink.rolling-policy.check-interval": rolling_policy_check_interval,
+ "sink.partition-commit.policy.kind": partition_commit_policy_kind,
+ },
+ format_options,
+ )
+
+
+@PublicEvolving()
+def read_parquet(
+ path: str,
+ *,
+ schema: Dict[str, DataType],
+ monitor_interval: Optional[str] = None,
+ path_regex_pattern: Optional[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 a
+ continuous source that discovers new files.
+
+ :param path: File or directory URI supported by Flink's filesystem implementations.
+ :param schema: Mapping of column names to DataFrame data types.
+ :param monitor_interval: Optional file discovery interval, for example ``"60s"``.
Review Comment:
Could we document the exact semantics here? Setting `monitor_interval` changes the source from a bounded scan to continuous file discovery, while `path_regex_pattern` performs a regex full match against the file path. This distinction is important for users configuring streaming reads.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,125 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ compression: str = "SNAPPY",
Review Comment:
JSON provides a generic `format_options` escape hatch. Could we provide equivalent `format_options` for Parquet and consider a generic `sink_options` escape hatch for advanced upstream options? This avoids adding a new public argument for every supported connector option.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,125 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
Review Comment:
There are many options defined in the filesystem connector, I think we are not limited to the parameters defined in the FLIP-591.
Do you think it makes sense to make the following changes:
- Evaluate the options defined in the filesystem connector and parquet / json format and make sure frequently used options are defined here.
- Introduce parameters connector_options and format_options for the options which are not defined here. It could also avoid functionality break if users introduce new features in the filesystem connector or parquet/json format and forget to update this API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4090276170
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -16,15 +16,269 @@
# 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,
+ options: Dict[str, Optional[str]],
+ format_options: Optional[Dict[str, str]] = None,
+ *,
+ connector_options: Optional[Dict[str, str]] = None,
+ format_parameters: Optional[Dict[str, Optional[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 connector_options is not None:
+ _validate_options(connector_options)
+ for key in 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, connector_options)
+ _merge_options(result, {key: value for key, value in options.items() if value is not None})
+
+ normalized_format_options: Dict[str, str] = {}
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in 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)
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ options: Dict[str, Optional[str]],
+ format_parameters: Dict[str, Optional[str]],
+ connector_options: Optional[Dict[str, str]] = None,
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ result = _build_filesystem_options(
+ path, file_format, options, format_options,
+ connector_options=connector_options, format_parameters=format_parameters,
+ )
+ # Apply defaults after merging explicit settings from either API entry point.
+ defaults = {
Review Comment:
Removed the Python-side defaults. Options left unspecified are now resolved by the connector or format factory, and the docstrings describe the current defaults.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4090279420
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -1369,6 +1369,205 @@ def columns(self) -> List[str]:
# ======================== I/O ========================
+ @PublicEvolving()
+ def write_parquet(
+ self,
+ path: str,
+ *,
+ mode: str = "overwrite",
+ partition_by: Optional[Union[str, List[str]]] = None,
+ compression: Optional[str] = None,
+ utc_timezone: Optional[bool] = None,
+ sink_parallelism: Optional[int] = 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,
Review Comment:
Added all three to both writers, with tests for option forwarding and streaming compaction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40426][python] Add Parquet and JSON filesystem I/O to DataFrame API [flink]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4078545277
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -24,7 +24,155 @@
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,
+ options: Dict[str, Optional[str]],
+ 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}
+ result.update({key: value for key, value in options.items() if value is not None})
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in format_options.items():
+ option = key if key.startswith(file_format + ".") else file_format + "." + key
+ if option in result:
+ raise ValueError(f"duplicate format option: {option!r}")
+ result[option] = value
+ _validate_options(result)
+ return result
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ rolling_policy_file_size: str,
+ rolling_policy_rollover_interval: str,
+ rolling_policy_check_interval: Optional[str],
+ partition_commit_trigger: str,
+ partition_commit_delay: str,
+ partition_commit_policy_kind: Optional[str],
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ options = {
+ "sink.rolling-policy.file-size": rolling_policy_file_size,
+ "sink.rolling-policy.rollover-interval": rolling_policy_rollover_interval,
+ "sink.partition-commit.trigger": partition_commit_trigger,
+ "sink.partition-commit.delay": partition_commit_delay,
+ }
+ _validate_options(options)
+ return _build_filesystem_options(
+ path,
+ file_format,
+ {
+ **options,
+ "sink.rolling-policy.check-interval": rolling_policy_check_interval,
+ "sink.partition-commit.policy.kind": partition_commit_policy_kind,
+ },
+ format_options,
+ )
+
+
+@PublicEvolving()
+def read_parquet(
+ path: str,
+ *,
+ schema: Dict[str, DataType],
+ monitor_interval: Optional[str] = None,
+ path_regex_pattern: Optional[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 a
+ continuous source that discovers new files.
+
+ :param path: File or directory URI supported by Flink's filesystem implementations.
+ :param schema: Mapping of column names to DataFrame data types.
+ :param monitor_interval: Optional file discovery interval, for example ``"60s"``.
Review Comment:
Updated both reader docstrings and the I/O docs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org