[
https://issues.apache.org/jira/browse/FLINK-40192?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18126211#comment-18126211
]
Dian Fu commented on FLINK-40192:
---------------------------------
[~beetleyibo] Sure. Thanks for working on it!
> Add as-of join support to DataFrame API
> ---------------------------------------
>
> Key: FLINK-40192
> URL: https://issues.apache.org/jira/browse/FLINK-40192
> Project: Flink
> Issue Type: Sub-task
> Components: API / Python
> Reporter: Dian Fu
> Assignee: Bettle
> Priority: Major
>
> def join_asof(
> self,
> other: "DataFrame",
> *,
> how: str = "inner",
> on: Union[str, Expression] = None,
> by: Union[str, Expression, List[Union[str, Expression]]] = None,
> left_by: Union[str, Expression, List[Union[str, Expression]]] = None,
> right_by: Union[str, Expression, List[Union[str, Expression]]] = None,
> ) -> "DataFrame"
> Perform an ASOF join with a dimension table at a time attribute. For each
> left row, the matching dimension row is selected by key as of the row's time
> attribute.
> * other — the right DataFrame (the dimension table).
> * how — "inner" or "left".
> * on — the ASOF time attribute. Provide it as pf.proctime() (or a proctime
> column on the left DataFrame) for a processing-time lookup, or as a rowtime
> (event-time / watermark) column for an event-time temporal join. The
> time-attribute column must not also be used as a join key.
> * by — join key(s) shared by both sides. Accepts the same shapes as join: a
> column name, an Expression, or a list thereof. The resulting condition must
> contain at least one equi-predicate (AND-combined); additional non-equality
> conditions may be added, but a purely non-equi or OR-based condition is
> rejected.
> * left_by / right_by — join keys when the key columns have different names on
> each side; supply both together instead of by.
> Example:
> {code:python}
> # Lookup join on an equality key with explicit processing time
> enriched = probe_df.join_asof(
> dim_df,
> on=pf.proctime(),
> by=pf.col("id") == pf.col("d_id"),
> )
> # Explicit processing-time attribute, left outer join
> enriched = probe_df.join_asof(
> dim_df,
> on=pf.proctime(),
> by=pf.col("id") == pf.col("d_id"),
> how="left",
> )
> # Event-time temporal join: match the dimension version as of the
> # left row's rowtime (watermark) attribute
> enriched = probe_df.join_asof(
> dim_df,
> on=pf.col("event_time"),
> by=pf.col("id") == pf.col("d_id"),
> )
> # Time attribute given as a proctime column, with differing key names
> enriched = probe_df.join_asof(
> dim_df,
> on=pf.col("proc_time"),
> left_by="id",
> right_by="d_id",
> )
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
