Dian Fu created FLINK-40192:
-------------------------------
Summary: 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
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)
