Realtime Compute for Apache Flink:Ikhtisar fitur

Gunakan data lokal

Saat pengembangan dan debugging, Anda dapat dengan cepat membuat DataFrame dari dict Python, daftar catatan (records), atau DataFrame Pandas. DataFrame yang digunakan sepanjang topik ini dibuat sebagai berikut: tabel pengguna users, tabel pesanan orders, tabel event events, dan tabel profil pengguna 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)

Baca data eksternal

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_generic(
    "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")

Proyeksi, penyaringan, dan kolom turunan

paid_orders berikut menyaring pesanan yang telah dibayar dari tabel pesanan orders dan mengelompokkannya berdasarkan jumlah:

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"})
)

Anda juga dapat menggunakan sintaks subscript untuk mereferensikan atau memilih kolom:

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

Penanganan nilai null dan NaN

drop_null / fill_null menangani SQL NULL, sedangkan drop_nan / fill_nan menangani NaN pada kolom floating-point.

clean_orders berikut merepresentasikan tabel pesanan setelah pembersihan kualitas:

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

Deduplikasi

Gunakan drop_duplicates untuk menghapus duplikat berdasarkan semua bidang atau subset tertentu.

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

Explode

Ketika suatu bidang berisi array, map, atau multiset, Anda dapat menggunakan explode untuk memperluas satu baris menjadi beberapa baris:

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

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

Operasi himpunan

Beberapa DataFrame yang kompatibel secara struktural dapat digabungkan dengan union, union_all, intersect, intersect_all, minus, dan 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"])
)

Agregasi

Gunakan group_by bersama dengan agg untuk melakukan agregasi berkelompok. Kolom pengelompokan tetap dipertahankan dalam DataFrame hasil.

user_summary berikut adalah tabel metrik pesanan yang diagregasi per pengguna, berisi jumlah pesanan, total jumlah, dan rata-rata jumlah:

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

Anda juga dapat melakukan agregasi pada seluruh tabel:

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

Join

join mendukung jenis join inner, left, right, full, dan outer.

Contoh berikut menggabungkan tabel pengguna users dengan tabel event events untuk menghasilkan tabel lebar enriched yang berisi nama pengguna, kota, dan event:

enriched = (
    users
    .join(
        events,
        on="user_id",
        how="left",
    )
    .select(
        "user_id",
        "name",
        "city",
        "event_id",
        "event",
    )
)

join_asof dapat digunakan untuk melakukan join tabel dimensi. Data sisi kanan harus berasal dari sumber tabel dimensi.

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",
)

Gunakan kueri SQL

Anda dapat menjalankan kueri SELECT dengan sql. SQL dapat mereferensikan variabel DataFrame dan user-defined function dalam cakupan saat ini tanpa registrasi manual.

Contoh berikut menggunakan SQL untuk memilih pengguna dengan pengeluaran tinggi dari ringkasan pesanan per pengguna user_summary dan menggabungkannya dengan tabel profil pengguna 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
""")

Konfigurasi penyedia

Contoh berikut menggunakan model bawaan yang disediakan oleh Flink AI Service. Hanya parameter task yang wajib; Anda tidak perlu mengonfigurasi endpoint atau api_key. Flink AI Service harus diaktifkan sebelum Anda dapat menggunakan model bawaan. Untuk informasi lebih lanjut, lihat Flink AI Service (built-in models).

pf.set_model_provider("chat", pf.OpenAICompatProvider(task="chat/completions"))
pf.set_model_provider("embedding", pf.OpenAICompatProvider(task="embeddings"))

Pemanggilan generik

predict digunakan untuk pemanggilan LLM generik. Bidang output dapat ditentukan dengan output_type.

Sebagai contoh, questions berikut merepresentasikan pertanyaan tiket layanan pelanggan, dan answers adalah tabel hasil dengan kolom tambahan output yang berisi jawaban dari 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",
    model="qwen3.6-plus",
    system_prompt="You are a helpful assistant.",
    user_prompt="Answer the customer support question concisely. Return only the answer.",
    temperature=0.2,
    config={"max-concurrent-operations": "20"},
)

Fungsi AI bawaan

Anda dapat menggunakan fungsi AI bawaan untuk tugas umum. Fungsi AI bawaan menambahkan kolom hasil ke DataFrame asli; untuk detailnya, lihat AI / LLM functions.

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

# sentiment analysis
sentiment = questions.llm.ai_sentiment("question", provider="chat", model="qwen3.6-plus")

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

# summarize
summaries = questions.llm.ai_summarize(
    "question",
    max_length=50,
    provider="chat",
    model="qwen3.6-plus",
)

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

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

Fungsi AI di atas juga mendukung caching hasil inferensi berulang dalam tabel Fluss melalui parameter cache_table dan cache_key, yang mengurangi biaya token. Untuk detailnya, lihat Generic invocation.

Pencarian vektor

Anda dapat menggunakan vector_search untuk terhubung ke database vektor seperti Milvus dan menjalankan pencarian vektor.

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(),
    },
    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"],
)

Gunakan fungsi sink bawaan

Anda dapat menggunakan fungsi sink bawaan DataFrame API untuk menulis data ke sistem umum.

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",
)

Untuk fungsi sink lainnya, lihat Data sinks.

Gunakan konektor kustom

Anda dapat menggunakan fungsi write_generic untuk menulis data melalui konektor atau konektor kustom lain yang didukung.

Parameter connector adalah pengenal konektor yang akan digunakan, dan options meneruskan parameter WITH dari konektor tersebut.

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

Jika konektor memerlukan batasan primary key, gunakan parameter primary_key:

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

Tulis ke tabel Catalog

Jika tabel target sudah ada di Catalog, Anda juga dapat menggunakan write_catalog_table untuk menulis langsung:

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

Tulis ke beberapa target

Saat Anda perlu menulis pipeline komputasi yang sama ke beberapa target, gunakan create_statement_set untuk membuat StatementSet, teruskan ke parameter statement_set pada setiap metode write, lalu eksekusi secara bersamaan:

statement_set = pf.create_statement_set()

enriched.write_kafka(
    "broker-1:9092,broker-2:9092",
    topic="user_summary",
    format="json",
    statement_set=statement_set,
)
enriched.write_json(
    "oss://my-bucket/user_summary/",
    statement_set=statement_set,
)

statement_set.execute()