Dian Fu created FLINK-40197:
-------------------------------
Summary: Add sql bindings to DataFrame API
Key: FLINK-40197
URL: https://issues.apache.org/jira/browse/FLINK-40197
Project: Flink
Issue Type: Sub-task
Components: API / Python
Reporter: Dian Fu
Fix For: 2.4.0
def sql(
query: str,
*,
auto_bind: bool = True,
**bindings,
) -> DataFrame
Run a SQL SELECT query and return its result as a DataFrame, so SQL and the
DataFrame API mix freely. DataFrames referenced in the query are exposed as
temporary views, and @pf.udf / @pf.udtf functions as temporary functions; both
are registered only for the duration of the call and dropped afterwards.
There are two ways to make Python objects visible to the query:
* Auto-bind (default, auto_bind=True): sql scans the caller's local and global
variables for DataFrame and UDF/UDTF objects and registers each under its
Python variable name. Auto-bind is best-effort — it warns and skips names that
are not valid SQL identifiers or that collide with an existing
table/view/function, and it never shadows permanent catalog objects or built-in
functions.
* Explicit bindings: pass keyword arguments to choose the SQL name yourself,
e.g. pf.sql("SELECT * FROM s", s=df). Explicit bindings are strict (conflicts
raise ValueError), take precedence over auto-bind on name collisions, and are
required to intentionally shadow a catalog table/view or a built-in function.
Only SELECT queries are supported (no INSERT / DDL). The returned DataFrame can
be further transformed with the DataFrame API.
Example:
{code:python}
import pyflink.dataframe as pf
df1 = pf.from_dict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
df2 = pf.from_dict({"a": [1, 2, 3], "c": ["p", "q", "r"]})
# Auto-bind: df1 / df2 are registered under their variable names
joined = pf.sql("SELECT df1.a, b, c FROM df1 JOIN df2 ON df1.a = df2.a")
# UDFs are auto-bound too
@pf.udf
def add_one(x: int) -> int:
return x + 1
incremented = pf.sql("SELECT add_one(a) AS a1 FROM df1")
# Explicit bindings: pick the SQL names, turn off scanning
result = pf.sql(
"SELECT * FROM src WHERE a > 1",
auto_bind=False,
src=df1,
)
# Mix SQL and the DataFrame API
pf.sql("SELECT a, b FROM df1").filter(pf.col("a") > 1).to_pandas()
{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
