### Complete Kafka Source and Sink Example

Source: https://github.com/apache/flink/blob/master/flink-python/docs/user_guide/table/connectors.md

A full implementation example demonstrating environment setup, dependency configuration, DDL execution, and data processing.

```python
from pyflink.table import TableEnvironment, EnvironmentSettings

def log_processing():
    env_settings = EnvironmentSettings.in_streaming_mode()
    t_env = TableEnvironment.create(env_settings)
    # specify connector and format jars
    t_env.get_config().set("pipeline.jars", "file:///my/jar/path/connector.jar;file:///my/jar/path/json.jar")

    source_ddl = """
            CREATE TABLE source_table(
                a VARCHAR,
                b INT
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'source_topic',
              'properties.bootstrap.servers' = 'kafka:9092',
              'properties.group.id' = 'test_3',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
            """

    sink_ddl = """
            CREATE TABLE sink_table(
                a VARCHAR
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'sink_topic',
              'properties.bootstrap.servers' = 'kafka:9092',
              'format' = 'json'
            )
            """

    t_env.execute_sql(source_ddl)
    t_env.execute_sql(sink_ddl)

    t_env.sql_query("SELECT a FROM source_table") \
        .execute_insert("sink_table").wait()


if __name__ == '__main__':
    log_processing()
```

--------------------------------

### Start Flink SQL Client using Local Installation

Source: https://github.com/apache/flink/blob/master/docs/content/docs/getting-started/quickstart-sql.md

Starts the Flink SQL Client using a local installation. This command-line tool allows interactive SQL query submission and result visualization.

```bash
$ ./bin/sql-client.sh
```

--------------------------------

### Bash Script for Flink Gradle Quickstart Setup

Source: https://github.com/apache/flink/blob/master/docs/content/docs/dev/configuration/overview.md

This bash script automates the setup of a Flink Gradle quickstart project. It downloads and executes a setup script from the Flink website, passing the Flink and Scala versions as arguments to configure the project.

```bash
bash -c "$(curl https://flink.apache.org/q/gradle-quickstart.sh)" -- {{< version >}} {{< scala_version >}}
```

--------------------------------

### Install Flink and Dependencies

Source: https://github.com/apache/flink/blob/master/docs/content/docs/sql/materialized-table/quickstart.md

Commands to extract the Flink binary distribution and install the required test-filesystem connector.

```bash
tar -xzf flink-*.tgz
cp flink-table-filesystem-test-utils-{VERSION}.jar flink-*/lib/
```

--------------------------------

### Run the PyFlink Word Count Example

Source: https://github.com/apache/flink/blob/master/flink-python/docs/cookbook/word_count.md

Commands to start a socket server, execute the Python script, and provide input data.

```bash
nc -lk 9999
```

```bash
python word_count.py
```

```text
hello world
hello flink
world of streaming
```

--------------------------------

### Start Flink Services

Source: https://github.com/apache/flink/blob/master/docs/content/docs/sql/materialized-table/quickstart.md

Shell commands to initialize the Flink cluster, SQL gateway, and SQL client.

```bash
./bin/start-cluster.sh
./bin/sql-gateway.sh start
./bin/sql-client.sh gateway --endpoint http://127.0.0.1:8083
```

--------------------------------

### Start Flink SQL Client Locally

Source: https://github.com/apache/flink/blob/master/docs/content/docs/getting-started/local_installation.md

Command to start an interactive Flink SQL client session locally. Requires Flink to be installed and the cluster to be running.

```bash
$ ./bin/sql-client.sh

```

--------------------------------

### Show Catalogs Examples

Source: https://github.com/apache/flink/blob/master/docs/content/docs/sql/reference/utility/show.md

Examples of listing catalogs and using pattern matching.

```sql
show catalogs;
show catalogs like '%log1';
-- show catalogs ilike '%log1';
-- show catalogs ilike '%LOG1';
```

--------------------------------

### Get Overview of All Applications

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves a summary of all applications currently managed by Flink. This includes basic details like duration, start and end times, and status.

```APIDOC
## GET /applications/overview

### Description
Returns an overview over all applications.

### Method
GET

### Endpoint
/applications/overview

### Parameters
#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200 OK)
- **applications** (array) - An array of application details.
  - **duration** (integer) - The duration of the application in milliseconds.
  - **end-time** (integer) - The timestamp when the application ended.
  - **id** (any) - The unique identifier of the application.
  - **jobs** (object) - An object where keys are job IDs and values are integers (likely representing some job metric).
  - **name** (string) - The name of the application.
  - **start-time** (integer) - The timestamp when the application started.
  - **status** (string) - The current status of the application.

#### Response Example
{
  "applications": [
    {
      "duration": 12345,
      "end-time": 1678886400000,
      "id": "app_123",
      "jobs": {
        "job_abc": 100
      },
      "name": "My Flink App",
      "start-time": 1678880000000,
      "status": "RUNNING"
    }
  ]
}
```

--------------------------------

### Get WebUI Configuration

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the configuration settings for the Flink Web UI.

```APIDOC
## GET /config

### Description
Returns the configuration of the WebUI.

### Method
GET

### Endpoint
/config

### Response
#### Success Response (200 OK)
- **features** (object)
  - **web-cancel** (boolean)
  - **web-history** (boolean)
  - **web-rescale** (boolean)
  - **web-submit** (boolean)
- **flink-revision** (string)
- **flink-version** (string)
- **refresh-interval** (integer)
- **timezone-name** (string)
- **timezone-offset** (integer)
```

--------------------------------

### GET /jobmanager/environment

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the environment details of the JobManager, including JVM information and classpath.

```APIDOC
## GET /jobmanager/environment

### Description
Returns the jobmanager environment.

### Method
GET

### Endpoint
/jobmanager/environment

### Parameters

### Request Body

### Request Example
```json
{}
```

### Response
#### Success Response (200 OK)
- **classpath**: array of strings
- **jvm**: object
  - **arch** (string)
  - **options**: array of strings
  - **version** (string)

#### Response Example
```json
{
  "classpath": [
    "/path/to/jar1.jar",
    "/path/to/jar2.jar"
  ],
  "jvm": {
    "arch": "amd64",
    "options": [
      "-Xmx1024m",
      "-Djava.net.preferIPv4Stack=true"
    ],
    "version": "1.8.0_291"
  }
}
```
```

--------------------------------

### GET /jobmanager/metrics

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Provides access to JobManager metrics. You can optionally specify which metrics to retrieve using the 'get' query parameter.

```APIDOC
## GET /jobmanager/metrics

### Description
Provides access to job manager metrics.

### Method
GET

### Endpoint
/jobmanager/metrics

### Parameters
#### Query Parameters
- **get** (string) - Optional - Comma-separated list of string values to select specific metrics.

### Request Body

### Request Example
```json
{}
```

### Response
#### Success Response (200 OK)
- **type**: any

#### Response Example
```json
{
  "metric.name.1": 123.45,
  "metric.name.2": 678.90
}
```
```

--------------------------------

### Build Docs Locally with Hugo

Source: https://github.com/apache/flink/blob/master/docs/README.md

Builds the Flink documentation locally after installing Hugo and running a setup script. The documentation will be accessible at http://localhost:1313/.

```shell
#!/bin/bash
./setup_hugo.sh
./build_docs.sh
```

--------------------------------

### GET /jobs/overview

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves a summary of all jobs currently managed by Flink. This includes basic information for each job.

```APIDOC
## GET /jobs/overview

### Description
Returns an overview over all jobs.

### Method
GET

### Endpoint
/jobs/overview

### Response
#### Success Response (200 OK)
- **jobs** (array) - An array of job details.
  - **duration** (integer)
  - **end-time** (integer)
  - **jid** (any)
  - **jobType** (string, enum: ["BATCH", "STREAMING"])
  - **last-modification** (integer)
  - **name** (string)
  - **pending-operators** (integer)
  - **schedulerType** (string, enum: ["Default", "Adaptive", "AdaptiveBatch"])
  - **start-time** (integer)
  - **state** (string, enum: ["INITIALIZING", "CREATED", "RUNNING", "FAILING", "FAILED", "CANCELLING", "CANCELED", "FINISHED", "RESTARTING", "SUSPENDED", "RECONCILING"])
  - **tasks** (object) - Additional properties are integers representing task counts.

#### Response Example
{
  "jobs": [
    {
      "duration": 12345,
      "end-time": 1678886400000,
      "jid": "some-job-id",
      "jobType": "STREAMING",
      "last-modification": 1678886400000,
      "name": "MyStreamingJob",
      "pending-operators": 0,
      "schedulerType": "Default",
      "start-time": 1678886300000,
      "state": "RUNNING",
      "tasks": {
        "total": 10,
        "running": 5
      }
    }
  ]
}
```

--------------------------------

### Start Flink Session Cluster on Docker (Shell)

Source: https://github.com/apache/flink/blob/master/docs/content/docs/deployment/resource-providers/standalone/docker.md

Commands to set up Flink properties, create a Docker network, and launch JobManager and TaskManager containers for a Flink Session cluster. Requires Docker to be installed.

```shell
$ FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager"
$ docker network create flink-network

$ docker run \
    --rm \
    --name=jobmanager \
    --network flink-network \
    --publish 8081:8081 \
    --env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \
    flink:{{< stable >}}{{< version >}}-scala{{< scala_version >}}{{< /stable >}}{{< unstable >}}latest{{< /unstable >}} jobmanager

$ docker run \
    --rm \
    --name=taskmanager \
    --network flink-network \
    --env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \
    flink:{{< stable >}}{{< version >}}-scala{{< scala_version >}}{{< /stable >}}{{< unstable >}}latest{{< /unstable >}} taskmanager
```

--------------------------------

### Create New User Guide Page

Source: https://github.com/apache/flink/blob/master/flink-python/docs/README.md

This snippet outlines the steps to create a new user guide page in RST format. It includes creating the `.rst` file, adding it to the `index.rst` toctree, generating translation stubs, and optionally populating the translation file.

```bash
vi user_guide/my_new_page.rst
```

```rst
.. toctree::
    :maxdepth: 2

    existing_page
    my_new_page        <-- add here
```

```bash
make gettext
sphinx-intl update -p _build/gettext -l zh
```

--------------------------------

### Get Task Manager Details

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Fetches detailed information about a specific Task Manager identified by its ID.

```APIDOC
## GET /taskmanagers/:taskmanagerid

### Description
Retrieves detailed information about a specific Task Manager.

### Method
GET

### Endpoint
/taskmanagers/:taskmanagerid

### Parameters
#### Path Parameters
- **taskmanagerid** (string) - Required - 32-character hexadecimal string that identifies a task manager.

### Response
#### Success Response (200)
- **allocatedSlots** (array) - List of slot information for the task manager.
- **assignedTasks** (integer) - The number of tasks assigned to this task manager.
- **blocked** (boolean) - Indicates if the task manager is blocked.
- **dataPort** (integer) - The data port of the task manager.
- **freeResource** (object) - Resource profile of the free resources.
- **freeSlots** (integer) - The number of free slots available on the task manager.
- **hardware** (object) - Hardware description of the task manager.
- **id** (any) - The ID of the task manager.
- **jmxPort** (integer) - The JMX port of the task manager.
- **memoryConfiguration** (object) - Memory configuration details for the task manager.
- **metrics** (object) - Metrics information for the task manager.

#### Response Example
{
  "allocatedSlots": [
    {
      "assignedTasks": 1,
      "jobId": "some-job-id",
      "resource": {}
    }
  ],
  "assignedTasks": 5,
  "blocked": false,
  "dataPort": 6123,
  "freeResource": {},
  "freeSlots": 2,
  "hardware": {
    "cpuCores": 4,
    "freeMemory": 1073741824,
    "managedMemory": 268435456,
    "physicalMemory": 8589934592
  },
  "id": "some-task-manager-id",
  "jmxPort": 9091,
  "memoryConfiguration": {
    "frameworkHeap": 134217728,
    "frameworkOffHeap": 134217728,
    "jvmMetaspace": 268435456,
    "jvmOverhead": 134217728,
    "managedMemory": 268435456,
    "networkMemory": 134217728,
    "taskHeap": 536870912,
    "taskOffHeap": 134217728,
    "totalFlinkMemory": 1342177280,
    "totalProcessMemory": 1610612736
  },
  "metrics": {
    "directCount": 0,
    "directMax": 0,
    "directUsed": 0,
    "garbageCollectors": [],
    "heapCommitted": 134217728,
    "heapMax": 1073741824,
    "heapUsed": 53687091,
    "mappedCount": 0,
    "mappedMax": 0,
    "mappedUsed": 0,
    "nettyShuffleMemoryAvailable": 0,
    "nettyShuffleMemorySegmentsAvailable": 0,
    "nettyShuffleMemorySegmentsTotal": 0,
    "nettyShuffleMemorySegmentsUsed": 0,
    "nettyShuffleMemoryTotal": 0,
    "nettyShuffleMemoryUsed": 0
  }
}
```

--------------------------------

### Get Job Rescales Summary

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves a summary of rescaling operations for a specific job, including statistics and counts.

```APIDOC
## GET /jobs/:jobid/rescales/summary

### Description
Return job rescales summary.

### Method
GET

### Endpoint
/jobs/:jobid/rescales/summary

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.

### Response
#### Success Response (200 OK)
- **completedRescalesDurationStatsInMillis** (object) - Statistics summary for completed rescale operations.
- **failedRescalesDurationStatsInMillis** (object) - Statistics summary for failed rescale operations.
- **ignoredRescalesDurationStatsInMillis** (object) - Statistics summary for ignored rescale operations.
- **rescalesCounts** (object) - Counts of different rescaling statuses (completed, failed, ignored, inProgress).
- **rescalesDurationStatsInMillis** (object) - Overall statistics summary for rescale operations.

#### Response Example
```json
{
  "completedRescalesDurationStatsInMillis": { ... },
  "failedRescalesDurationStatsInMillis": { ... },
  "ignoredRescalesDurationStatsInMillis": { ... },
  "rescalesCounts": {
    "completed": 0,
    "failed": 0,
    "ignored": 0,
    "inProgress": 0
  },
  "rescalesDurationStatsInMillis": { ... }
}
```
```

--------------------------------

### Build Docs Skipping Connector Integration

Source: https://github.com/apache/flink/blob/master/docs/README.md

Builds the Flink documentation locally, skipping the integration of external connector documentation. This is useful if connector docs have already been synced. Requires Hugo to be installed and setup scripts to be run.

```shell
#!/bin/bash
./build_docs.sh --skip-integrate-connector-docs
```

--------------------------------

### GET /overview

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves a comprehensive overview of the Flink cluster's current state, including version information, job statistics, and resource utilization.

```APIDOC
## GET /overview

### Description
Returns an overview over the Flink cluster.

### Method
GET

### Endpoint
/overview

### Response
#### Success Response (200 OK)
- **flink-commit** (string)
- **flink-version** (string)
- **jobs-cancelled** (integer)
- **jobs-failed** (integer)
- **jobs-finished** (integer)
- **jobs-running** (integer)
- **slots-available** (integer)
- **slots-free-and-blocked** (integer)
- **slots-total** (integer)
- **taskmanagers** (integer)
- **taskmanagers-blocked** (integer)

### Response Example
{
  "flink-commit": "<commit-sha>",
  "flink-version": "1.17.0",
  "jobs-cancelled": 0,
  "jobs-failed": 0,
  "jobs-finished": 0,
  "jobs-running": 1,
  "slots-available": 5,
  "slots-free-and-blocked": 2,
  "slots-total": 6,
  "taskmanagers": 2,
  "taskmanagers-blocked": 0
}
```

--------------------------------

### Get Job Manager Environment

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the environment information of the Job Manager for a specific job. This includes details about the JVM and classpath.

```APIDOC
## GET /jobs/:jobid/jobmanager/environment

### Description
Returns the jobmanager's environment of a specific job.

### Method
GET

### Endpoint
/jobs/:jobid/jobmanager/environment

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.

### Request Example
```json
{}
```

### Response
#### Success Response (200)
- **classpath** (array of strings)
- **jvm** (object)
  - **arch** (string)
  - **options** (array of strings)
  - **version** (string)

#### Response Example
```json
{
  "classpath": [
    "/path/to/jar1.jar",
    "/path/to/jar2.jar"
  ],
  "jvm": {
    "arch": "amd64",
    "options": [
      "-Xmx1024m",
      "-Djava.net.preferIPv4Stack=true"
    ],
    "version": "1.8.0_291"
  }
}
```
```

--------------------------------

### GET /jobmanager/thread-dump

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves a thread dump of the JobManager, which can be useful for diagnosing performance issues or deadlocks.

```APIDOC
## GET /jobmanager/thread-dump

### Description
Returns the thread dump of the JobManager.

### Method
GET

### Endpoint
/jobmanager/thread-dump

### Parameters

### Request Body

### Request Example
```json
{}
```

### Response
#### Success Response (200 OK)
- **threadInfos**: array of objects
  - **stringifiedThreadInfo** (string): The full string representation of the thread information.
  - **threadName** (string): The name of the thread.

#### Response Example
```json
{
  "threadInfos": [
    {
      "stringifiedThreadInfo": "\"main\" prio=5 tid=0x1 NPTL running\n  java.lang.Thread.State: RUNNABLE\n\t at java.lang.Object.wait(Native Method)\n\t at java.lang.Object.wait(Object.java:175)\n\t at org.apache.flink.runtime.jobmaster.JobMaster.run(JobMaster.java:178)\n\t at java.lang.Thread.run(Thread.java:748)",
      "threadName": "main"
    }
  ]
}
```
```

--------------------------------

### Start MinIO Server

Source: https://github.com/apache/flink/blob/master/flink-filesystems/flink-s3-fs-native/README.md

Start a MinIO server instance using Docker for testing S3 compatibility.

```bash
# Start MinIO
docker run -d -p 9000:9000 -p 9001:9001 \
  -e "MINIO_ROOT_USER=minioadmin" \
  -e "MINIO_ROOT_PASSWORD=minioadmin" \
  minio/minio server /data --console-address ":9001"
```

--------------------------------

### GET /jobmanager/logs

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves a list of available log files on the JobManager. Each log entry includes its name, modification time, and size.

```APIDOC
## GET /jobmanager/logs

### Description
Returns the list of log files on the JobManager.

### Method
GET

### Endpoint
/jobmanager/logs

### Parameters

### Request Body

### Request Example
```json
{}
```

### Response
#### Success Response (200 OK)
- **logs**: array of objects
  - **mtime** (integer): Modification time of the log file.
  - **name** (string): Name of the log file.
  - **size** (integer): Size of the log file in bytes.

#### Response Example
```json
{
  "logs": [
    {
      "mtime": 1678886400,
      "name": "jobmanager.log",
      "size": 10240
    },
    {
      "mtime": 1678886300,
      "name": "jobmanager.log.1",
      "size": 5120
    }
  ]
}
```
```

--------------------------------

### GET /jobs/metrics

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Provides access to aggregated metrics for all jobs. You can filter by specific jobs, select metrics, and specify aggregation modes.

```APIDOC
## GET /jobs/metrics

### Description
Provides access to aggregated job metrics.

### Method
GET

### Endpoint
/jobs/metrics

### Parameters
#### Query Parameters
- **get** (string) - Optional - Comma-separated list of string values to select specific metrics.
- **agg** (string) - Optional - Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew".
- **jobs** (string) - Optional - Comma-separated list of 32-character hexadecimal strings to select specific jobs.

### Request Body

### Request Example
```json
{}
```

### Response
#### Success Response (200 OK)
- **type**: any

#### Response Example
```json
{
  "job_id_1": {
    "metric.name.1": {
      "min": 10,
      "max": 100,
      "avg": 55
    }
  }
}
```
```

--------------------------------

### Get Job Manager Log URL

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the URL for the Job Manager's log file for a specific job.

```APIDOC
## GET /jobs/:jobid/jobmanager/log-url

### Description
Returns the log url of jobmanager of a specific job.

### Method
GET

### Endpoint
/jobs/:jobid/jobmanager/log-url

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.

### Request Example
```json
{}
```

### Response
#### Success Response (200)
- **url** (string)

#### Response Example
```json
{
  "url": "http://localhost:8081/jobs/some_job_id/jobmanager/log"
}
```
```

--------------------------------

### Get Job Resource Requirements

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the resource requirements for a specific Flink job.

```APIDOC
## GET /jobs/:jobid/resource-requirements

### Description
Request details on the job's resource requirements.

### Method
GET

### Endpoint
/jobs/:jobid/resource-requirements

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.

### Response
#### Success Response (200 OK)
- **jobVertexResourceRequirements** (object) - Contains resource requirements for each job vertex.
  - **parallelism** (object) - Resource requirements for parallelism.
    - **lowerBound** (integer) - The lower bound for parallelism.
    - **upperBound** (integer) - The upper bound for parallelism.
```

--------------------------------

### Display Execution Plan with Advice

Source: https://github.com/apache/flink/blob/master/flink-python/docs/examples/table/basic_operations.md

Use the explain method with ExplainDetail.PLAN_ADVICE to output the query plan along with optimization suggestions.

```python
print(table.join_lateral(split.alias('a')).explain(ExplainDetail.PLAN_ADVICE))
```

--------------------------------

### Get TaskManager Logs

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves a list of log files available on a specific TaskManager. This is useful for debugging and monitoring TaskManager operations.

```APIDOC
## GET /taskmanagers/:taskmanagerid/logs

### Description
Returns the list of log files on a TaskManager.

### Method
GET

### Endpoint
/taskmanagers/:taskmanagerid/logs

### Parameters
#### Path Parameters
- **taskmanagerid** (string) - Required - 32-character hexadecimal string that identifies a task manager.

### Response
#### Success Response (200 OK)
- **logs** (array) - An array of log file information, where each item contains 'mtime' (integer), 'name' (string), and 'size' (integer).

#### Response Example
{
  "logs": [
    {
      "mtime": 1678886400,
      "name": "flink-taskmanager-0-log.out",
      "size": 10240
    }
  ]
}
```

--------------------------------

### Get Job Vertex Accumulators

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves user-defined accumulators for a specific job vertex, aggregated across all its subtasks.

```APIDOC
## GET /jobs/:jobid/vertices/:vertexid/accumulators

### Description
Returns user-defined accumulators of a task, aggregated across all subtasks.

### Method
GET

### Endpoint
/jobs/:jobid/vertices/:vertexid/accumulators

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.
- **vertexid** (string) - Required - 32-character hexadecimal string value that identifies a job vertex.

### Response
#### Success Response (200 OK)
- **id** (string) - The ID of the job vertex.
- **user-accumulators** (array) - An array of user-defined accumulators.
  - **name** (string) - The name of the accumulator.
  - **type** (string) - The type of the accumulator.
  - **value** (string) - The value of the accumulator.
```

--------------------------------

### GET /taskmanagers/metrics

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves aggregated metrics for task managers. This endpoint allows for filtering metrics and applying aggregation functions.

```APIDOC
## GET /taskmanagers/metrics

### Description
Provides access to aggregated task manager metrics.

### Method
GET

### Endpoint
/taskmanagers/metrics

### Parameters
#### Query Parameters
- **get** (string) - Optional - Comma-separated list of string values to select specific metrics.
- **agg** (string) - Optional - Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew".
- **taskmanagers** (string) - Optional - Comma-separated list of 32-character hexadecimal strings to select specific task managers.

#### Request Body
None

### Request Example
```json
{}
```

### Response
#### Success Response (200)
- **any** (any) - The response contains aggregated metrics data.

### Response Example
```json
{
  "metric_name_1": {
    "min": 10,
    "max": 100,
    "avg": 55
  }
}
```
```

--------------------------------

### GET /jobs/:jobid/vertices/:vertexid

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Fetches details for a specific job vertex and a summary for each of its subtasks.

```APIDOC
## GET /jobs/:jobid/vertices/:vertexid

### Description
Returns details for a task, with a summary for each of its subtasks.

### Method
GET

### Endpoint
/jobs/:jobid/vertices/:vertexid

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.
- **vertexid** (string) - Required - 32-character hexadecimal string value that identifies a job vertex.

### Response
#### Success Response (200 OK)
- **aggregated** (object) - Aggregated information for the vertex.
- **id** (any) - The ID of the vertex.
- **maxParallelism** (integer) - The maximum parallelism of the vertex.
- **name** (string) - The name of the vertex.
- **now** (integer) - The current timestamp.
- **parallelism** (integer) - The parallelism of the vertex.
- **subtasks** (array) - An array of subtask details.
  - **attempt** (integer) - The attempt number of the subtask.
  - **duration** (integer) - The duration of the subtask execution.
  - **end-time** (integer) - The end time of the subtask execution.
  - **endpoint** (string) - The endpoint of the subtask.
  - **metrics** (object) - Metrics for the subtask.
    - **accumulated-backpressured-time** (integer) - Accumulated backpressured time.
    - **accumulated-busy-time** (number) - Accumulated busy time.
    - **accumulated-idle-time** (integer) - Accumulated idle time.
    - **read-bytes** (integer) - Number of bytes read.
    - **read-bytes-complete** (boolean) - Indicates if read bytes are complete.
    - **read-records** (integer) - Number of records read.
    - **read-records-complete** (boolean) - Indicates if read records are complete.
    - **write-bytes** (integer) - Number of bytes written.
    - **write-bytes-complete** (boolean) - Indicates if write bytes are complete.
    - **write-records** (integer) - Number of records written.
    - **write-records-complete** (boolean) - Indicates if write records are complete.
  - **other-concurrent-attempts** (array) - Other concurrent attempts for this subtask.
  - **start-time** (integer) - The start time of the subtask execution.
  - **status** (string) - The status of the subtask (e.g., CREATED, RUNNING, FINISHED).
  - **status-duration** (object) - Duration of each status.
  - **subtask** (integer) - The subtask index.
  - **taskmanager-id** (string) - The ID of the TaskManager running the subtask.
```

--------------------------------

### Start Local Flink Cluster

Source: https://github.com/apache/flink/blob/master/docs/content/docs/dev/table/olap_quickstart.md

This script starts a local Flink cluster. After execution, you can access the Flink dashboard at http://localhost:8081.

```bash
./bin/start-cluster.sh
```

--------------------------------

### Example Part File Structure

Source: https://github.com/apache/flink/blob/master/docs/content.zh/docs/connectors/datastream/filesystem.md

Illustrates the directory structure and naming of In-progress Part files within a bucket.

```text
└── 2019-08-25--12
    ├── part-4005733d-a830-4323-8291-8866de98b582-0.inprogress.bd053eb0-5ecf-4c85-8433-9eff486ac334
    └── part-81fc4980-a6af-41c8-9937-9939408a734b-0.inprogress.ea65a428-a1d0-4a0b-bbc5-7a436a75e575
```

--------------------------------

### Get Job Checkpoint Statistics

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves statistics for all checkpoints of a specific job. This includes details about completed, in-progress, and failed checkpoints.

```APIDOC
## GET /jobs/:jobid/checkpoints

### Description
Retrieves statistics for all checkpoints of a specific job.

### Method
GET

### Endpoint
/jobs/:jobid/checkpoints

### Parameters
#### Path Parameters
- **jobid** (string) - Required - The ID of the job for which to retrieve checkpoint statistics.

### Response
#### Success Response (200)
- **properties** (object) - Contains statistics for completed and in-progress checkpoints.
  - **alignment_buffered** (integer)
  - **checkpoint_type** (string) - Enum: CHECKPOINT, UNALIGNED_CHECKPOINT, SAVEPOINT, SYNC_SAVEPOINT
  - **checkpointed_size** (integer)
  - **discarded** (boolean)
  - **end_to_end_duration** (integer)
  - **external_path** (string)
  - **id** (integer)
  - **is_savepoint** (boolean)
  - **latest_ack_timestamp** (integer)
  - **num_acknowledged_subtasks** (integer)
  - **num_subtasks** (integer)
  - **persisted_data** (integer)
  - **processed_data** (integer)
  - **savepointFormat** (string)
  - **state_size** (integer)
  - **status** (string) - Enum: IN_PROGRESS, COMPLETED, FAILED
  - **tasks** (object) - Statistics for individual tasks within a checkpoint.
  - **trigger_timestamp** (integer)
- **failed** (object) - Contains statistics for failed checkpoints.
  - **alignment_buffered** (integer)
  - **checkpoint_type** (string) - Enum: CHECKPOINT, UNALIGNED_CHECKPOINT, SAVEPOINT, SYNC_SAVEPOINT
  - **checkpointed_size** (integer)
  - **end_to_end_duration** (integer)
  - **failure_message** (string)
  - **failure_timestamp** (integer)
  - **id** (integer)
  - **is_savepoint** (boolean)
  - **latest_ack_timestamp** (integer)
  - **num_acknowledged_subtasks** (integer)
  - **num_subtasks** (integer)
  - **persisted_data** (integer)

#### Response Example
{
  "properties": {
    "alignment_buffered": 1024,
    "checkpoint_type": "CHECKPOINT",
    "checkpointed_size": 2048,
    "discarded": false,
    "end_to_end_duration": 500,
    "external_path": "hdfs:///flink/checkpoints/123",
    "id": 1,
    "is_savepoint": false,
    "latest_ack_timestamp": 1678886400000,
    "num_acknowledged_subtasks": 4,
    "num_subtasks": 4,
    "persisted_data": 1024,
    "processed_data": 2048,
    "savepointFormat": "FSC",
    "state_size": 1024,
    "status": "COMPLETED",
    "tasks": {
      "task_1": {
        "alignment_buffered": 256,
        "checkpointed_size": 512,
        "end_to_end_duration": 100,
        "id": 0,
        "latest_ack_timestamp": 1678886400000,
        "num_acknowledged_subtasks": 1,
        "num_subtasks": 1,
        "persisted_data": 256,
        "processed_data": 512,
        "state_size": 256,
        "status": "COMPLETED"
      }
    },
    "trigger_timestamp": 1678886300000
  },
  "failed": {
    "alignment_buffered": 0,
    "checkpoint_type": "CHECKPOINT",
    "checkpointed_size": 0,
    "end_to_end_duration": 0,
    "failure_message": "Task failed to complete",
    "failure_timestamp": 1678886500000,
    "id": 2,
    "is_savepoint": false,
    "latest_ack_timestamp": 0,
    "num_acknowledged_subtasks": 0,
    "num_subtasks": 4,
    "persisted_data": 0
  }
}
```

--------------------------------

### Installing GCS File System Plugin

Source: https://github.com/apache/flink/blob/master/docs/content/docs/deployment/filesystems/gcs.md

Instructions for installing the `flink-gs-fs-hadoop` plugin. Copy the JAR file from the `opt` directory to the `plugins` directory of your Flink distribution before starting Flink.

```bash
mkdir ./plugins/gs-fs-hadoop
cp ./opt/flink-gs-fs-hadoop-{{< version >}}.jar ./plugins/gs-fs-hadoop/
```

--------------------------------

### GET /jobmanager/config

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the current configuration of the Flink cluster. This endpoint returns an array of key-value pairs representing the configuration settings.

```APIDOC
## GET /jobmanager/config

### Description
Returns the cluster configuration.

### Method
GET

### Endpoint
/jobmanager/config

### Parameters

### Request Body

### Request Example
```json
{}
```

### Response
#### Success Response (200 OK)
- **type**: array
- **items**: object
  - **key** (string)
  - **value** (string)

#### Response Example
```json
[
  {
    "key": "some.config.key",
    "value": "some.config.value"
  }
]
```
```

--------------------------------

### Get Job Metrics

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Provides access to the metrics of a specific Flink job. You can optionally filter the metrics returned.

```APIDOC
## GET /jobs/:jobid/metrics

### Description
Provides access to job metrics.

### Method
GET

### Endpoint
/jobs/:jobid/metrics

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.

#### Query Parameters
- **get** (string) - Optional - Comma-separated list of string values to select specific metrics.

### Request Example
```json
{}
```

### Response
#### Success Response (200)
- **any** (any)

#### Response Example
```json
{
  "metric.name.1": 123,
  "metric.name.2": 456
}
```
```

--------------------------------

### Get Subtask Execution Attempt Details

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves details of the current or latest execution attempt of a subtask for a given job and vertex.

```APIDOC
## GET /jobs/:jobid/vertices/:vertexid/subtasks/:subtaskindex

### Description
Returns details of the current or latest execution attempt of a subtask.

### Method
GET

### Endpoint
/jobs/:jobid/vertices/:vertexid/subtasks/:subtaskindex

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.
- **vertexid** (string) - Required - 32-character hexadecimal string value that identifies a job vertex.
- **subtaskindex** (integer) - Required - Positive integer value that identifies a subtask.

### Response
#### Success Response (200 OK)
- **attempt** (integer)
- **duration** (integer)
- **end-time** (integer)
- **endpoint** (string)
- **metrics** (object)
  - **accumulated-backpressured-time** (integer)
  - **accumulated-busy-time** (number)
  - **accumulated-idle-time** (integer)
  - **read-bytes** (integer)
  - **read-bytes-complete** (boolean)
  - **read-records** (integer)
  - **read-records-complete** (boolean)
  - **write-bytes** (integer)
  - **write-bytes-complete** (boolean)
  - **write-records** (integer)
  - **write-records-complete** (boolean)
- **other-concurrent-attempts** (array)
- **start-time** (integer)
- **start_time** (integer)
- **status** (string) - Enum: CREATED, SCHEDULED, DEPLOYING, RUNNING, FINISHED, CANCELING, CANCELED, FAILED, RECONCILING, INITIALIZING
- **status-duration** (object)
- **subtask** (integer)
- **taskmanager-id** (string)

#### Response Example
{
  "attempt": 0,
  "duration": 12345,
  "end-time": 1678886400000,
  "endpoint": "taskmanager_1:12345",
  "metrics": {
    "accumulated-backpressured-time": 100,
    "accumulated-busy-time": 10000.5,
    "accumulated-idle-time": 200,
    "read-bytes": 102400,
    "read-bytes-complete": true,
    "read-records": 5000,
    "read-records-complete": true,
    "write-bytes": 204800,
    "write-bytes-complete": true,
    "write-records": 10000,
    "write-records-complete": true
  },
  "other-concurrent-attempts": [],
  "start-time": 1678874055000,
  "start_time": 1678874055000,
  "status": "RUNNING",
  "status-duration": {
    "RUNNING": 12345
  },
  "subtask": 0,
  "taskmanager-id": "taskmanager_1"
}
```

--------------------------------

### Configure Checkpoint Storage (Java Example)

Source: https://github.com/apache/flink/blob/master/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md

Demonstrates how to configure checkpoint storage in Java by setting the checkpoint storage type to 'filesystem' and specifying a directory for storing snapshots.

```java
Configuration config = new Configuration();
config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "...");
env.configure(config);
```

--------------------------------

### Start Flink Cluster Locally

Source: https://github.com/apache/flink/blob/master/docs/content/docs/getting-started/local_installation.md

Shell script to start a local Apache Flink cluster. Requires Flink to be downloaded and extracted, and Java to be installed.

```bash
$ ./bin/start-cluster.sh

```

--------------------------------

### Get TaskManager Metrics

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Provides access to the metrics exposed by a specific TaskManager. You can filter metrics by providing a comma-separated list of metric names.

```APIDOC
## GET /taskmanagers/:taskmanagerid/metrics

### Description
Provides access to task manager metrics.

### Method
GET

### Endpoint
/taskmanagers/:taskmanagerid/metrics

### Parameters
#### Path Parameters
- **taskmanagerid** (string) - Required - 32-character hexadecimal string that identifies a task manager.

#### Query Parameters
- **get** (string) - Optional - Comma-separated list of string values to select specific metrics.

### Response
#### Success Response (200 OK)
- The response type is 'any', indicating a flexible metric structure.

#### Response Example
{
  "metric.name.1": 123.45,
  "metric.name.2": 678
}
```

--------------------------------

### Get Job Manager Configuration

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the configuration details of the Job Manager for a specific job. This endpoint returns a list of key-value pairs representing the configuration.

```APIDOC
## GET /jobs/:jobid/jobmanager/config

### Description
Returns the jobmanager's configuration of a specific job.

### Method
GET

### Endpoint
/jobs/:jobid/jobmanager/config

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.

### Request Example
```json
{}
```

### Response
#### Success Response (200)
- **key** (string)
- **value** (string)

#### Response Example
```json
[
  {
    "key": "example.key",
    "value": "example.value"
  }
]
```
```

--------------------------------

### GET /jobs/:jobid/rescales/config

Source: https://github.com/apache/flink/blob/master/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

Retrieves the rescale configuration for a specific Flink job. This endpoint provides details on how the job is configured for rescaling.

```APIDOC
## GET /jobs/:jobid/rescales/config

### Description
Returns the job rescale configuration. This endpoint provides details on how the job is configured for rescaling.

### Method
GET

### Endpoint
/jobs/:jobid/rescales/config

### Parameters
#### Path Parameters
- **jobid** (string) - Required - 32-character hexadecimal string value that identifies a job.

### Request Example
```json
{}
```

### Response
#### Success Response (200 OK)
- **executingCooldownTimeoutInMillis** (integer)
- **executingResourceStabilizationTimeoutInMillis** (integer)
- **maximumDelayForTriggeringRescaleInMillis** (integer)
- **rescaleHistoryMax** (integer)
- **rescaleOnFailedCheckpointCount** (integer)
- **schedulerExecutionMode** (string) - Must be 'REACTIVE'.
- **slotIdleTimeoutInMillis** (integer)
- **submissionResourceStabilizationTimeoutInMillis** (integer)
- **submissionResourceWaitTimeoutInMillis** (integer)

#### Response Example
```json
{
  "executingCooldownTimeoutInMillis": 0,
  "executingResourceStabilizationTimeoutInMillis": 0,
  "maximumDelayForTriggeringRescaleInMillis": 0,
  "rescaleHistoryMax": 0,
  "rescaleOnFailedCheckpointCount": 0,
  "schedulerExecutionMode": "REACTIVE",
  "slotIdleTimeoutInMillis": 0,
  "submissionResourceStabilizationTimeoutInMillis": 0,
  "submissionResourceWaitTimeoutInMillis": 0
}
```
```

--------------------------------

### Configure Main Execution Entry Point

Source: https://github.com/apache/flink/blob/master/flink-python/docs/examples/table/basic_operations.md

Sets up logging and invokes various operation functions.

```python
if __name__ == '__main__':
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")

    basic_operations()
    sql_operations()
    column_operations()
    row_operations()
```