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