### Run Gelly Examples

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/libs/gelly/overview

Start the Flink cluster and run the Gelly examples jar.

```bash
./bin/start-cluster.sh
./bin/flink run examples/gelly/flink-gelly-examples_*.jar
```

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

### Instantiate Table Environment with Configuration (Scala)

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/config

Shows how to set up a TableEnvironment in Scala using a Configuration object within EnvironmentSettings. Includes examples of configuring options both during initial setup and via the TableConfig object.

```scala
val configuration = new Configuration;
configuration.setString("table.exec.mini-batch.enabled", "true")
configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
configuration.setString("table.exec.mini-batch.size", "5000")
val settings = EnvironmentSettings.newInstance
  .inStreamingMode.withConfiguration(configuration).build
val tEnv: TableEnvironment = TableEnvironment.create(settings)

val tableConfig = tEnv.getConfig()
tableConfig.set("table.exec.mini-batch.enabled", "true")
tableConfig.set("table.exec.mini-batch.allow-latency", "5 s")
tableConfig.set("table.exec.mini-batch.size", "5000")
```

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

### Complete Kafka Source/Sink Example with JSON Format

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/python_table_api_connectors

A full example demonstrating the setup and usage of Kafka source and sink with JSON format in PyFlink.

```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()
```

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

### Kafka Console Consumer Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/hive/hive_catalog

Command to start a Kafka console consumer to read messages from a topic. This example retrieves all messages from the 'test' topic.

```bash
localhost$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

tom,15
john,21
```

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

### Create New Flink Project using Quickstart

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/dataset/formats/azure_table_storage

Set up a new Apache Flink project using the provided quickstart script.

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

```

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

### Defining a Starting Pattern in Java

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

Shows how to define a starting point for a CEP pattern using Pattern.begin().

```Java
Pattern<Event, ?> start = Pattern.<Event>begin("start");
```

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

### Start Local Flink Cluster

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/gettingstarted

Use this command to start a local Flink cluster from the installation folder. This allows for local experimentation and monitoring via the Flink WebUI.

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

```

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

### Kafka Console Producer Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/hive/hive_catalog

Command to start a Kafka console producer to send messages to a topic. This example sends comma-separated 'name,age' pairs to the 'test' topic.

```bash
localhost$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>tom,15
>john,21
```

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

### Defining a Starting Pattern in Scala

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

Shows how to define a starting point for a CEP pattern using Pattern.begin().

```Scala
val start = Pattern.begin[Event]("start")
```

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

### Run WordCount Example (Default)

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/dataset/examples

Command to run the default WordCount example JAR. Assumes a running Flink instance.

```bash
./bin/flink run ./examples/batch/WordCount.jar

```

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

### Run WordCount Example with Input/Output Paths

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/dataset/examples

Command to run the WordCount example with specified input and output paths. Requires a running Flink instance.

```bash
./bin/flink run ./examples/batch/WordCount.jar --input /path/to/some/text/data --output /path/to/result

```

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

### Installing the GCS File System Plugin

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/filesystems/gcs

Instructions for installing the flink-gs-fs-hadoop JAR file into the Flink plugins directory.

```bash
mkdir ./plugins/gs-fs-hadoop
cp ./opt/flink-gs-fs-hadoop-1.15.4.jar ./plugins/gs-fs-hadoop/
```

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

### Java: Create, Show, and Use Catalog and Database

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/use

Demonstrates creating a catalog and database, then setting them as the default using USE CATALOG and USE statements in Java. Shows how to verify changes with SHOW CATALOGS and SHOW DATABASES.

```Java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

// create a catalog
tEnv.executeSql("CREATE CATALOG cat1 WITH (...)");
tEnv.executeSql("SHOW CATALOGS").print();
// +-----------------+
// |    catalog name |
// +-----------------+
// | default_catalog |
// | cat1            |
// +-----------------+

// change default catalog
tEnv.executeSql("USE CATALOG cat1");

tEnv.executeSql("SHOW DATABASES").print();
// databases are empty
// +---------------+ 
// | database name |
// +---------------+ 
// +---------------+ 

// create a database
tEnv.executeSql("CREATE DATABASE db1 WITH (...)");
tEnv.executeSql("SHOW DATABASES").print();
// +---------------+ 
// | database name |
// +---------------+ 
// |        db1    |
// +---------------+ 

// change default database
tEnv.executeSql("USE db1");

// change module resolution order and enabled status
tEnv.executeSql("USE MODULES hive");
tEnv.executeSql("SHOW FULL MODULES").print();
// +-------------+-------+
// | module name |  used |
// +-------------+-------+
// |        hive |  true |
// |        core | false |
// +-------------+-------+

```

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

### Create Cassandra Keyspace and Table

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/cassandra

This CQL script creates the 'example' keyspace and the 'wordcount' table, which are prerequisites for the Flink Cassandra sink examples.

```cql
CREATE KEYSPACE IF NOT EXISTS example
    WITH replication = {"class": "SimpleStrategy", "replication_factor": "1"};

CREATE TABLE IF NOT EXISTS example.wordcount (
    word text,
    count bigint,
    PRIMARY KEY(word)
);

```

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

### Scala: Create, Show, and Use Catalog and Database

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/use

Illustrates creating a catalog and database, then setting them as the default using USE CATALOG and USE statements in Scala. Includes verification steps using SHOW CATALOGS and SHOW DATABASES.

```Scala
val env = StreamExecutionEnvironment.getExecutionEnvironment()
val tEnv = StreamTableEnvironment.create(env)

// create a catalog
tEnv.executeSql("CREATE CATALOG cat1 WITH (...)")
tEnv.executeSql("SHOW CATALOGS").print()
// +-----------------+
// |    catalog name |
// +-----------------+
// | default_catalog |
// | cat1            |
// +-----------------+

// change default catalog
tEnv.executeSql("USE CATALOG cat1")

tEnv.executeSql("SHOW DATABASES").print()
// databases are empty
// +---------------+ 
// | database name |
// +---------------+ 
// +---------------+ 

// create a database
tEnv.executeSql("CREATE DATABASE db1 WITH (...)")
tEnv.executeSql("SHOW DATABASES").print()
// +---------------+ 
// | database name |
// +---------------+ 
// |        db1    |
// +---------------+ 

// change default database
tEnv.executeSql("USE db1")

// change module resolution order and enabled status
tEnv.executeSql("USE MODULES hive")
tEnv.executeSql("SHOW FULL MODULES").print()
// +-------------+-------+
// | module name |  used |
// +-------------+-------+
// |        hive |  true |
// |        core | false |
// +-------------+-------+

```

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

### Start Input Stream with Netcat

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/overview

This command starts a netcat listener to provide input to the Flink word count application. Ensure netcat is installed on your system.

```bash
nc -lk 9999

```

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

### Download and Execute Flink Quickstart Script (Maven)

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

This command downloads and executes a bash script to set up a Flink project with Maven. Specify the Flink version.

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

```

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

### Python: Create, Show, and Use Catalog and Database

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/use

Shows how to create a catalog and database, then set them as the default using USE CATALOG and USE statements in Python. Verification is done using SHOW CATALOGS and SHOW DATABASES.

```Python
table_env = StreamTableEnvironment.create(...)

# create a catalog
table_env.execute_sql("CREATE CATALOG cat1 WITH (...)")
table_env.execute_sql("SHOW CATALOGS").print()
# +-----------------+
# |    catalog name |
# +-----------------+
# | default_catalog |
# | cat1            |
# +-----------------+

# change default catalog
table_env.execute_sql("USE CATALOG cat1")

table_env.execute_sql("SHOW DATABASES").print()
# databases are empty
# +---------------+ 
# | database name |
# +---------------+ 
# +---------------+ 

# create a database
table_env.execute_sql("CREATE DATABASE db1 WITH (...)")
table_env.execute_sql("SHOW DATABASES").print()
# +---------------+ 
# | database name |
# +---------------+ 
# |           db1 |
# +---------------+ 

# change default database
table_env.execute_sql("USE db1")

# change module resolution order and enabled status
table_env.execute_sql("USE MODULES hive")
table_env.execute_sql("SHOW FULL MODULES").print()
# +-------------+-------+
# | module name |  used |
# +-------------+-------+
# |        hive |  true |
# |        core | false |
# +-------------+-------+

```

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

### DataStream API Job Structure Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/execution_mode

Illustrates a typical DataStream API job setup with various transformations like map, rebalance, keyBy, and sinkTo. This example is used to explain task scheduling and network shuffle behavior.

```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStreamSource<String> source = env.fromElements(...);

source.name("source")
	.map(...).name("map1")
	.map(...).name("map2")
	.rebalance()
	.map(...).name("map3")
	.map(...).name("map4")
	.keyBy((value) -> value)
	.map(...).name("map5")
	.map(...).name("map6")
	.sinkTo(...).name("sink");


```

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

### EXPLAIN Syntax with ExplainDetails

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/explain

Demonstrates the basic syntax for using the EXPLAIN command with specific ExplainDetails to retrieve detailed execution plan information.

```sql
EXPLAIN ([ExplainDetail[, ExplainDetail]*]) <query_statement_or_insert_statement_or_statement_set>
```

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

### Filtering Data with Lambda Functions

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/user_defined_functions

Use lambda functions for concise data filtering operations. This example filters strings starting with 'http://'.

```scala
val data: DataSet[String] = // [...]
data.filter { _.startsWith("http://") }
```

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

### TUMBLE Function Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/queries/window-tvf

Applies the TUMBLE function to the Bid table with a 10-minute window size, demonstrating the output with window start, end, and time columns.

```sql
-- tables must have time attribute, e.g. `bidtime` in this table
Flink SQL> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

Flink SQL> SELECT * FROM Bid;
+------------------+-------+------+
|          bidtime | price | item |
+------------------+-------+------+
| 2020-04-15 08:05 |  4.00 | C    |
| 2020-04-15 08:07 |  2.00 | A    |
| 2020-04-15 08:09 |  5.00 | D    |
| 2020-04-15 08:11 |  3.00 | B    |
| 2020-04-15 08:13 |  1.00 | E    |
| 2020-04-15 08:17 |  6.00 | F    |
+------------------+-------+------+

Flink SQL> SELECT * FROM TABLE(
   TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES));
-- or with the named params
-- note: the DATA param must be the first
Flink SQL> SELECT * FROM TABLE(
   TUMBLE(
     DATA => TABLE Bid,
     TIMECOL => DESCRIPTOR(bidtime),
     SIZE => INTERVAL '10' MINUTES));
+------------------+-------+------+------------------+------------------+-------------------------+
|          bidtime | price | item |     window_start |       window_end |            window_time  |
+------------------+-------+------+------------------+------------------+-------------------------+
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:07 |  2.00 | A    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:09 |  5.00 | D    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:17 |  6.00 | F    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
+------------------+-------+------+------------------+------------------+-------------------------+

```

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

### Download and Execute Flink Gradle Quickstart Script

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

This command downloads and executes a bash script to set up a Flink project with Gradle. Specify the Flink version and Scala binary version.

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

```

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

### Start Flink SQL Client

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/gettingstarted

Launch the interactive SQL client to submit queries to Flink and view results. This script is run from the Flink installation directory.

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

```

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

### Create a Kafka Table

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/kafka

Example of creating a Kafka table with basic configurations for reading user behavior data.

```sql
CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

```

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

### Example Output of Max Ride Duration

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/learn-flink/etl

Illustrates the output format for a keyed stream after applying a maxBy aggregation, showing the start cell and the maximum duration achieved.

```text
...
4> (64549,5M)
4> (46298,18M)
1> (51549,14M)
1> (53043,13M)
1> (56031,22M)
1> (50797,6M)
...
1> (50797,8M)
...
1> (50797,11M)
...
1> (50797,12M)
```

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

### SQL Client Help Command

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sqlclient

Displays the help message with descriptions of all available options for the SQL Client.

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

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

### Explain Table Plan using Table.explain()

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/intro_to_table_api

Demonstrates how to use the `Table.explain()` method to retrieve the query plan for a single table. This includes setting up a streaming TableEnvironment, creating tables from elements, applying a filter and union, and then printing the explained plan.

```python
# using a stream TableEnvironment
from pyflink.table import EnvironmentSettings, TableEnvironment
from pyflink.table.expressions import col

env_settings = EnvironmentSettings.in_streaming_mode()
table_env = TableEnvironment.create(env_settings)

table1 = table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['id', 'data'])
table2 = table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['id', 'data'])
table = table1 \
    .where(col("data").like('H%')) \
    .union_all(table2)
print(table.explain())

```

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

### TYPEOF Function Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/system_functions

Shows how to get the string representation of an expression's data type. Setting force_serializable to TRUE provides a full data type string.

```sql
TYPEOF(input)
```

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

### Start a distributed cluster with multiple JobManagers

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

Define 'conf/masters' and 'conf/workers' for a distributed setup across multiple machines. Ensure network reachability and correct 'jobmanager.rpc.address' configuration.

```text
master1


```

```text
worker1
worker2
worker3


```

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

### Create Tables for EXPLAIN Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/explain

Sets up two datagen tables, MyTable1 and MyTable2, which are used in subsequent EXPLAIN statements.

```sql
Flink SQL> CREATE TABLE MyTable1 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen');
[INFO] Table has been created.

Flink SQL> CREATE TABLE MyTable2 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen');
[INFO] Table has been created.
```

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

### Python Table API Batch Example

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

Demonstrates creating a batch table sink and inserting data using Flink's Python Table API. Ensure PyFlink is installed and the shell is running.

```python
>>> import tempfile
>>> import os
>>> import shutil
>>> sink_path = tempfile.gettempdir() + '/batch.csv'
>>> if os.path.exists(sink_path):
...     if os.path.isfile(sink_path):
...         os.remove(sink_path)
...     else:
...         shutil.rmtree(sink_path)
>>> b_env.set_parallelism(1)
>>> t = bt_env.from_elements([(1, 'hi', 'hello'), (2, 'hi', 'hello')], ['a', 'b', 'c'])
>>> st_env.create_temporary_table("batch_sink", TableDescriptor.for_connector("filesystem")
...     .schema(Schema.new_builder()
...         .column("a", DataTypes.BIGINT())
...         .column("b", DataTypes.STRING())
...         .column("c", DataTypes.STRING())
...         .build())
...     .option("path", path)
...     .format(FormatDescriptor.for_format("csv")
...         .option("field-delimiter", ",")
...         .build())
...     .build())
>>> t.select("a + 1, b, c")
...     .execute_insert("batch_sink").wait()
>>> # If the job runs in local mode, you can exec following code in Python shell to see the result:
>>> with open(os.path.join(sink_path, os.listdir(sink_path)[0]), 'r') as f:
...     print(f.read())
```

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

### Create Table with Parquet Format

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/parquet

Example of creating a partitioned table using the Filesystem connector and Parquet format. Ensure the 'path' and 'format' options are correctly set.

```sql
CREATE TABLE user_behavior (
  user_id BIGINT,
  item_id BIGINT,
  category_id BIGINT,
  behavior STRING,
  ts TIMESTAMP(3),
  dt STRING
) PARTITIONED BY (dt) WITH (
 'connector' = 'filesystem',
 'path' = '/tmp/user_behavior',
 'format' = 'parquet'
)

```

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

### Instantiate Table Environment with Configuration (Python)

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/config

Provides a Python example for creating a TableEnvironment with custom configurations. It illustrates setting parameters via a Configuration object and accessing/modifying them through the TableConfig interface.

```python
# instantiate table environment
configuration = Configuration()
configuration.set("table.exec.mini-batch.enabled", "true")
configuration.set("table.exec.mini-batch.allow-latency", "5 s")
configuration.set("table.exec.mini-batch.size", "5000")
settings = EnvironmentSettings.new_instance() \
    .in_streaming_mode() \
    .with_configuration(configuration) \
    .build()

t_env = TableEnvironment.create(settings)

# access flink configuration after table environment instantiation
table_config = t_env.get_config()
# set low-level key-value options
table_config.set("table.exec.mini-batch.enabled", "true")
table_config.set("table.exec.mini-batch.allow-latency", "5 s")
table_config.set("table.exec.mini-batch.size", "5000")
```

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

### Python Table API Streaming Example

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

Demonstrates creating a streaming table sink and inserting data using Flink's Python Table API. Ensure PyFlink is installed and the shell is running.

```python
>>> import tempfile
>>> import os
>>> import shutil
>>> sink_path = tempfile.gettempdir() + '/streaming.csv'
>>> if os.path.exists(sink_path):
...     if os.path.isfile(sink_path):
...         os.remove(sink_path)
...     else:
...         shutil.rmtree(sink_path)
>>> s_env.set_parallelism(1)
>>> t = st_env.from_elements([(1, 'hi', 'hello'), (2, 'hi', 'hello')], ['a', 'b', 'c'])
>>> st_env.create_temporary_table("stream_sink", TableDescriptor.for_connector("filesystem")
...     .schema(Schema.new_builder()
...         .column("a", DataTypes.BIGINT())
...         .column("b", DataTypes.STRING())
...         .column("c", DataTypes.STRING())
...         .build())
...     .option("path", path)
...     .format(FormatDescriptor.for_format("csv")
...         .option("field-delimiter", ",")
...         .build())
...     .build())
>>> t.select("a + 1, b, c")
...     .execute_insert("stream_sink").wait()
>>> # If the job runs in local mode, you can exec following code in Python shell to see the result:
>>> with open(os.path.join(sink_path, os.listdir(sink_path)[0]), 'r') as f:
...     print(f.read())
```

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

### Instantiate Table Environment with Configuration (Java)

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/config

Demonstrates how to instantiate a TableEnvironment in Java by providing a Configuration object to EnvironmentSettings. It shows setting low-level key-value options both before and after TableEnvironment instantiation.

```java
Configuration configuration = new Configuration();
configuration.setString("table.exec.mini-batch.enabled", "true");
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
configuration.setString("table.exec.mini-batch.size", "5000");
EnvironmentSettings settings = EnvironmentSettings.newInstance()
        .inStreamingMode().withConfiguration(configuration).build();
TableEnvironment tEnv = TableEnvironment.create(settings);

TableConfig tableConfig = tEnv.getConfig();
tableConfig.set("table.exec.mini-batch.enabled", "true");
tableConfig.set("table.exec.mini-batch.allow-latency", "5 s");
tableConfig.set("table.exec.mini-batch.size", "5000");
```

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

### Full Example: Creating and Using an Upsert Kafka Table

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/upsert-kafka

Demonstrates how to define an upsert-kafka sink table and an upstream Kafka source table, then populate the sink with aggregated data from the source.

```sql
CREATE TABLE pageviews_per_region (
  user_region STRING,
  pv BIGINT,
  uv BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'pageviews_per_region',
  'properties.bootstrap.servers' = '...',
  'key.format' = 'avro',
  'value.format' = 'avro'
);
```

```sql
CREATE TABLE pageviews (
  user_id BIGINT,
  page_id BIGINT,
  viewtime TIMESTAMP,
  user_region STRING,
  WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'pageviews',
  'properties.bootstrap.servers' = '...',
  'format' = 'json'
);
```

```sql
-- calculate the pv, uv and insert into the upsert-kafka sink
INSERT INTO pageviews_per_region
SELECT
  user_region,
  COUNT(*),
  COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_region;
```

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

### Create Table with Kafka Connector and JSON Format

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/json

Example of creating a Flink table using the Kafka connector and specifying the JSON format. This setup includes options for handling missing fields and ignoring parse errors.

```sql
CREATE TABLE user_behavior (
  user_id BIGINT,
  item_id BIGINT,
  category_id BIGINT,
  behavior STRING,
  ts TIMESTAMP(3)
) WITH (
 'connector' = 'kafka',
 'topic' = 'user_behavior',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'json',
 'json.fail-on-missing-field' = 'false',
 'json.ignore-parse-errors' = 'true'
)
```

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

### Pulsar Start from Specific Message ID

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/pulsar

Configure the Pulsar source to start consuming from a specific message ID. If the message ID does not exist, consumption starts from the latest available message. The start message is included.

```java
StartCursor.fromMessageId(MessageId);
```

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

### Create and Use JDBC Catalog (SQL)

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/jdbc

Demonstrates how to create a JDBC catalog and set it as the current catalog using SQL.

```sql
CREATE CATALOG my_catalog WITH(
    'type' = 'jdbc',
    'default-database' = '...',
    'username' = '...',
    'password' = '...',
    'base-url' = '...'
);

USE CATALOG my_catalog;


```

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

### Example PyFlink Program Output

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/datastream_tutorial

This is an example of the output you can expect after running a PyFlink DataStream API program, such as a word count example.

```text
(a,5)
(Be,1)
(Is,1)
(No,2)
...

```

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

### File System Connector Full Example

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

Demonstrates using the file system connector to write streaming data from Kafka to a file system table and then reading it back in batch mode. Includes table definitions and INSERT/SELECT statements.

```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';
```

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

### Pulsar Start from Specific Message ID with Inclusion Option

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/pulsar

Configure the Pulsar source to start consuming from a specific message ID, with an option to include or exclude the start message using a boolean parameter. If the message ID does not exist, consumption starts from the latest available message.

```java
StartCursor.fromMessageId(MessageId, boolean);
```

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

### Show All Columns Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/show

Demonstrates how to show all columns of a table named 'orders'. This can be done by specifying the table name directly or with its catalog and database.

```sql
show columns from orders;
-- show columns from database1.orders;
-- show columns from catalog1.database1.orders;
-- show columns in orders;
-- show columns in database1.orders;
-- show columns in catalog1.database1.orders;
```

```text
+---------+-----------------------------+-------+-----------+---------------+----------------------------+
|    name |                        type |  null |       key |        extras |                  watermark |
+---------+-----------------------------+-------+-----------+---------------+----------------------------+
|    user |                      BIGINT | false | PRI(user) |               |                            |
| product |                 VARCHAR(32) |  true |           |               |                            |
|  amount |                         INT |  true |           |               |                            |
|      ts |      TIMESTAMP(3) *ROWTIME* |  true |           |               | `ts` - INTERVAL '1' SECOND |
|   ptime | TIMESTAMP_LTZ(3) *PROCTIME* | false |           | AS PROCTIME() |                            |
+---------+-----------------------------+-------+-----------+---------------+----------------------------+
5 rows in set
```

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

### Create Table with File Metadata

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

Demonstrates how to create a table and expose file metadata like 'file.path'.

```sql
CREATE TABLE MyUserTableWithFilepath (
  column_name1 INT,
  column_name2 STRING,
  `file.path` STRING NOT NULL METADATA
) WITH (
  'connector' = 'filesystem',
  'path' = 'file:///path/to/whatever',
  'format' = 'json'
)

```

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

### Start a Pattern Sequence (Scala)

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

Initializes a new pattern sequence with a starting event.

```Scala
val start : Pattern[Event, _] = Pattern.begin("start")

```

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

### Start a Pattern Sequence (Java)

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

Initializes a new pattern sequence with a starting event.

```Java
Pattern<Event, ?> start = Pattern.<Event>begin("start");

```

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

### Create Kafka Table with Metadata

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/kafka

Demonstrates creating a Kafka table and exposing Kafka metadata fields like partition and offset.

```sql
CREATE TABLE KafkaTable (
  `event_time` TIMESTAMP(3) METADATA FROM 'timestamp',
  `partition` BIGINT METADATA VIRTUAL,
  `offset` BIGINT METADATA VIRTUAL,
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

```

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

### Check Java Version

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/try-flink/local_installation

Verify that Java 11 is installed on your system before proceeding with Flink installation.

```bash
$ java -version

```

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

### Reading, Writing, and Checkpointing with GCS

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/filesystems/gcs

Examples demonstrating how to use GCS paths for reading text files, writing text files, and configuring GCS as a checkpoint storage location.

```java
// Read from GCS bucket
env.readTextFile("gs://<bucket>/<endpoint>");

// Write to GCS bucket
stream.writeAsText("gs://<bucket>/<endpoint>");

// Use GCS as checkpoint storage
env.getCheckpointConfig().setCheckpointStorage("gs://<bucket>/<endpoint>");
```

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

### Pulsar Start from Publish Time

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/pulsar

Configure the Pulsar source to start consuming from a specified publish time.

```java
StartCursor.fromPublishTime(long);
```

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

### Start Additional TaskManager

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/elastic_scaling

Starts an additional TaskManager to scale up the Flink job running in Reactive Mode.

```bash
# Start additional TaskManager
./bin/taskmanager.sh start

```

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

### Explain StatementSet Plan using StatementSet.explain()

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/intro_to_table_api

Shows how to use `StatementSet.explain()` to get the query plan for a job with multiple sinks. This involves creating a streaming TableEnvironment, defining two sink tables ('print_sink_table' and 'black_hole_sink_table'), adding insert statements to a StatementSet, and then printing the explained plan.

```python
# using a stream TableEnvironment
from pyflink.table import EnvironmentSettings, TableEnvironment
from pyflink.table.expressions import col

env_settings = EnvironmentSettings.in_streaming_mode()
table_env = TableEnvironment.create(env_settings)

table1 = table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['id', 'data'])
table2 = table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['id', 'data'])
table_env.execute_sql("""
    CREATE TABLE print_sink_table (
        id BIGINT, 
        data VARCHAR 
    ) WITH (
        'connector' = 'print'
    )
""")
table_env.execute_sql("""
    CREATE TABLE black_hole_sink_table (
        id BIGINT, 
        data VARCHAR 
    ) WITH (
        'connector' = 'blackhole'
    )
""")

statement_set = table_env.create_statement_set()

statement_set.add_insert("print_sink_table", table1.where(col("data").like('H%')))
statement_set.add_insert("black_hole_sink_table", table2)

print(statement_set.explain())

```

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

### Install PyFlink Libraries

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/flinkdev/building

Install the `apache-flink-libraries` package from the generated sdist file. The file is located in `./flink-python/apache-flink-libraries/dist/`.

```bash
python -m pip install apache-flink-libraries/dist/*.tar.gz

```

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

### Java Keyed DataStream Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/state

Demonstrates how to partition a DataStream using a KeySelector function in Java to enable keyed state operations.

```java
public class WC {
  public String word;
  public int count;

  public String getWord() { return word; }
}
DataStream<WC> words = // [...]
KeyedStream<WC> keyed = words
  .keyBy(WC::getWord);
```

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

### Install PyFlink

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/datastream_tutorial

Install the PyFlink package using pip. This is a prerequisite for using the Python DataStream API.

```bash
python -m pip install apache-flink

```

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

### Flink SQL CLI Session Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/use

Demonstrates a typical Flink SQL CLI session, including creating a catalog, using it, creating a database, using the database, and enabling modules.

```sql
Flink SQL> CREATE CATALOG cat1 WITH (...);
[INFO] Catalog has been created.

Flink SQL> SHOW CATALOGS;
default_catalog
cat1

Flink SQL> USE CATALOG cat1;

Flink SQL> SHOW DATABASES;

Flink SQL> CREATE DATABASE db1 WITH (...);
[INFO] Database has been created.

Flink SQL> SHOW DATABASES;
db1

Flink SQL> USE db1;

Flink SQL> USE MODULES hive;
[INFO] Use modules succeeded!
Flink SQL> SHOW FULL MODULES;
+-------------+-------+
| module name |  used |
+-------------+-------+
|        hive |  true |
|        core | false |
+-------------+-------+
2 rows in set

```

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

### Create Kafka Table with Format and Kafka Metadata

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/kafka

Example showing how to access both Kafka connector metadata and Debezium format metadata fields.

```sql
CREATE TABLE KafkaTable (
  `event_time` TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,  -- from Debezium format
  `origin_table` STRING METADATA FROM 'value.source.table' VIRTUAL, -- from Debezium format
  `partition_id` BIGINT METADATA FROM 'partition' VIRTUAL,  -- from Kafka connector
  `offset` BIGINT METADATA VIRTUAL,  -- from Kafka connector
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json'
);

```

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

### Install PyFlink Package

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/installation

Install the Apache Flink Python package (PyFlink) from PyPi using pip.

```bash
$ python -m pip install apache-flink==1.15.4

```

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

### Start the Flink Operations Playground

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/try-flink/flink-operations-playground

Start the Flink operations playground environment in detached mode using docker-compose.

```bash
docker-compose up -d
```

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

### Example External Resource Configuration

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/advanced/external_resources

This configuration defines two external resources, 'gpu' and 'fpga', specifying their driver factories, amounts, and integration details for Yarn and custom parameters.

```properties
external-resources: gpu;fpga

external-resource.gpu.driver-factory.class: org.apache.flink.externalresource.gpu.GPUDriverFactory
external-resource.gpu.amount: 2
external-resource.gpu.param.discovery-script.args: --enable-coordination

external-resource.fpga.driver-factory.class: org.apache.flink.externalresource.fpga.FPGADriverFactory
external-resource.fpga.amount: 1
external-resource.fpga.yarn.config-key: yarn.io/fpga

```

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

### Start Flink TaskManager

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

Initiate a Flink TaskManager process. Multiple TaskManagers can be started to provide more resources for your application.

```bash
$ ./bin/taskmanager.sh start

```

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

### Create Table with Filesystem Connector and ORC Format

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/orc

Example SQL statement to create a partitioned table using the Filesystem connector and specifying ORC as the data format. The 'path' and 'format' options are crucial for configuration.

```sql
CREATE TABLE user_behavior (
  user_id BIGINT,
  item_id BIGINT,
  category_id BIGINT,
  behavior STRING,
  ts TIMESTAMP(3),
  dt STRING
) PARTITIONED BY (dt) WITH (
 'connector' = 'filesystem',
 'path' = '/tmp/user_behavior',
 'format' = 'orc'
)
```

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

### Pulsar Start from Latest Message

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/pulsar

Configure the Pulsar source to start consuming from the latest available message in the topic.

```java
StartCursor.latest();
```

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

### Pulsar Start from Earliest Message

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/pulsar

Configure the Pulsar source to start consuming from the earliest available message in the topic.

```java
StartCursor.earliest();
```

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

### Create Kafka Table with JSON Key and Value Formats

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/kafka

Example demonstrating how to configure both key and value formats for a Kafka table using JSON. It includes options for key fields and value field inclusion.

```sql
CREATE TABLE KafkaTable (
  `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING
) WITH (
  'connector' = 'kafka',
  ...

  'key.format' = 'json',
  'key.json.ignore-parse-errors' = 'true',
  'key.fields' = 'user_id;item_id',

  'value.format' = 'json',
  'value.json.fail-on-missing-field' = 'false',
  'value.fields-include' = 'ALL'
)

```

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

### Start or Stop the History Server

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/advanced/historyserver

Use this script to manage the HistoryServer process. It can be started, run in the foreground, or stopped.

```bash
# Start or stop the HistoryServer
bin/historyserver.sh (start|start-foreground|stop)

```

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

### Example Output

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table_api_tutorial

This is an example of the execution results you might see after running the Flink Python Table API program.

```text
+I[To, 1]
+I[be,, 1]
+I[or, 1]
+I[not, 1]
...
```

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

### Create and Use JDBC Catalog (Java)

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/jdbc

Shows how to programmatically create a JDBC catalog instance and register it with the TableEnvironment in Java.

```java
EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);

String name            = "my_catalog";
String defaultDatabase = "mydb";
String username        = "...";
String password        = "...";
String baseUrl         = "..."

JdbcCatalog catalog = new JdbcCatalog(name, defaultDatabase, username, password, baseUrl);
tableEnv.registerCatalog("my_catalog", catalog);

// set the JdbcCatalog as the current catalog of the session
tableEnv.useCatalog("my_catalog");


```

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

### Canal JSON Changelog Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/canal

An example of an update operation captured in Canal JSON format from a MySQL table.

```json
{
  "data": [
    {
      "id": "111",
      "name": "scooter",
      "description": "Big 2-wheel scooter",
      "weight": "5.18"
    }
  ],
  "database": "inventory",
  "es": 1589373560000,
  "id": 9,
  "isDdl": false,
  "mysqlType": {
    "id": "INTEGER",
    "name": "VARCHAR(255)",
    "description": "VARCHAR(512)",
    "weight": "FLOAT"
  },
  "old": [
    {
      "weight": "5.15"
    }
  ],
  "pkNames": [
    "id"
  ],
  "sql": "",
  "sqlType": {
    "id": 4,
    "name": 12,
    "description": 12,
    "weight": 7
  },
  "table": "products",
  "ts": 1589373560798,
  "type": "UPDATE"
}
```

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

### Execute SHOW Statements in Java

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/show

Demonstrates how to execute various SHOW statements using the executeSql() method of the TableEnvironment in Java. Includes examples for catalogs, databases, tables, views, columns, functions, and modules.

```Java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

// show catalogs
tEnv.executeSql("SHOW CATALOGS").print();
// +-----------------+
// |    catalog name |
// +-----------------+
// | default_catalog |
// +-----------------+

// show current catalog
tEnv.executeSql("SHOW CURRENT CATALOG").print();
// +----------------------+
// | current catalog name |
// +----------------------+
// |      default_catalog |
// +----------------------+

// show databases
tEnv.executeSql("SHOW DATABASES").print();
// +------------------+
// |    database name |
// +------------------+
// | default_database |
// +------------------+

// show current database
tEnv.executeSql("SHOW CURRENT DATABASE").print();
// +-----------------------+
// | current database name |
// +-----------------------+
// |      default_database |
// +-----------------------+

// create a table
tEnv.executeSql("CREATE TABLE my_table (...) WITH (...)");
// show tables
tEnv.executeSql("SHOW TABLES").print();
// +------------+
// | table name |
// +------------+ 
// |   my_table |
// +------------+

// show create table
tEnv.executeSql("SHOW CREATE TABLE my_table").print();
// CREATE TABLE `default_catalog`.`default_db`.`my_table` (
//   ...
// ) WITH (
//   ...
// )

// show columns
tEnv.executeSql("SHOW COLUMNS FROM my_table LIKE '%f%'").print();
// +--------+-------+------+-----+--------+-----------+
// |   name |  type | null | key | extras | watermark |
// +--------+-------+------+-----+--------+-----------+
// | field2 | BYTES | true |     |        |           |
// +--------+-------+------+-----+--------+-----------+


// create a view
tEnv.executeSql("CREATE VIEW my_view AS SELECT * FROM my_table");
// show views
tEnv.executeSql("SHOW VIEWS").print();
// +-----------+
// | view name |
// +-----------+
// |   my_view |
// +-----------+

// show create view
tEnv.executeSql("SHOW CREATE VIEW my_view").print();
// CREATE VIEW `default_catalog`.`default_db`.`my_view`(`field1`, `field2`, ...) as
// SELECT *
// FROM `default_catalog`.`default_database`.`my_table`

// show functions
tEnv.executeSql("SHOW FUNCTIONS").print();
// +---------------+ 
// | function name |
// +---------------+ 
// |           mod |
// |        sha256 |
// |           ... |
// +---------------+ 

// create a user defined function
tEnv.executeSql("CREATE FUNCTION f1 AS ...");
// show user defined functions
tEnv.executeSql("SHOW USER FUNCTIONS").print();
// +---------------+ 
// | function name |
// +---------------+ 
// |            f1 |
// |           ... |
// +---------------+ 

// show modules
tEnv.executeSql("SHOW MODULES").print();
// +-------------+
// | module name |
// +-------------+
// |        core |
// +-------------+

// show full modules
tEnv.executeSql("SHOW FULL MODULES").print();
// +-------------+-------+
// | module name |  used |
// +-------------+-------+
// |        core |  true |
// |        hive | false |
// +-------------+-------+

```