Feature overview

Use local data

For development and debugging, you can build a DataFrame from a Python dictionary, a list of records, or a Pandas DataFrame. The following code creates the DataFrames used in this topic: the user table users, the order table orders, the event table events, and the user profile table profiles.

import pandas as pd
import pyflink.dataframe as pf
from pyflink.dataframe import col, lit, udf, udtf, DataType

# User table
users = pf.from_dict({
    "user_id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"],
    "city": ["Hangzhou", "Shanghai", "Beijing"],
})

# Order fact table
orders = pf.from_records(
    [
        (1001, 1, 99.9, "PAID"),
        (1002, 2, 35.5, "CREATED"),
        (1003, 1, 188.0, "PAID"),
        (1004, 3, 88.8, None),
    ],
    schema=["order_id", "user_id", "amount", "status"],
)

# User behavior event table
events = pf.from_records(
    [
        {"event_id": "e1", "user_id": 1, "event": "login"},
        {"event_id": "e2", "user_id": 2, "event": "pay"},
    ],
    schema=["event_id", "user_id", "event"],
)

# User profile table
profiles = pf.from_records(
    [
        (1, "gold"),
        (2, "silver"),
        (3, "gold"),
    ],
    schema=["user_id", "member_level"],
)

# You can also build from pandas DataFrame
pdf = pd.DataFrame({"id": [1, 2], "score": [0.8, 0.95]})
df_from_pandas = pf.from_pandas(pdf)

Read external data

kafka_events = pf.read_kafka(
    "broker-1:9092,broker-2:9092",
    topic="user_events",
    group_id="df-api-demo",
    startup_mode="earliest-offset",
    format="json",
    schema={
        "user_id": DataType.int64(),
        "event": DataType.string(),
        "event_time": DataType.timestamp(3),
        "payload": DataType.string(),
    },
)

source = pf.read_custom(
    "datagen",
    schema={
        "id": DataType.int64(),
        "name": DataType.string(),
    },
    options={
        "number-of-rows": "1000",
        "fields.id.kind": "sequence",
        "fields.id.start": "1",
        "fields.id.end": "1000",
    },
)

pf.create_catalog(
    "lake",
    {
        "type": "paimon",
        "warehouse": "oss://my-bucket/warehouse",
    },
)

pf.use_catalog("lake")
pf.use_database("commerce")

historical_orders = pf.read_catalog_table("orders")

Projection, filtering, and derived columns

The following paid_orders example filters paid orders from the orders table and categorizes them by amount:

paid_orders = (
    orders
    .select("order_id", "user_id", "amount", "status")
    .filter("status = 'PAID' AND amount > 0")
    .with_columns(
        amount_yuan=col("amount"),
        amount_level=(
            (col("amount") >= 100).then("high", "normal")
        ),
    )
    .drop("status", "amount")
    .rename({"amount_yuan": "amount"})
)

You can also use bracket syntax to reference or select columns:

amount_expr = paid_orders["amount"]
simple_view = paid_orders[["order_id", "user_id", "amount"]]
large_orders = paid_orders[col("amount") > 100]

Handle null values and NaN

drop_null / fill_null handle SQL NULL values, and drop_nan / fill_nan handle NaN in floating-point columns.

The following clean_orders represents the order table after data cleansing:

clean_orders = (
    orders
    .drop_null(subset=["order_id", "user_id"])
    .fill_null("UNKNOWN", subset=["status"])
    .fill_nan(0.0, subset=["amount"])
)

Deduplication

Use drop_duplicates to deduplicate by all columns or by a subset of columns.

unique_events = events.drop_duplicates(
    subset=["event_id"],
)

Explode

If a column contains an array, a Map, or a Multiset, you can use explode to expand one row into multiple rows:

order_items = pf.from_records(
    [
        (1001, ["phone", "case"]),
        (1002, ["book"]),
    ],
    schema=["order_id", "items"],
)

item_rows = order_items.explode("items", "item")

Set operations

Multiple DataFrames with compatible schemas can be combined using union, union_all, intersect, intersect_all, minus, and minus_all.

all_paid_orders = (
    paid_orders.select("order_id", "user_id", "amount")
    .union_all(historical_orders.select("order_id", "user_id", "amount"))
    .drop_duplicates(subset=["order_id"])
)

Aggregation

Use group_by and agg to aggregate data by groups. You must explicitly include the grouping columns in the result.

The following user_summary DataFrame summarizes order metrics by user, including the order count, total amount, and average amount:

user_summary = (
    clean_orders
    .filter("status = 'PAID'")
    .group_by("user_id")
    .agg(
        col("user_id"),
        col("order_id").count.alias("order_count"),
        col("amount").sum.alias("total_amount"),
        col("amount").avg.alias("avg_amount"),
    )
)

To aggregate over the entire table, use select directly:

overall = clean_orders.select(
    col("order_id").count.alias("order_count"),
    col("amount").sum.alias("total_amount"),
)

Join

join supports inner, left, right, full, and outer join types.

The following example joins the users table with the events table to create an enriched wide table containing user names, cities, and behavior events:

events_renamed = events.rename("user_id", "event_user_id")

enriched = (
    users
    .join(
        events_renamed,
        left_on="user_id",
        right_on="event_user_id",
        how="left",
    )
    .select(
        "user_id",
        "name",
        "city",
        "event_id",
        "event",
    )
)

join_asof can be used to perform a dimension table join. The right table must read from a dimension table data source.

dim_users = pf.read_hologres(
    endpoint="hgpostcn-cn-xxx-cn-hangzhou.hologres.aliyuncs.com:80",
    db_name="commerce",
    table_name="dim_user_profile",
    username="${secret_values.hg_user}",
    password="${secret_values.hg_password}",
    schema={
        "user_id": DataType.int64(),
        "member_level": DataType.string(),
        "city": DataType.string(),
    },
    primary_key="user_id",
    binlog=False,
    cache="LRU",
)

events_with_profile = events.join_asof(
    dim_users,
    by="user_id",
    how="left",
)

SQL queries

Use sql to run SELECT queries. Table names in the SQL statement automatically bind to Python DataFrame variables with the same name in the current scope, so you do not need to register them manually.

The following example uses SQL to select users with high cumulative spending from the user_summary table and joins it with the user profile table profiles:

top_users = pf.sql("""
    SELECT
        user_summary.user_id,
        profiles.member_level,
        user_summary.order_count,
        user_summary.total_amount
    FROM user_summary
    LEFT JOIN profiles
        ON user_summary.user_id = profiles.user_id
    WHERE user_summary.total_amount >= 200
""")

Configure providers

The following example registers two model providers: one for text-based tasks and another for vectorization. Use different names to distinguish between providers.

API_KEY="sk-..."

pf.set_provider(
    "chat",
    pf.OpenAICompatProvider(
        endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
        api_key=API_KEY,
        model="qwen-plus",
        system_prompt="You are a helpful assistant.",
        temperature=0.2,
    ),
)

pf.set_provider(
    "embedding",
    pf.OpenAICompatProvider(
        endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings",
        api_key=API_KEY,
        model="text-embedding-v4",
    ),
)

Generic invocation

Use predict for generic LLM invocation. It accepts single-column input, and you can specify the output field with output_type.

In the following example, the questions DataFrame contains customer support ticket text. The resulting answers DataFrame includes an answer column appended by the model.

questions = pf.from_records(
    [
        (1, "Why hasn't my order been shipped yet? My order number is 1. Charlie"),
        (2, "How do I apply for a refund? My email is bob@example.com. Order number: 2."),
    ],
    schema=["ticket_id", "question"],
)

answers = questions.llm.predict(
    "question",
    provider="chat",
    output_type={"answer": DataType.string()},
    config={
        "user-prompt": "Answer the customer support question concisely. Return only the answer.",
    },
)

Built-in AI functions

Built-in AI functions handle common tasks and append result columns to the original DataFrame. For details on the output format, see AI / LLM Functions.

# classify
classified = questions.llm.ai_classify(
    "question",
    labels=["logistics", "refund", "account", "other"],
    provider="chat",
)

# sentiment analysis
sentiment = questions.llm.ai_sentiment("question", provider="chat")

# information extraction, schema described as JSON string
extracted = questions.llm.ai_extract(
    "question",
    '{"order_id":"STRING","intent":"STRING"}',
    provider="chat",
)

# summarize
summaries = questions.llm.ai_summarize(
    "question",
    max_length=50,
    provider="chat",
)

# mask sensitive data
masked = questions.llm.ai_mask(
    "question",
    entities=["PERSON", "PHONE", "EMAIL"],
    provider="chat",
)

# embedding
embeddings = questions.llm.ai_embed(
    "question",
    dimension=1024,
    provider="embedding",
)

Vector search

Use vector_search to connect to a vector database such as Milvus and perform vector search.

documents = pf.read_milvus(
    endpoint="http://milvus.example.com",
    username="${secret_values.milvus_user}",
    password="${secret_values.milvus_password}",
    database_name="commerce",
    collection_name="support_docs",
    schema={
        "doc_id": DataType.int64(),
        "embedding": DataType.list(DataType.float32()),
        "title": DataType.string(),
    },
    columns=["doc_id", "embedding", "title"],
    search_metric="COSINE",
)

matched_docs = embeddings.llm.vector_search(
    documents,
    column_to_search="embedding",
    column_to_query="embedding",
    top_k=3,
    output_columns=["doc_id", "doc_embedding", "doc_title", "score"],
)

Built-in sink functions

Use the built-in sink functions to write data to common systems.

enriched.write_kafka(
    "broker-1:9092,broker-2:9092",
    topic="user_summary",
    format="json",
    key_format="json",
    key_fields=["user_id"],
    delivery_guarantee="at-least-once",
)

For more sink functions, see Data Output.

Custom connectors

Use the write_custom function to write data through other supported connectors or a custom connector.

The connector parameter specifies the connector's identifier, and the options parameter passes WITH options to the connector.

enriched.write_custom(
    "blackhole",
    options={
        "sink.parallelism": "4",
    },
)

If the connector requires a primary key constraint, you can pass the primary_key parameter:

enriched.write_custom(
    "my-custom-sink",
    primary_key="user_id",
    options={
        "endpoint": "example.com:9000",
        "format": "json",
    },
)

Write to Catalog tables

If the target table already exists in a Catalog, you can use write_catalog_table to write to it directly:

matched_docs.write_catalog_table(
    "lake.commerce.support_ticket_matches",
    overwrite=True,
)