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