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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
