Apache Flink

https://nightlies.apache.org/flink/flink-docs-master/
Admin
Apache Flink is a framework and distributed processing engine designed for stateful computations...
Tokens:11,495,712
Snippets:76,199
Trust Score:10
Update:1 month ago
Show doc for...
Tokens:
Raw
### Example: window start and end

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/_modules/pyflink/table/expression.html

Shows using start and end properties on a window reference. From the start property docstring.

```python
>>> tab.window(Tumble
>>>         .over(row_interval(2))
>>>         .on(col("a"))
>>>         .alias("w")) \
>>>     .group_by(col("c"), col("w")) \
>>>     .select(col("c"), col("w").start, col("w").end, col("w").proctime)
```

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

### Get window start time with Expression.start

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/reference/pyflink.table/api/pyflink.table.expression.Expression.start.html

Shows how to use the start property on a window reference to get the start time of a tumbling window. The example groups by the window and selects the start, end, and proctime columns.

```python
>>> tab.window(Tumble
>>>         .over(row_interval(2))
>>>         .on(col("a"))
>>>         .alias("w")) \
>>>     .group_by(col("c"), col("w")) \
>>>     .select(col("c"), col("w").start, col("w").end, col("w").proctime)

```

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

### ADD JAR Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/add

Examples showing how to add local and remote JAR files.

```sql
-- add a local jar
ADD JAR t.jar;

-- add a remote jar
ADD JAR hdfs://namenode-host:port/path/t.jar
```

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

### setup()

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.html

Performs setup operations for the gate.

```APIDOC
### void setup()

Initializes the gate. This may be a heavy-weight or blocking operation.
```

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

### LOAD DATA Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/load-data

Examples demonstrating how to load data into a table and into a specific partition.

```sql
-- load data into table
LOAD DATA LOCAL INPATH '/user/warehouse/hive/t1' OVERWRITE INTO TABLE t1;

-- load data into partition
LOAD DATA LOCAL INPATH '/user/warehouse/hive/t1/p1=1' INTO TABLE t1 PARTITION (p1=1);
```

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

### Add Partition Example

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/alter

Example of adding multiple partitions to a table.

```sql
ALTER TABLE t1 ADD PARTITION (dt='2022-08-08', country='china') location '/path/to/us/part080808'
                   PARTITION (dt='2022-08-09', country='china') location '/path/to/us/part080809';
```

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

### TableConfig usage example with add_configuration

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/_modules/pyflink/table/table_config.html

Example from the class docstring showing how to get the table config from an environment and add a Configuration object with parallelism.default, pipeline.auto-watermark-interval, and execution.checkpointing.interval settings.

```python
>>> table_config = t_env.get_config()
>>> config = Configuration()
>>> config.set_string("parallelism.default", "128") \
...       .set_string("pipeline.auto-watermark-interval", "800ms") \
...       .set_string("execution.checkpointing.interval", "30s")
>>> table_config.add_configuration(config)
```

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

### Create Flink project with quickstart

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/formats/azure_table_storage

Sets up a new Flink project using the official quickstart script.

```bash
curl https://flink.apache.org/q/quickstart.sh | bash

```

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

### Start the Flink SQL Client

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/getting-started/quickstart-sql

Commands to launch the interactive SQL client environment based on the installation method.

```bash
$ docker compose run sql-client
```

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

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

### SORT BY Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/queries/sort-cluster-distribute-by

Examples demonstrating how to use SORT BY for partial ordering within partitions.

```sql
SELECT x, y FROM t SORT BY x;
SELECT x, y FROM t SORT BY abs(y) DESC;
```

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

### Define Source and Target Tables

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/materialized-table/statements

Example setup of a source table and an existing regular table intended for conversion.

```sql
-- Source table the query reads from
CREATE TABLE orders (
    user_id BIGINT NOT NULL,
    amount BIGINT,
    order_time TIMESTAMP(3)
) WITH (
    'connector' = '...'
);

-- The existing regular table to convert. It has a primary key and a watermark,
-- which the conversion does not carry over.
CREATE TABLE user_spending (
    user_id BIGINT NOT NULL,
    total_amount BIGINT,
    last_order TIMESTAMP(3),
    PRIMARY KEY (user_id) NOT ENFORCED,
    WATERMARK FOR last_order AS last_order - INTERVAL '5' SECOND
) WITH (
    'connector' = '...'
);
```

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

### Alter Database Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/alter

Examples for setting database properties and location.

```sql
-- alter database's properties
ALTER DATABASE d1 SET DBPROPERTIES ('p1' = 'v1', 'p2' = 'v2');

-- alter database's location
ALTER DATABASE d1 SET LOCATION '/new/path';
```

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

### SQL Gateway Start Command Help

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/interfaces/sql-gateway/overview

Shows help for the 'start' command, detailing options for running the SQL Gateway as a daemon.

```bash
$ ./bin/sql-gateway.sh start --help

Start the Flink SQL Gateway as a daemon to submit Flink SQL.

  Syntax: start [OPTIONS]
     -D <property=value>   Use value for given property
     -h,--help             Show the help message with descriptions of all
                           options.

```

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

### Install PyFlink and start the Python shell

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/repls/python_shell

Install PyFlink from PyPI and start the Python shell with an integrated Flink cluster. The shell runs the command "python" and requires PyFlink to be installed.

```bash
# install PyFlink
$ python -m pip install apache-flink
# execute the shell
$ pyflink-shell.sh local
```

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

### DISTRIBUTE BY Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/queries/sort-cluster-distribute-by

Examples demonstrating data repartitioning using DISTRIBUTE BY, optionally combined with SORT BY.

```sql
-- only use DISTRIBUTE BY clause
SELECT x, y FROM t DISTRIBUTE BY x;
SELECT x, y FROM t DISTRIBUTE BY abs(y);

-- use both DISTRIBUTE BY and SORT BY clause
SELECT x, y FROM t DISTRIBUTE BY x SORT BY y DESC;
```

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

### Method: setup

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/state/v2/adaptor/AsyncKeyedStateBackendAdaptor.html

Initializes the adaptor with the provided `StateRequestHandler`.

```APIDOC
## Method: setup

### Description
Initializes with some contexts.

### Method
`void setup(StateRequestHandler stateRequestHandler)`

### Parameters
* **stateRequestHandler** (`StateRequestHandler`) - The state request handler to set up with.
```

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

### Get Initialization Start Timestamp

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/checkpoint/SubTaskInitializationMetricsBuilder.html

Retrieves the start timestamp used during builder initialization.

```java
public long getInitializationStartTs()
```

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

### Create Database Syntax and Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/create

Syntax for defining a database and examples showing basic creation and creation with properties.

```sql
CREATE (DATABASE|SCHEMA) [IF NOT EXISTS] database_name
  [COMMENT database_comment]
  [LOCATION hdfs_path]
  [WITH DBPROPERTIES (property_name=property_value, ...)];
```

```sql
CREATE DATABASE db1;
CREATE DATABASE IF NOT EXISTS db1 COMMENT 'db1' LOCATION '/user/hive/warehouse/db1'
    WITH DBPROPERTIES ('name'='example-db');
```

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

### Example ZooKeeper HA Configuration

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/ha/zookeeper_ha

A complete example of the required configuration keys for a Flink HA setup.

```yaml
high-availability.type: zookeeper
high-availability.zookeeper.quorum: localhost:2181
high-availability.zookeeper.path.root: /flink
high-availability.cluster-id: /cluster_one # important: customize per cluster
high-availability.storageDir: hdfs:///flink/recovery

```

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

### Method: setup

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/operators/shipping/OutputEmitter.html

Initializes the channel selector with the total number of available output channels.

```APIDOC
## setup

### Description
Initializes the channel selector with the number of output channels.

### Parameters
#### Query Parameters
- **numberOfChannels** (int) - Required - The total number of output channels attached to the respective output gate.
```

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

### Word count Python Table API program

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/user_guide/table_api_tutorial.html

Defines a word count program using the Flink Python Table API. It parses command-line arguments for input and output files, then calls the word_count function. This is the main example to get started with writing Flink Python Table API programs.

```python
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--input',
        dest='input',
        required=False,
        help='Input file to process.')
    parser.add_argument(
        '--output',
        dest='output',
        required=False,
        help='Output file to write results to.')

    argv = sys.argv[1:]
    known_args, _ = parser.parse_known_args(argv)

    word_count(known_args.input, known_args.output)
```

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

### CLUSTER BY Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/queries/sort-cluster-distribute-by

Examples demonstrating the use of CLUSTER BY as a shortcut for both DISTRIBUTE BY and SORT BY.

```sql
SELECT x, y FROM t CLUSTER BY x;
SELECT x, y FROM t CLUSTER BY abs(y);
```

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

### get with setup

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/tests/util/flink/FlinkResource.html

Returns the configured FlinkResource implementation, allowing for specific setup instructions. Defaults to LocalStandaloneFlinkResource if no configuration is found.

```APIDOC
## get

### Description
Returns the configured FlinkResource implementation, or a `LocalStandaloneFlinkResource` if none is configured.

### Method
static FlinkResource get(FlinkResourceSetup setup)

### Parameters
#### Path Parameters
- **setup** (FlinkResourceSetup) - Required - setup instructions for the FlinkResource

### Returns
- **FlinkResource** - configured FlinkResource, or `LocalStandaloneFlinkResource` is none is configured
```

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

### Gradle: settings.gradle for Flink quickstart project

Source: https://nightlies.apache.org/flink/flink-docs-master//docs/dev/configuration/overview

Defines the root project name for the Gradle build, necessary for the quickstart setup.

```groovy
rootProject.name = 'quickstart'
```

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

### KafkaSinkBuilder minimum setup example

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/_modules/pyflink/datastream/connectors/kafka.html

Shows the minimum setup to create a KafkaSink that writes String values to a Kafka topic. Uses KafkaRecordSerializationSchema.builder() and KafkaSink.builder().

```python
>>> record_serializer = KafkaRecordSerializationSchema.builder() \
...     .set_topic(MY_SINK_TOPIC) \
...     .set_value_serialization_schema(SimpleStringSchema()) \
...     .build()
>>> sink = KafkaSink.builder() \
...     .set_bootstrap_servers(MY_BOOTSTRAP_SERVERS) \
...     .set_record_serializer(record_serializer) \
...     .build()
```

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

### start

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironment.html

Starts the internal services required for the shuffle environment.

```APIDOC
## POST /start

### Description
Start the internal related services before using the shuffle service environment.

### Method
POST

### Response
#### Success Response (200)
- **port** (int) - A port to connect for the shuffle data exchange, -1 if only local connection is possible.
```

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

### Start Netcat Server for Input

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/_sources/cookbook/word_count.rst.txt

This bash command starts a netcat listener on port 9999, which will be used as the input source for the PyFlink word count examples.

```bash
nc -lk 9999
```

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

### setupPartitionsAndGates(ResultPartitionWriter[], InputGate[])

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/taskmanager/Task.html

Sets up the produced partitions and input gates for the task.

```APIDOC
## setupPartitionsAndGates(ResultPartitionWriter[], InputGate[])

### Description
Sets up the produced partitions and input gates for the task.

### Method
`static void setupPartitionsAndGates(ResultPartitionWriter[] producedPartitions, InputGate[] inputGates)`

### Parameters
- `producedPartitions` (ResultPartitionWriter[]) - Required - The array of produced partitions.
- `inputGates` (InputGate[]) - Required - The array of input gates.

### Return Type
`void`
```

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

### Maven: Quickstart script

Source: https://nightlies.apache.org/flink/flink-docs-master//docs/dev/configuration/overview

Downloads and runs the Flink quickstart script for Maven to create a project. Pass the version as an argument.

```bash
$ curl https://flink.apache.org/q/quickstart.sh | bash -s 2.4-SNAPSHOT
```

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

### KafkaSourceBuilder minimum setup example

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/_modules/pyflink/datastream/connectors/kafka.html

Minimum setup to create a KafkaSource that reads String values from a Kafka topic. Requires bootstrap servers, topics/partitions, and record deserializer.

```python
>>> source = KafkaSource.builder() \
...     .set_bootstrap_servers('MY_BOOTSTRAP_SERVERS') \
...     .set_topics('TOPIC1', 'TOPIC2') \
...     .set_value_only_deserializer(SimpleStringSchema()) \
...     .build()
```

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

### setup Method

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/HashBufferAccumulator.html

Sets up the accumulator with a flusher for accumulated buffers.

```APIDOC
## setup Method

### Description
Setup the accumulator. This method is specified by the `setup` method in the `BufferAccumulator` interface.

### Method
`void setup(TriConsumer<TieredStorageSubpartitionId, Buffer, Integer> accumulatedBufferFlusher)`

### Parameters
- **accumulatedBufferFlusher** (TriConsumer<TieredStorageSubpartitionId, Buffer, Integer>) - Accepts the accumulated buffers. The first field is the subpartition id, the second is the accumulated buffer to flush, and the third is the number of remaining buffers to be written consecutively to the same segment.
```

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

### Example configuration

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/ha/kubernetes_ha

Full example of the required configuration keys in the Flink configuration file.

```yaml
kubernetes.cluster-id: <cluster-id>
high-availability.type: kubernetes
high-availability.storageDir: hdfs:///flink/recovery
```

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

### POST /state/setup

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/state/forst/ForStKeyedStateBackend.html

Initializes the state backend with a request handler.

```APIDOC
## POST /state/setup

### Description
Initializes the ForStKeyedStateBackend with a provided StateRequestHandler.

### Method
POST

### Parameters
#### Request Body
- **stateRequestHandler** (StateRequestHandler) - Required - The handler responsible for processing state requests.
```

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

### SessionWindowing Constructor

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.html

Initializes the SessionWindowing example class. No specific setup is required for this constructor.

```java
public SessionWindowing()
```

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

### RESET Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/materialized-table/statements

Examples demonstrating the removal of table options, handling of non-existent keys, and de-duplication of keys.

```sql
-- Remove the 'format' and 'sink.parallelism' options
ALTER MATERIALIZED TABLE my_materialized_table RESET ('format', 'sink.parallelism');

-- 'sink.parallelism' is not currently set on the table: this is a no-op for that key,
-- 'format' is still removed and the statement succeeds.
ALTER MATERIALIZED TABLE my_materialized_table RESET ('format', 'sink.parallelism');

-- Duplicates collapse to a single reset for 'format'.
ALTER MATERIALIZED TABLE my_materialized_table RESET ('format', 'format');
```

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

### ZooKeeperSharedValue Methods

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/zookeeper/ZooKeeperSharedValue.html

Provides methods for starting, closing, getting, and setting shared values.

```APIDOC
## ZooKeeperSharedValue Methods

### Method: start

### Description
Starts the ZooKeeperSharedValue.

### Method
`void start()`

### Throws
`Exception`

---

### Method: close

### Description
Closes the ZooKeeperSharedValue.

### Method
`void close()`

### Throws
`IOException`

---

### Method: setValue

### Description
Sets a new value for the shared value.

### Method
`void setValue(byte[] newValue)`

### Parameters
* **newValue** (byte[]) - The new byte array value to set.

### Throws
`Exception`

---

### Method: getValue

### Description
Gets the current value of the shared value.

### Method
`byte[] getValue()`

### Returns
* **byte[]** - The current shared value as a byte array.
```

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

### INSERT TABLE Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/insert

Examples demonstrating inserting values, overwriting tables, using select statements, and handling static or dynamic partitions.

```sql
-- insert into table using values
INSERT INTO t1 VALUES ('k1', 'v1'), ('k2', 'v2');

-- insert overwrite
INSERT OVERWRITE TABLE t1 VALUES ('k1', 'v1'), ('k2', 'v2');;

-- insert into table using select statement
INSERT INTO TABLE t1 SELECT * FROM t2;

-- insert into  partition
--- static partition
INSERT INTO t1 PARTITION (year = 2022, month = 12) SELECT value FROM t2;

--- dynamic partition 
INSERT INTO t1 PARTITION (year = 2022, month) SELECT value, month FROM t2;
INSERT INTO t1 PARTITION (year, month) SELECT value, 2022, month FROM t2;
```

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

### Start and Stop Flink Cluster (Session Mode)

Source: https://nightlies.apache.org/flink/flink-docs-master/deployment/resource-providers/standalone

Use these commands to start a Flink cluster, submit an example job, and then stop the cluster. The web interface is accessible at http://localhost:8081.

```bash
# we assume to be in the root directory of the unzipped Flink distribution

# (1) Start Cluster
$ ./bin/start-cluster.sh

# (2) You can now access the Flink Web Interface on http://localhost:8081

# (3) Submit example job
$ ./bin/flink run ./examples/streaming/TopSpeedWindowing.jar

# (4) Stop the cluster again
$ ./bin/stop-cluster.sh

```

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

### setup Method

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.html

Sets up the operator with the containing task, configuration, and output.

```APIDOC
## setup Method

### Description
Sets up the operator with the containing task, configuration, and output.

### Method
setup

### Signature
`public void setup(StreamTask<?,?> containingTask, StreamConfig config, Output<StreamRecord<OUT>> output)`

### Parameters
- **containingTask** (StreamTask<?,?>) - Required - The containing stream task.
- **config** (StreamConfig) - Required - The stream configuration.
- **output** (Output<StreamRecord<OUT>>) - Required - The output collector for stream records.

### Usage
This method is called during operator initialization to provide the operator with its runtime context.
```

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

### PyFlink Sliding Window Example

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/_sources/examples/table/window.rst.txt

Illustrates the use of a sliding window in PyFlink. Requires PyFlink installation.

```python
from pyflink.common import WatermarkStrategy, Time
from pyflink.table import EnvironmentSettings, TableEnvironment


def sliding_window_example():
    settings = EnvironmentSettings.new_instance().in_batch_mode().build()
    table_env = TableEnvironment.create(settings)

    # Sample data
    data = [
        ("user1", "event1", 1678886400000),
        ("user2", "event2", 1678886460000),
        ("user1", "event3", 1678886520000),
        ("user3", "event4", 1678886580000),
        ("user2", "event5", 1678886640000),
    ]
    table = table_env.from_elements(data, ["user", "event", "timestamp"])

    # Define watermark strategy
    watermark_strategy = WatermarkStrategy.for_bounded_out_of_orders_ness(
        max_out_of_orderness=10000
    ).with_timestamp_assigner(lambda event: event[2])

    table = table.assign_timestamps_and_watermarks(watermark_strategy)

    # Apply sliding window
    windowed_table = table.window(
        table.rowtime.slide.window(Time.minutes(10), Time.minutes(5))
    ).group_by("user") \
    .select("user, count(event) as event_count, window_end() as window_end")

    table_env.execute_sql("CREATE TABLE output (
        user VARCHAR,
        event_count BIGINT,
        window_end TIMESTAMP(3)
    ) WITH (
        'connector' = 'print'
    )")
    windowed_table.execute_insert("output")

if __name__ == '__main__':
    sliding_window_example()

```

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

### Manage SerDe Properties Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/alter

Examples for adding and removing SerDe properties.

```sql
-- add serde properties
ALTER TABLE t1 SET SERDEPROPERTIES ('field.delim' = ',');

-- remove serde properties
ALTER TABLE t1 UNSET SERDEPROPERTIES ('field.delim');
```

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

### StateReaderOperator Setup

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/state/class-use/KeyedStateBackend.html

Demonstrates the setup method of StateReaderOperator which accepts a KeyedStateBackend.

```APIDOC
## StateReaderOperator.setup

### Description
Sets up the StateReaderOperator with necessary components including a KeyedStateBackend.

### Method
`void`

### Parameters
- **serializerFactory** (SerializerFactory) - Description not provided
- **keyKeyedStateBackend** (KeyedStateBackend<KEY>) - The keyed state backend to be used.
- **timerServiceManager** (InternalTimeServiceManager<KEY>) - Description not provided
- **ctx** (SavepointRuntimeContext) - Description not provided
```

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

### Get Default TableConfig

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/table/api/TableConfig.html

Returns the default TableConfig instance. This provides a starting point for custom configurations.

```java
static TableConfig getDefault();
```

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

### Show All Catalogs Example

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/reference/utility/show

An example output of the 'SHOW CATALOGS' command, listing available catalogs including the default catalog.

```sql
show catalogs;
+-----------------+
|    catalog name |
+-----------------+
|        catalog1 |
|        catalog2 |
| default_catalog |
+-----------------+
3 rows in set
```

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

### PyFlink Tumble Window Example

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/_sources/examples/table/window.rst.txt

Demonstrates the usage of a tumble window in PyFlink. Ensure PyFlink is installed and configured.

```python
from pyflink.common import WatermarkStrategy, Time
from pyflink.table import EnvironmentSettings, TableEnvironment


def tumble_window_example():
    settings = EnvironmentSettings.new_instance().in_batch_mode().build()
    table_env = TableEnvironment.create(settings)

    # Sample data
    data = [
        ("user1", "event1", 1678886400000),
        ("user2", "event2", 1678886460000),
        ("user1", "event3", 1678886520000),
        ("user3", "event4", 1678886580000),
        ("user2", "event5", 1678886640000),
    ]
    table = table_env.from_elements(data, ["user", "event", "timestamp"])

    # Define watermark strategy
    watermark_strategy = WatermarkStrategy.for_bounded_out_of_orders_ness(
        max_out_of_orderness=10000
    ).with_timestamp_assigner(lambda event: event[2])

    table = table.assign_timestamps_and_watermarks(watermark_strategy)

    # Apply tumble window
    windowed_table = table.window(
        table.rowtime.tumble.window(Time.minutes(5))
    ).group_by("user") \
    .select("user, count(event) as event_count, window_end() as window_end")

    table_env.execute_sql("CREATE TABLE output (
        user VARCHAR,
        event_count BIGINT,
        window_end TIMESTAMP(3)
    ) WITH (
        'connector' = 'print'
    )")
    windowed_table.execute_insert("output")

if __name__ == '__main__':
    tumble_window_example()

```

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

### QuickstartExample Class

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/quickstarts/test/QuickstartExample.html

Overview of the QuickstartExample class and its main execution method.

```APIDOC
## Class: org.apache.flink.quickstarts.test.QuickstartExample

### Description
End to end test for quickstarts.

### Methods
#### main
`public static void main(String[] args) throws Exception`

- **args** (String[]) - Command line arguments.
- **Throws** (Exception) - Throws an exception if the execution fails.
```

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

### Create Macro Syntax and Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/create

Syntax for creating temporary macros and examples for various expression types.

```sql
CREATE TEMPORARY MACRO macro_name([col_name col_type, ...]) expression;
```

```sql
CREATE TEMPORARY MACRO fixed_number() 42;
CREATE TEMPORARY MACRO string_len_plus_two(x string) length(x) + 2;
CREATE TEMPORARY MACRO simple_add (x int, y int) x + y;
```

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

### POST /execution-plans/start

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/runtime/jobmanager/ExecutionPlanStore.html

Starts the ExecutionPlanStore service.

```APIDOC
## POST /execution-plans/start

### Description
Starts the ExecutionPlanStore service.

### Method
POST

### Endpoint
/execution-plans/start

### Request Body
- **executionPlanListener** (ExecutionPlanStore.ExecutionPlanListener) - Required - A listener for ExecutionPlan instances.
```

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

### Implementing RichFilterFunction with open()

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/api/common/functions/RichFunction.html

Example of extending RichFilterFunction to perform one-time setup work within the open() method.

```java
 public class MyFilter extends RichFilterFunction<String> {

     private String searchString;

     public void open(OpenContext openContext) {
         // initialize the value of searchString
     }

     public boolean filter(String value) {
         return value.equals(searchString);
     }
 }
```

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

### Full File System Connector SQL Example

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/filesystem

Demonstrates creating tables and executing streaming inserts and batch queries using the filesystem connector.

```sql
CREATE TABLE kafka_table (
  user_id STRING,
  order_amount DOUBLE,
  log_ts TIMESTAMP(3),
  WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND
) WITH (...);

CREATE TABLE fs_table (
  user_id STRING,
  order_amount DOUBLE,
  dt STRING,
  `hour` STRING
) PARTITIONED BY (dt, `hour`) WITH (
  'connector'='filesystem',
  'path'='...',
  'format'='parquet',
  'sink.partition-commit.delay'='1 h',
  'sink.partition-commit.policy.kind'='success-file'
);

-- streaming sql, insert into file system table
INSERT INTO fs_table 
SELECT 
    user_id, 
    order_amount, 
    DATE_FORMAT(log_ts, 'yyyy-MM-dd'),
    DATE_FORMAT(log_ts, 'HH') 
FROM kafka_table;

-- batch sql, select with partition pruning
SELECT * FROM fs_table WHERE dt='2020-05-20' and `hour`='12';
```

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

### RuntimeContext.get_index_of_this_subtask()

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/reference/pyflink.datastream/api/pyflink.datastream.functions.RuntimeContext.get_index_of_this_subtask.html

Gets the number of this parallel subtask. The numbering starts from 0 and goes up to parallelism-1 (parallelism as returned by `get_number_of_parallel_subtasks()`).

```APIDOC
## RuntimeContext.get_index_of_this_subtask()

### Description
Gets the number of this parallel subtask. The numbering starts from 0 and goes up to parallelism-1 (parallelism as returned by `get_number_of_parallel_subtasks()`).

### Method
`get_index_of_this_subtask()`

### Return Type
`int`

### Usage
This method is part of the `RuntimeContext` class in the PyFlink DataStream API. It is used to retrieve the index of the current parallel subtask within the overall parallelism.

### Example
```python
index = runtime_context.get_index_of_this_subtask()
```
```

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

### TaskManagerLocation resourceID field

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/serialized-form.html

The ID of the resource in which the TaskManager is started. This can be for example the YARN container ID or any other unique identifier.

```java
ResourceID resourceID
```

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

### SQL Client startup options usage

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/interfaces/sql-client

Shows the general syntax for starting the SQL Client with optional modes and options, followed by the details of the embedded mode options. Use this to see all available flags and their descriptions.

```plaintext
./sql-client [MODE] [OPTIONS]

The following options are available:

Mode "embedded" (default) submits Flink jobs from the local machine.

  Syntax: [embedded] [OPTIONS]
  "embedded" mode options:
     -D <session dynamic config key=val>        The dynamic config key=val for a
                                                session.
     -f,--file <script file>                    Script file that should be
                                                executed. In this mode, the
                                                client will not open an
                                                interactive terminal.
     -h,--help                                  Show the help message with
                                                descriptions of all options.
     -hist,--history <History file path>        The file which you want to save
                                                the command history into. If not
                                                specified, we will auto-generate
                                                one under your user's home
                                                directory.
     -i,--init <initialization file>            Script file that used to init
                                                the session context. If get
                                                error in execution, the sql
                                                client will exit. Notice it's
                                                not allowed to add query or
                                                insert into the init file.
     -j,--jar <JAR file>                        A JAR file to be imported into
                                                the session. The file might
                                                contain user-defined classes
                                                needed for the execution of
                                                statements such as functions,
                                                table sources, or sinks. Can be
                                                used multiple times.
     -l,--library <lib directory>              A JAR file directory with which
                                                every new session is
                                                initialized. The files might
                                                contain user-defined classes
                                                needed for the execution of
                                                statements such as functions,
                                                table sources, or sinks. Can be
                                                used multiple times.
     -pyarch,--pyArchives <arg>                 Add python archive files for
                                                job. The archive files will be
                                                extracted to the working
                                                directory of python UDF worker.
                                                For each archive file, a target
                                                directory be specified. If the
                                                target directory name is
                                                specified, the archive file will
                                                be extracted to a directory with
                                                the specified name. Otherwise,
                                                the archive file will be
                                                extracted to a directory with
                                                the same name of the archive
                                                file. The files uploaded via
                                                this option are accessible via
                                                the relative path. '#' could be
                                                used as the separator of the
                                                file path and the target
                                                directory name. Comma (',')
                                                could be used as the separator
                                                to specify multiple archive
                                                files. This option can be used
                                                to upload the virtual
                                                environment, the data files used
                                                in Python UDF (e.g.,
                                                --pyArchives
```

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

### CREATE TABLE AS Syntax Example

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/sql/parser/ddl/table/SqlCreateTableAs.html

Demonstrates the SQL syntax for defining a base table and a derived table using the CREATE TABLE AS statement.

```sql
 CREATE TABLE base_table (
     id BIGINT,
     name STRING,
     tstmp TIMESTAMP,
     PRIMARY KEY(id)
 ) WITH (
     'connector' = 'kafka',
     'connector.starting-offset': '12345',
     'format' =  'json'
 )

 CREATE TABLE derived_table
 WITH (
   'connector' = 'jdbc',
   'url' = 'http://localhost:10000',
   'table-name' = 'syncedTable'
 )
 AS SELECT * FROM base_table;
```

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

### Initialize WindowOperatorBuilder

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/table/runtime/operators/window/groupwindow/operator/WindowOperatorBuilder.html

A static method to start building a WindowOperator. No specific setup is required beyond calling this method.

```java
WindowOperatorBuilder.builder()
```

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

### Create View Syntax and Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/create

Syntax for creating a logical view and an example demonstrating column comments and filtering.

```sql
CREATE VIEW [IF NOT EXISTS] [db_name.]view_name [(column_name, ...) ]
  [COMMENT view_comment]
  [TBLPROPERTIES (property_name = property_value, ...)]
  AS SELECT ...;
```

```sql
CREATE VIEW IF NOT EXISTS v1
    (key COMMENT 'key') 
    COMMENT 'View for key=1'
    AS SELECT key FROM src
        WHERE key = '1';
```

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

### Gradle: Quickstart script

Source: https://nightlies.apache.org/flink/flink-docs-master//docs/dev/configuration/overview

Download and execute the Gradle quickstart script to create a fully functional startup Flink project.

```bash
bash -c "$(curl https://flink.apache.org/q/gradle-quickstart.sh)" -- 2.4-SNAPSHOT _2.12
```

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

### Simple Condition Example

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/libs/cep

Filters events where the name starts with 'foo'. This condition only considers properties of the current event.

```java
start.where(SimpleCondition.of(value -> value.getName().startsWith("foo")));

```

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

### CREATE MODEL Examples

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create

Examples demonstrating the registration of sentiment analysis and Triton-based classification models.

```sql
CREATE MODEL sentiment_analysis_model 
INPUT (text STRING COMMENT 'Input text for sentiment analysis') 
OUTPUT (sentiment STRING COMMENT 'Predicted sentiment (positive/negative/neutral/mixed)')
COMMENT 'A model for sentiment analysis of text'
WITH (
    'provider' = 'openai',
    'endpoint' = 'https://api.openai.com/v1/chat/completions',
    'api-key' = '<YOUR KEY>',
    'model'='gpt-3.5-turbo',
    'system-prompt' = 'Classify the text below into one of the following labels: [positive, negative, neutral, mixed]. Output only the label.'
);
```

```sql
CREATE MODEL triton_text_classifier
INPUT (input STRING COMMENT 'Input text for classification')
OUTPUT (output STRING COMMENT 'Classification result')
COMMENT 'A Triton-based text classification model'
WITH (
    'provider' = 'triton',
    'endpoint' = 'http://localhost:8000/v2/models',
    'model-name' = 'text-classification',
    'model-version' = '1',
    'timeout' = '10000',
    'max-retries' = '3'
);
```

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

### Configure Flink with Dynamic Properties

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/resource-providers/standalone/overview

Example of setting configuration parameters via dynamic properties when starting a script.

```bash
$ ./bin/jobmanager.sh start -D jobmanager.rpc.address=localhost -D rest.port=8081
```

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

### Setup Input and Output Tables

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/models/openai

Initializes the source view for movie comments and the print sink table for storing prediction results.

```sql
CREATE TEMPORARY VIEW movie_comment(id, movie_name,  user_comment, actual_label)
AS VALUES
  (1, 'Good Stuff', 'The part where children guess the sounds is my favorite. It's a very romantic narrative compared to other movies I've seen. Very gentle and full of love.', 'positive');

CREATE TEMPORARY TABLE print_sink(
  id BIGINT,
  movie_name VARCHAR,
  predicit_label VARCHAR,
  actual_label VARCHAR
) WITH (
  'connector' = 'print'
);
```

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

### Example requirements.txt content

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/user_guide/dependency_management.html

Example content of a requirements.txt file with pinned package versions.

```text
numpy==1.16.5
pandas==1.0.0
```

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

### Prepare Secure Test Environment

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/test/util/SecureTestEnvironment.html

Use the prepare method to initialize the MiniKDC and set up the secure environment. It accepts a temporary folder and an optional list of additional principals.

```java
public static void prepare(File tempFolder, String... additionalPrincipals)
```

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

### Create a KafkaSource

Source: https://nightlies.apache.org/flink/flink-docs-master/api/python/reference/pyflink.datastream/api/pyflink.datastream.connectors.kafka.KafkaSourceBuilder.html

The following example shows the minimum setup to create a KafkaSource that reads the String values from a Kafka topic.

```python
>>> source = KafkaSource.builder() \
...     .set_bootstrap_servers('MY_BOOTSTRAP_SERVERS') \
...     .set_topics('MY_TOPIC')
```

```python
>>> source = KafkaSource.builder() \
...     .set_bootstrap_servers('MY_BOOTSTRAP_SERVERS') \
...     .set_topics('MY_TOPIC') \
...     .set_group_id('MY_GROUP_ID') \
...     .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
...     .set_value_only_deserializer(SimpleStringSchema())
```

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

### Method: setup

Source: https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceWithAsyncState.html

Configures the async execution controller for the timer service.

```APIDOC
## setup

### Description
Set up the async execution controller.

### Method
void

### Parameters
- **asyncExecutionController** (AsyncExecutionController<K,?>) - Required - The async execution controller to be used.
```

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

### Example: Partition by two columns with ORDER BY

Source: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/hive-compatibility/hive-dialect/queries/window-functions

Shows using SUM with PARTITION BY on two columns and ORDER BY on two columns, without a specific window frame.

```sql
SELECT a, SUM(b) OVER (PARTITION BY c, d ORDER BY e, f) FROM t;
```
Tokens:
Raw