---
name: "pyflink-realtime-event-processing"
description: "Develop scalable real-time stream processing applications using Python and Apache Flink."
metadata:
  author: "Apache Software Foundation"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "write pyflink table api job"
  - "define pyflink scalar udf"
  - "read kafka stream with pyflink"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# PyFlink Python Stream & Table API Pipelines

Writes PyFlink streaming jobs: using Table API and DataStream API, defining Python User-Defined Functions (UDFs / UDAFs), deserializing JSON/Kafka streams, windowed aggregations, and deploying PyFlink on YARN/K8s.

## When to Invoke
Invoke this skill whenever you need to:
- write pyflink table api job
- define pyflink scalar udf
- read kafka stream with pyflink

## Execution SOP & Rules
1. Initialize environment: env = StreamExecutionEnvironment.get_execution_environment(); t_env = StreamTableEnvironment.create(env).
2. Register Kafka source table with DDL: CREATE TABLE source_table (...) WITH ('connector' = 'kafka', ...).
3. Register scalar Python UDF using @udf(result_type=DataTypes.STRING()) decorator.
4. Execute pipeline: t_env.execute_sql('INSERT INTO sink_table SELECT ... FROM source_table').

## Verification Checklist
- Run local linting, syntax validation, and type-checks before reporting completion.
- Ensure all created or modified files adhere to workspace boundaries and avoid uncommitted secrets.
- Verify tests or build commands exit cleanly with return code 0.