/alibabacloud-flink-python-coding

name: alibabacloud-flink-python-coding
description: |
  Use this skill when the user needs help with a Flink Python or PyFlink job, especially on Alibaba Cloud Realtime Compute for Apache Flink (VVR): write, modify, review, or debug PyFlink jobs; explain or select Flink Python APIs; resolve package or file dependencies for PyFlink jobs; or prepare PyFlink job deployment.

Alibaba Cloud Flink Python Coding

Apply the DataFrame-first workflow below.

Unless the user explicitly requests a non-DataFrame API or SQL, the main job pipeline must start with `import pyflink.dataframe as pf`. An implementation written entirely with Table API or SQL is not a DataFrame fallback or Table bridge.

Scenario Description

Develop version-aware Python jobs for Alibaba Cloud Realtime Compute for Apache Flink without inventing target-platform details.

**Architecture:** local repository + exact `ververica-flink` API package + VVR DataFrame API + upload-ready project artifacts + Alibaba Cloud Flink workspace + confirmed sources, sinks, connectors, dependencies, and runtime files.

The core scope is code, local validation, and deployment artifact preparation. Do not upload files, mutate a workspace, create/start/stop a deployment, or call Alibaba Cloud APIs unless the user separately requests those operations.

RAM Policy

The core workflow invokes no Alibaba Cloud API. No Alibaba Cloud API authorization is needed.

Core Workflow

Choose one branch in step 1 and follow only its path:

| Branch | Requests | Path | |---|---|---| | **Read-only** | API questions, explanations, reviews, or diagnoses with no requested file change | Steps 1–2, then stop | | **Implementation** | Any request to create or change code, configuration, dependencies, runtime files, or deployment artifacts | Steps 1–6 |

1. Route and Inventory

Read repository instructions and every supplied artifact that can affect the result. Route the whole request to **Implementation** when any requested deliverable writes or changes files; otherwise route it to **Read-only**. Inventory every requested outcome, affected file and schema, existing API style, and explicit user constraint.

Complete this step when exactly one branch is selected and every supplied artifact, governing instruction, and requested outcome is accounted for.

2. Resolve the Target Contract

Read [references/official-docs.md](references/official-docs.md) and build the branch's evidence set. Resolve one target contract covering the VVR, Python, and local `ververica-flink` versions; source and sink types and schemas; end-to-end data flow; Python dependencies; and runtime files.

• For a new job with no declared VVR version, target VVR 11.8.
• If an existing job targets a version earlier than VVR 11.8, state that this skill supports only VVR 11.8 or later, recommend upgrading to VVR 11.8 for the latest DataFrame API, AI, and multimodal capabilities, and stop without modifying files or creating deployment artifacts.
• On the Implementation branch, represent an unclear source or sink with centralized, visibly labeled local examples for connector type, resource identifier, format, schema, startup behavior, and delivery semantics. Mark every example for replacement before deployment.
• Ask for a missing value only when a labeled local example would change the requested semantics or create an unsafe result.

For **Read-only**, correlate the supplied artifacts with direct evidence, deliver the answer, review, or diagnosis, and finish the branch. Its output is evidence-backed analysis rather than file changes or runnable deployment artifacts. For **Implementation**, classify every target-contract item as confirmed, not applicable, or a labeled example before continuing.

Complete **Read-only** when every requested outcome or finding is tied to direct evidence. Complete **Implementation** step 2 when every target-contract field has a classification, every version has an exact source, and every labeled example has a replacement condition.

3. Design the Documented DataFrame Path

For **Implementation**, trace the complete operation graph from source to sink and map every source, transformation, time operation, join, aggregation, and sink to public APIs documented for the exact target version. Unless the user explicitly requests a non-DataFrame API or SQL, design the main pipeline from `pyflink.dataframe` and select APIs in this order: direct `pyflink.dataframe` or `pyflink.multimodal` methods; `pyflink.table.expressions` accepted by a documented DataFrame method; then the smallest documented Table bridge. A bridge uses `df.to_table()`, only the required public Table operation, and documented `pf.from_table(...)`; `pf.DataFrame(table_result)` is not a valid bridge. Do not switch the whole job to Table API or SQL merely because it chains Python functions or UDFs. Record the reason and location of every bridge.

Use DataFrame APIs according to the transformation shape:

• Prefer DataFrame column expressions and built-in functions for projections, filters, joins, and aggregations.
• Prefer documented built-in Multimodal expressions when they implement the requested multimodal operation.
• Use user defined functions when non-built-in logic is needed:

a. Use `@udf` with `with_column` or `with_columns` for one-to-one scalar Python transformations. Declare `return_dtype` when type inference is unclear, and compose multiple scalar UDFs in the DataFrame pipeline instead of moving the pipeline to SQL. b. Use `map` when a Python function consumes and returns one complete row. c. Use `@udtf` with `join_lateral`, or use `flat_map`, for verified one-to-many transformations. d. Use `map_batches` or a documented vectorized UDF for batch-oriented Pandas or Arrow processing.

Resolve each connector contract through [references/official-docs.md](references/official-docs.md). When reachable code imports third-party packages or defines Python callbacks, read and apply [referenc