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