[
https://issues.apache.org/jira/browse/FLINK-40199?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu reassigned FLINK-40199:
-------------------------------
Assignee: Liu Liu
> Add generic connector source and sink APIs to PyFlink DataFrame
> ---------------------------------------------------------------
>
> Key: FLINK-40199
> URL: https://issues.apache.org/jira/browse/FLINK-40199
> Project: Flink
> Issue Type: Sub-task
> Components: API / Python
> Reporter: Dian Fu
> Assignee: Liu Liu
> Priority: Major
> Labels: pull-request-available
> Fix For: 2.4.0
>
>
> def read_generic(
> connector: str,
> *,
> schema: Dict[str, DataType],
> options: Dict[str, str]
> ) -> DataFrame
> def DataFrame.write_generic(
> self,
> connector: str,
> *,
> options: Dict[str, str]
> ) -> None
> These generic APIs are escape hatches for connectors that do not need a
> dedicated Python helper or for advanced connector options. Other
> connector-specific APIs will follow the same design: expose typed and
> convenient Python parameters for commonly used connector and format options,
> keep connector-specific options aligned with the corresponding Flink
> connector and format options, and provide read_generic() / write_generic()
> for less common connectors or advanced options.
> Example:
> orders = pf.read_generic(
> "filesystem",
> schema={
> "order_id": DataType.string(),
> "amount": DataType.float64(),
> },
> options={
> "path": "s3:///warehouse/orders/",
> "format": "json",
> "json.ignore-parse-errors": "true"
> }
> )
> orders.write_generic(
> "filesystem",
> options={
> "path": "s3:///warehouse/filtered-orders/",
> "sink.rolling-policy.file-size": "128mb",
> "format": "json"
> }
> )
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
