Flink Python Job Submission

1. Scope

Create a Python deployment from ready PyFlink code: upload its artifacts, fill its deployment configuration, and verify the saved result.

For SQL or existing deployment operations (start, stop, update, status, logs, metrics, or deletion), use alibabacloud-flink-workspace-ops. If available, load that skill; otherwise point the user to installing it. Code development belongs to alibabacloud-flink-python-coding.

2. Installation

Pre-check: Aliyun CLI >= 3.3.3 required

[MUST] Verify: aliyun version — must be >= 3.3.3.

• First install or major upgrade: /bin/bash -c "$(curl -fsSL --connect-timeout 10 --max-time 120 https://aliyuncli.alicdn.com/setup.sh)"
• Routine update (CLI >= 3.3.5): aliyun upgrade — prefer this built-in self-update over re-running the install script.
• See references/cli-installation-guide.md for full installation instructions.

Before the first non-version aliyun command, apply the conversation-scoped CLI initialization in section 7.

Python SDK dependencies:

pip3 install alibabacloud_tea_openapi alibabacloud_credentials \
  alibabacloud_tea_util alibabacloud_openapi_util \
  'alibabacloud_foasconsole20211028==2.2.1'

3. Environment Variables

4. Authentication

Pre-check: Alibaba Cloud Credentials Required

Security Rules:

• NEVER read, echo, or print AK/SK values (e.g., echo $ALIBABA_CLOUD_ACCESS_KEY_ID is FORBIDDEN)
• NEVER ask the user to input AK/SK directly in the conversation or command line
• NEVER use aliyun configure set with literal credential values
• ONLY use aliyun configure list to check credential status

aliyun configure list

Check the output for a valid profile (AK, STS, or OAuth identity).

If no valid profile exists, STOP here.

1. Obtain credentials from Alibaba Cloud Console
2. Configure credentials outside of this session (via aliyun configure in terminal or environment variables in shell profile)
3. Return and re-run after aliyun configure list shows a valid profile

5. RAM Policy

Creation permissions are listed below; the reference also includes workspace operations. See references/ram-policies.md for full list.

[MUST] Permission Failure Handling: When any command or API call fails due to permission errors at any point during execution, follow this process:

1. Read references/ram-policies.md to get the full list of permissions required by this SKILL
2. Use ram-permission-diagnose skill to guide the user through requesting the necessary permissions
3. Pause and wait until the user confirms that the required permissions have been granted

6. Create a Deployment

1. Resolve the exact region, workspace, and namespace from the user or explicit project configuration. If any is missing, ask for it and pause until the user supplies all three. This gate applies to every cloud discovery call, including workspace listing and engine queries; a partially known location stays pending.

2. Follow storage routing to discover the bound storage and console URL, upload the main artifact and dependencies, and verify each filename and size. Resolve exact-name conflicts with the user.

3. Use scripts/flink_python_submit.py create-deployment --help for the argument schema and deployment configuration for Python-specific choices. Present the resolved creation parameters together, including defaults that will be sent, and obtain approval before creation.

4. Create the deployment and read it back with get-deployment. Compare its Python artifact, dependencies, engine, queue, resources, and flinkConf with the approved values, allowing equivalent server normalization. Return the deployment ID, location, and a clickable VVP configuration link:

   <consoleUrl>/web/<workspace>/zh/#/workspaces/<workspace>/namespaces/<namespace>/operations/stream/<deployment-id>/configuration
   

Use the exact consoleUrl from storage discovery. Creation is complete when the saved configuration matches; runtime state is outside this workflow.

7. CLI Initialization and Observability

For each skill invocation, generate a fresh random 32-character lowercase hex session ID and reuse it throughout that invocation. Before the first cloud API call, read manifest.json and validate its version as a non-empty string without whitespace. If the file is missing, unreadable, invalid JSON, or has an invalid version, stop and report the error; do not guess a version or make a cloud call.

Reuse successful CLI initialization already visible in the conversation. Before the first non-version aliyun command, run each pending command once:

aliyun configure set --auto-plugin-install true
aliyun plugin update

Retry only a failed initialization command. Keep this state in the conversation. For cloud OpenAPI CLI commands, include:

--user-agent 'AlibabaCloud-Agent-Skills/alibabacloud-flink-python-job-submission/{session-id} skill-version/{version}'

Omit that flag for ossutil and local commands (configure, plugin, version). For SDK script calls, supply the same session ID through the environment:

SKILL_SESSION_ID={session-id} python3 scripts/flink_python_submit.py create-deployment ...

The SDK script validates the manifest before constructing either cloud client and derives the same versioned user-agent from it.