Concepts & Common API

The Table API and SQL are integrated in a joint API. The central concept of this API is a Table which serves as input and output of queries. This document shows the common structure of programs with Table API and SQL queries, how to register a Table, how to query a Table, and how to emit a Table.

Structure of Table API and SQL Programs

The following code example shows the common structure of Table API and SQL programs.

Java

  2. // create a TableEnvironment for batch or streaming execution
  3. TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  5. // create an input Table
  6. tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )");
  7. // register an output Table
  8. tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )");
  10. // create a Table object from a Table API query
  11. Table table2 = tableEnv.from("table1").select(...);
  12. // create a Table object from a SQL query
  13. Table table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");
  15. // emit a Table API result Table to a TableSink, same for SQL result
  16. TableResult tableResult = table2.executeInsert("outputTable");
  17. tableResult...

Scala

  2. // create a TableEnvironment for batch or streaming execution
  3. val tableEnv = ... // see "Create a TableEnvironment" section
  5. // create an input Table
  6. tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )")
  7. // register an output Table
  8. tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )")
  10. // create a Table from a Table API query
  11. val table2 = tableEnv.from("table1").select(...)
  12. // create a Table from a SQL query
  13. val table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ...")
  15. // emit a Table API result Table to a TableSink, same for SQL result
  16. val tableResult = table2.executeInsert("outputTable")
  17. tableResult...

Python

  2. # create a TableEnvironment for batch or streaming execution
  3. table_env = ... # see "Create a TableEnvironment" section
  5. # register an input Table
  6. table_env.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )")
  7. # register an output Table
  8. table_env.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )")
  10. # create a Table from a Table API query
  11. table2 = table_env.from_path("table1").select(...)
  12. # create a Table from a SQL query
  13. table3 = table_env.sql_query("SELECT ... FROM table1 ...")
  15. # emit a Table API result Table to a TableSink, same for SQL result
  16. table_result = table3.execute_insert("outputTable")
  17. table_result...

Table API and SQL queries can be easily integrated with and embedded into DataStream programs. Have a look at the DataStream API Integration page to learn how DataStreams can be converted into Tables and vice versa.

Create a TableEnvironment

The TableEnvironment is the entrypoint for Table API and SQL integration and is responsible for:

  • Registering a Table in the internal catalog
  • Registering catalogs
  • Loading pluggable modules
  • Executing SQL queries
  • Registering a user-defined (scalar, table, or aggregation) function
  • Converting between DataStream and Table (in case of StreamTableEnvironment)

A Table is always bound to a specific TableEnvironment. It is not possible to combine tables of different TableEnvironments in the same query, e.g., to join or union them. A TableEnvironment is created by calling the static TableEnvironment.create() method.

Java

  1. import org.apache.flink.table.api.EnvironmentSettings;
  2. import org.apache.flink.table.api.TableEnvironment;
  4. EnvironmentSettings settings = EnvironmentSettings
  5.     .newInstance()
  6.     .inStreamingMode()
  7.     //.inBatchMode()
  8.     .build();
  10. TableEnvironment tEnv = TableEnvironment.create(setting);

Scala

  1. import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
  3. val settings = EnvironmentSettings
  4.     .newInstance()
  5.     .inStreamingMode()
  6.     //.inBatchMode()
  7.     .build()
  9. val tEnv = TableEnvironment.create(setting)

Python

  1. from pyflink.table import EnvironmentSettings, TableEnvironment
  3. # create a blink streaming TableEnvironment
  4. env_settings = EnvironmentSettings.new_instance().use_blink_planner().build()
  5. table_env = TableEnvironment.create(env_settings)
  7. # create a blink batch TableEnvironment
  8. env_settings = EnvironmentSettings.new_instance().in_batch_mode().build()
  9. table_env = TableEnvironment.create(env_settings)

Alternatively, users can create a StreamTableEnvironment from an existing StreamExecutionEnvironment to interoperate with the DataStream API.

Java

  1. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  2. import org.apache.flink.table.api.EnvironmentSettings;
  3. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  5. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  6. StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

Scala

  1. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  2. import org.apache.flink.table.api.EnvironmentSettings
  3. import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
  5. val env = StreamExecutionEnvironment.getExecutionEnvironment
  6. val tEnv = StreamTableEnvironment.create(env)

Python

  1. from pyflink.datastream import StreamExecutionEnvironment
  2. from pyflink.table import StreamTableEnvironment
  4. s_env = StreamExecutionEnvironment.get_execution_environment()
  5. t_env = StreamTableEnvironment.create(s_env)

Create Tables in the Catalog

A TableEnvironment maintains a map of catalogs of tables which are created with an identifier. Each identifier consists of 3 parts: catalog name, database name and object name. If a catalog or database is not specified, the current default value will be used (see examples in the Table identifier expanding section).

Tables can be either virtual (VIEWS) or regular (TABLES). VIEWS can be created from an existing Table object, usually the result of a Table API or SQL query. TABLES describe external data, such as a file, database table, or message queue.

Temporary vs Permanent tables.

Tables may either be temporary, and tied to the lifecycle of a single Flink session, or permanent, and visible across multiple Flink sessions and clusters.

Permanent tables require a catalog (such as Hive Metastore) to maintain metadata about the table. Once a permanent table is created, it is visible to any Flink session that is connected to the catalog and will continue to exist until the table is explicitly dropped.

On the other hand, temporary tables are always stored in memory and only exist for the duration of the Flink session they are created within. These tables are not visible to other sessions. They are not bound to any catalog or database but can be created in the namespace of one. Temporary tables are not dropped if their corresponding database is removed.

Shadowing

It is possible to register a temporary table with the same identifier as an existing permanent table. The temporary table shadows the permanent one and makes the permanent table inaccessible as long as the temporary one exists. All queries with that identifier will be executed against the temporary table.

This might be useful for experimentation. It allows running exactly the same query first against a temporary table that e.g. has just a subset of data, or the data is obfuscated. Once verified that the query is correct it can be run against the real production table.

Create a Table

Virtual Tables

A Table API object corresponds to a VIEW (virtual table) in a SQL terms. It encapsulates a logical query plan. It can be created in a catalog as follows:

Java

  1. // get a TableEnvironment
  2. TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // table is the result of a simple projection query 
  5. Table projTable = tableEnv.from("X").select(...);
  7. // register the Table projTable as table "projectedTable"
  8. tableEnv.createTemporaryView("projectedTable", projTable);

Scala

  1. // get a TableEnvironment
  2. val tableEnv = ... // see "Create a TableEnvironment" section
  4. // table is the result of a simple projection query 
  5. val projTable: Table = tableEnv.from("X").select(...)
  7. // register the Table projTable as table "projectedTable"
  8. tableEnv.createTemporaryView("projectedTable", projTable)

Python

  1. # get a TableEnvironment
  2. table_env = ... # see "Create a TableEnvironment" section
  4. # table is the result of a simple projection query 
  5. proj_table = table_env.from_path("X").select(...)
  7. # register the Table projTable as table "projectedTable"
  8. table_env.register_table("projectedTable", proj_table)

Note: Table objects are similar to VIEW’s from relational database systems, i.e., the query that defines the Table is not optimized but will be inlined when another query references the registered Table. If multiple queries reference the same registered Table, it will be inlined for each referencing query and executed multiple times, i.e., the result of the registered Table will not be shared.

Connector Tables

It is also possible to create a TABLE as known from relational databases from a connector declaration. The connector describes the external system that stores the data of a table. Storage systems such as Apache Kafka or a regular file system can be declared here.

  1. tableEnvironment.executeSql("CREATE [TEMPORARY] TABLE MyTable (...) WITH (...)")

Expanding Table identifiers

Tables are always registered with a 3-part identifier consisting of catalog, database, and table name.

Users can set one catalog and one database inside it to be the “current catalog” and “current database”. With them, the first two parts in the 3-parts identifier mentioned above can be optional - if they are not provided, the current catalog and current database will be referred. Users can switch the current catalog and current database via table API or SQL.

Identifiers follow SQL requirements which means that they can be escaped with a backtick character (` ).

Java

  1. TableEnvironment tEnv = ...;
  2. tEnv.useCatalog("custom_catalog");
  3. tEnv.useDatabase("custom_database");
  5. Table table = ...;
  7. // register the view named 'exampleView' in the catalog named 'custom_catalog'
  8. // in the database named 'custom_database' 
  9. tableEnv.createTemporaryView("exampleView", table);
  11. // register the view named 'exampleView' in the catalog named 'custom_catalog'
  12. // in the database named 'other_database' 
  13. tableEnv.createTemporaryView("other_database.exampleView", table);
  15. // register the view named 'example.View' in the catalog named 'custom_catalog'
  16. // in the database named 'custom_database' 
  17. tableEnv.createTemporaryView("`example.View`", table);
  19. // register the view named 'exampleView' in the catalog named 'other_catalog'
  20. // in the database named 'other_database' 
  21. tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table);

Scala

  1. // get a TableEnvironment
  2. val tEnv: TableEnvironment = ...;
  3. tEnv.useCatalog("custom_catalog")
  4. tEnv.useDatabase("custom_database")
  6. val table: Table = ...;
  8. // register the view named 'exampleView' in the catalog named 'custom_catalog'
  9. // in the database named 'custom_database' 
  10. tableEnv.createTemporaryView("exampleView", table)
  12. // register the view named 'exampleView' in the catalog named 'custom_catalog'
  13. // in the database named 'other_database' 
  14. tableEnv.createTemporaryView("other_database.exampleView", table)
  16. // register the view named 'example.View' in the catalog named 'custom_catalog'
  17. // in the database named 'custom_database' 
  18. tableEnv.createTemporaryView("`example.View`", table)
  20. // register the view named 'exampleView' in the catalog named 'other_catalog'
  21. // in the database named 'other_database' 
  22. tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table)

Query a Table

Table API

The Table API is a language-integrated query API for Scala and Java. In contrast to SQL, queries are not specified as Strings but are composed step-by-step in the host language.

The API is based on the Table class which represents a table (streaming or batch) and offers methods to apply relational operations. These methods return a new Table object, which represents the result of applying the relational operation on the input Table. Some relational operations are composed of multiple method calls such as table.groupBy(...).select(), where groupBy(...) specifies a grouping of table, and select(...) the projection on the grouping of table.

The Table API document describes all Table API operations that are supported on streaming and batch tables.

The following example shows a simple Table API aggregation query:

Java

  1. // get a TableEnvironment
  2. TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // register Orders table
  6. // scan registered Orders table
  7. Table orders = tableEnv.from("Orders");
  8. // compute revenue for all customers from France
  9. Table revenue = orders
  10.   .filter($("cCountry").isEqual("FRANCE"))
  11.   .groupBy($("cID"), $("cName"))
  12.   .select($("cID"), $("cName"), $("revenue").sum().as("revSum"));
  14. // emit or convert Table
  15. // execute query

Scala

  1. // get a TableEnvironment
  2. val tableEnv = ... // see "Create a TableEnvironment" section
  4. // register Orders table
  6. // scan registered Orders table
  7. val orders = tableEnv.from("Orders")
  8. // compute revenue for all customers from France
  9. val revenue = orders
  10.   .filter($"cCountry" === "FRANCE")
  11.   .groupBy($"cID", $"cName")
  12.   .select($"cID", $"cName", $"revenue".sum AS "revSum")
  14. // emit or convert Table
  15. // execute query

Note: The Scala Table API uses Scala String interpolation that starts with a dollar sign ($) to reference the attributes of a Table. The Table API uses Scala implicits. Make sure to import

  • org.apache.flink.table.api._ - for implicit expression conversions

  • org.apache.flink.api.scala._ and org.apache.flink.table.api.bridge.scala._ if you want to convert from/to DataStream.

    Python

  1. # get a TableEnvironment
  2. table_env = # see "Create a TableEnvironment" section
  4. # register Orders table
  6. # scan registered Orders table
  7. orders = table_env.from_path("Orders")
  8. # compute revenue for all customers from France
  9. revenue = orders \
  10.     .filter(orders.cCountry == 'FRANCE') \
  11.     .group_by(orders.cID, orders.cName) \
  12.     .select(orders.cID, orders.cName, orders.revenue.sum.alias('revSum'))
  14. # emit or convert Table
  15. # execute query

SQL

Flink’s SQL integration is based on Apache Calcite, which implements the SQL standard. SQL queries are specified as regular Strings.

The SQL document describes Flink’s SQL support for streaming and batch tables.

The following example shows how to specify a query and return the result as a Table.

Java

  1. // get a TableEnvironment
  2. TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // register Orders table
  6. // compute revenue for all customers from France
  7. Table revenue = tableEnv.sqlQuery(
  8.     "SELECT cID, cName, SUM(revenue) AS revSum " +
  9.     "FROM Orders " +
  10.     "WHERE cCountry = 'FRANCE' " +
  11.     "GROUP BY cID, cName"
  12.   );
  14. // emit or convert Table
  15. // execute query

Scala

  1. // get a TableEnvironment
  2. val tableEnv = ... // see "Create a TableEnvironment" section
  4. // register Orders table
  6. // compute revenue for all customers from France
  7. val revenue = tableEnv.sqlQuery("""
  8.   |SELECT cID, cName, SUM(revenue) AS revSum
  9.   |FROM Orders
  10.   |WHERE cCountry = 'FRANCE'
  11.   |GROUP BY cID, cName
  12.   """.stripMargin)
  14. // emit or convert Table
  15. // execute query

Python

  1. # get a TableEnvironment
  2. table_env = ... # see "Create a TableEnvironment" section
  4. # register Orders table
  6. # compute revenue for all customers from France
  7. revenue = table_env.sql_query(
  8.     "SELECT cID, cName, SUM(revenue) AS revSum "
  9.     "FROM Orders "
  10.     "WHERE cCountry = 'FRANCE' "
  11.     "GROUP BY cID, cName"
  12. )
  14. # emit or convert Table
  15. # execute query

The following example shows how to specify an update query that inserts its result into a registered table.

Java

  1. // get a TableEnvironment
  2. TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // register "Orders" table
  5. // register "RevenueFrance" output table
  7. // compute revenue for all customers from France and emit to "RevenueFrance"
  8. tableEnv.executeSql(
  9.     "INSERT INTO RevenueFrance " +
  10.     "SELECT cID, cName, SUM(revenue) AS revSum " +
  11.     "FROM Orders " +
  12.     "WHERE cCountry = 'FRANCE' " +
  13.     "GROUP BY cID, cName"
  14.   );

Scala

  1. // get a TableEnvironment
  2. val tableEnv = ... // see "Create a TableEnvironment" section
  4. // register "Orders" table
  5. // register "RevenueFrance" output table
  7. // compute revenue for all customers from France and emit to "RevenueFrance"
  8. tableEnv.executeSql("""
  9.   |INSERT INTO RevenueFrance
  10.   |SELECT cID, cName, SUM(revenue) AS revSum
  11.   |FROM Orders
  12.   |WHERE cCountry = 'FRANCE'
  13.   |GROUP BY cID, cName
  14.   """.stripMargin)

Python

  1. # get a TableEnvironment
  2. table_env = ... # see "Create a TableEnvironment" section
  4. # register "Orders" table
  5. # register "RevenueFrance" output table
  7. # compute revenue for all customers from France and emit to "RevenueFrance"
  8. table_env.execute_sql(
  9.     "INSERT INTO RevenueFrance "
  10.     "SELECT cID, cName, SUM(revenue) AS revSum "
  11.     "FROM Orders "
  12.     "WHERE cCountry = 'FRANCE' "
  13.     "GROUP BY cID, cName"
  14. )

Mixing Table API and SQL

Table API and SQL queries can be easily mixed because both return Table objects:

  • A Table API query can be defined on the Table object returned by a SQL query.
  • A SQL query can be defined on the result of a Table API query by registering the resulting Table in the TableEnvironment and referencing it in the FROM clause of the SQL query.

Emit a Table

A Table is emitted by writing it to a TableSink. A TableSink is a generic interface to support a wide variety of file formats (e.g. CSV, Apache Parquet, Apache Avro), storage systems (e.g., JDBC, Apache HBase, Apache Cassandra, Elasticsearch), or messaging systems (e.g., Apache Kafka, RabbitMQ).

A batch Table can only be written to a BatchTableSink, while a streaming Table requires either an AppendStreamTableSink, a RetractStreamTableSink, or an UpsertStreamTableSink.

Please see the documentation about Table Sources & Sinks for details about available sinks and instructions for how to implement a custom DynamicTableSink.

The Table.executeInsert(String tableName) method emits the Table to a registered TableSink. The method looks up the TableSink from the catalog by the name and validates that the schema of the Table is identical to the schema of the TableSink.

The following examples shows how to emit a Table:

Java

  1. // get a TableEnvironment
  2. TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // create an output Table
  5. final Schema schema = new Schema()
  6.     .field("a", DataTypes.INT())
  7.     .field("b", DataTypes.STRING())
  8.     .field("c", DataTypes.BIGINT());
  10. tableEnv.connect(new FileSystem().path("/path/to/file"))
  11.     .withFormat(new Csv().fieldDelimiter('|').deriveSchema())
  12.     .withSchema(schema)
  13.     .createTemporaryTable("CsvSinkTable");
  15. // compute a result Table using Table API operators and/or SQL queries
  16. Table result = ...
  18. // emit the result Table to the registered TableSink
  19. result.executeInsert("CsvSinkTable");

Scala

  1. // get a TableEnvironment
  2. val tableEnv = ... // see "Create a TableEnvironment" section
  4. // create an output Table
  5. val schema = new Schema()
  6.     .field("a", DataTypes.INT())
  7.     .field("b", DataTypes.STRING())
  8.     .field("c", DataTypes.BIGINT())
  10. tableEnv.connect(new FileSystem().path("/path/to/file"))
  11.     .withFormat(new Csv().fieldDelimiter('|').deriveSchema())
  12.     .withSchema(schema)
  13.     .createTemporaryTable("CsvSinkTable")
  15. // compute a result Table using Table API operators and/or SQL queries
  16. val result: Table = ...
  18. // emit the result Table to the registered TableSink
  19. result.executeInsert("CsvSinkTable")

Python

  1. # get a TableEnvironment
  2. table_env = ... # see "Create a TableEnvironment" section
  4. # create a TableSink
  5. table_env.connect(FileSystem().path("/path/to/file")))
  6.     .with_format(Csv()
  7.                  .field_delimiter(',')
  8.                  .deriveSchema())
  9.     .with_schema(Schema()
  10.                  .field("a", DataTypes.INT())
  11.                  .field("b", DataTypes.STRING())
  12.                  .field("c", DataTypes.BIGINT()))
  13.     .create_temporary_table("CsvSinkTable")
  15. # compute a result Table using Table API operators and/or SQL queries
  16. result = ...
  18. # emit the result Table to the registered TableSink
  19. result.execute_insert("CsvSinkTable")

Translate and Execute a Query

The behavior of translating and executing a query is different for the two planners. Table API and SQL queries are translated into DataStream programs whether their input is streaming or batch. A query is internally represented as a logical query plan and is translated in two phases:

  1. Optimization of the logical plan,
  2. Translation into a DataStream program.

a Table API or SQL query is translated when:

  • TableEnvironment.executeSql() is called. This method is used for executing a given statement, and the sql query is translated immediately once this method is called.
  • Table.executeInsert() is called. This method is used for inserting the table content to the given sink path, and the Table API is translated immediately once this method is called.
  • Table.execute() is called. This method is used for collecting the table content to local client, and the Table API is translated immediately once this method is called.
  • StatementSet.execute() is called. A Table (emitted to a sink through StatementSet.addInsert()) or an INSERT statement (specified through StatementSet.addInsertSql()) will be buffered in StatementSet first. They are translated once StatementSet.execute() is called. All sinks will be optimized into one DAG.
  • A Table is translated when it is converted into a DataStream (see Integration with DataStream). Once translated, it’s a regular DataStream program and is executed when StreamExecutionEnvironment.execute() is called.

Query Optimization

Apache Flink leverages and extends Apache Calcite to perform sophisticated query optimization. This includes a series of rule and cost-based optimizations such as:

  • Subquery decorrelation based on Apache Calcite
  • Project pruning
  • Partition pruning
  • Filter push-down
  • Sub-plan deduplication to avoid duplicate computation
  • Special subquery rewriting, including two parts:
    • Converts IN and EXISTS into left semi-joins
    • Converts NOT IN and NOT EXISTS into left anti-join
  • Optional join reordering
    • Enabled via table.optimizer.join-reorder-enabled

Note: IN/EXISTS/NOT IN/NOT EXISTS are currently only supported in conjunctive conditions in subquery rewriting.

The optimizer makes intelligent decisions, based not only on the plan but also rich statistics available from the data sources and fine-grain costs for each operator such as io, cpu, network, and memory.

Advanced users may provide custom optimizations via a CalciteConfig object that can be provided to the table environment by calling TableEnvironment#getConfig#setPlannerConfig.

Explaining a Table

The Table API provides a mechanism to explain the logical and optimized query plans to compute a Table. This is done through the Table.explain() method or StatementSet.explain() method. Table.explain()returns the plan of a Table. StatementSet.explain() returns the plan of multiple sinks. It returns a String describing three plans:

  1. the Abstract Syntax Tree of the relational query, i.e., the unoptimized logical query plan,
  2. the optimized logical query plan, and
  3. the physical execution plan.

TableEnvironment.explainSql() and TableEnvironment.executeSql() support execute a EXPLAIN statement to get the plans, Please refer to EXPLAIN page.

The following code shows an example and the corresponding output for given Table using Table.explain() method:

Java

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  4. DataStream<Tuple2<Integer, String>> stream1 = env.fromElements(new Tuple2<>(1, "hello"));
  5. DataStream<Tuple2<Integer, String>> stream2 = env.fromElements(new Tuple2<>(1, "hello"));
  7. // explain Table API
  8. Table table1 = tEnv.fromDataStream(stream1, $("count"), $("word"));
  9. Table table2 = tEnv.fromDataStream(stream2, $("count"), $("word"));
  10. Table table = table1
  11.   .where($("word").like("F%"))
  12.   .unionAll(table2);
  14. System.out.println(table.explain());

Scala

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment
  2. val tEnv = StreamTableEnvironment.create(env)
  4. val table1 = env.fromElements((1, "hello")).toTable(tEnv, $"count", $"word")
  5. val table2 = env.fromElements((1, "hello")).toTable(tEnv, $"count", $"word")
  6. val table = table1
  7.   .where($"word".like("F%"))
  8.   .unionAll(table2)
  10. println(table.explain())

Python

  1. env = StreamExecutionEnvironment.get_execution_environment()
  2. t_env = StreamTableEnvironment.create(env)
  4. table1 = t_env.from_elements([(1, "hello")], ["count", "word"])
  5. table2 = t_env.from_elements([(1, "hello")], ["count", "word"])
  6. table = table1 \
  7.     .where(table1.word.like('F%')) \
  8.     .union_all(table2)
  9. print(table.explain())

The result of the above exmaple is

Explain ↕

  2. == Abstract Syntax Tree ==
  3. LogicalUnion(all=[true])
  4.   LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  5.     FlinkLogicalDataStreamScan(id=[1], fields=[count, word])
  6.   FlinkLogicalDataStreamScan(id=[2], fields=[count, word])
  8. == Optimized Logical Plan ==
  9. DataStreamUnion(all=[true], union all=[count, word])
  10.   DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  11.     DataStreamScan(id=[1], fields=[count, word])
  12.   DataStreamScan(id=[2], fields=[count, word])
  14. == Physical Execution Plan ==
  15. Stage 1 : Data Source
  16.     content : collect elements with CollectionInputFormat
  18. Stage 2 : Data Source
  19.     content : collect elements with CollectionInputFormat
  21.     Stage 3 : Operator
  22.         content : from: (count, word)
  23.         ship_strategy : REBALANCE
  25.         Stage 4 : Operator
  26.             content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
  27.             ship_strategy : FORWARD
  29.             Stage 5 : Operator
  30.                 content : from: (count, word)
  31.                 ship_strategy : REBALANCE

The following code shows an example and the corresponding output for multiple-sinks plan using StatementSet.explain() method:

Java

  2. EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  3. TableEnvironment tEnv = TableEnvironment.create(settings);
  5. final Schema schema = new Schema()
  6.     .field("count", DataTypes.INT())
  7.     .field("word", DataTypes.STRING());
  9. tEnv.connect(new FileSystem().path("/source/path1"))
  10.     .withFormat(new Csv().deriveSchema())
  11.     .withSchema(schema)
  12.     .createTemporaryTable("MySource1");
  13. tEnv.connect(new FileSystem().path("/source/path2"))
  14.     .withFormat(new Csv().deriveSchema())
  15.     .withSchema(schema)
  16.     .createTemporaryTable("MySource2");
  17. tEnv.connect(new FileSystem().path("/sink/path1"))
  18.     .withFormat(new Csv().deriveSchema())
  19.     .withSchema(schema)
  20.     .createTemporaryTable("MySink1");
  21. tEnv.connect(new FileSystem().path("/sink/path2"))
  22.     .withFormat(new Csv().deriveSchema())
  23.     .withSchema(schema)
  24.     .createTemporaryTable("MySink2");
  26. StatementSet stmtSet = tEnv.createStatementSet();
  28. Table table1 = tEnv.from("MySource1").where($("word").like("F%"));
  29. stmtSet.addInsert("MySink1", table1);
  31. Table table2 = table1.unionAll(tEnv.from("MySource2"));
  32. stmtSet.addInsert("MySink2", table2);
  34. String explanation = stmtSet.explain();
  35. System.out.println(explanation);

Scala

  1. val settings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode.build
  2. val tEnv = TableEnvironment.create(settings)
  4. val schema = new Schema()
  5.     .field("count", DataTypes.INT())
  6.     .field("word", DataTypes.STRING())
  8. tEnv.connect(new FileSystem().path("/source/path1"))
  9.     .withFormat(new Csv().deriveSchema())
  10.     .withSchema(schema)
  11.     .createTemporaryTable("MySource1")
  12. tEnv.connect(new FileSystem().path("/source/path2"))
  13.     .withFormat(new Csv().deriveSchema())
  14.     .withSchema(schema)
  15.     .createTemporaryTable("MySource2")
  16. tEnv.connect(new FileSystem().path("/sink/path1"))
  17.     .withFormat(new Csv().deriveSchema())
  18.     .withSchema(schema)
  19.     .createTemporaryTable("MySink1")
  20. tEnv.connect(new FileSystem().path("/sink/path2"))
  21.     .withFormat(new Csv().deriveSchema())
  22.     .withSchema(schema)
  23.     .createTemporaryTable("MySink2")
  25. val stmtSet = tEnv.createStatementSet()
  27. val table1 = tEnv.from("MySource1").where($"word".like("F%"))
  28. stmtSet.addInsert("MySink1", table1)
  30. val table2 = table1.unionAll(tEnv.from("MySource2"))
  31. stmtSet.addInsert("MySink2", table2)
  33. val explanation = stmtSet.explain()
  34. println(explanation)

Python

  1. settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
  2. t_env = TableEnvironment.create(environment_settings=settings)
  4. schema = Schema()
  5.     .field("count", DataTypes.INT())
  6.     .field("word", DataTypes.STRING())
  8. t_env.connect(FileSystem().path("/source/path1")))
  9.     .with_format(Csv().deriveSchema())
  10.     .with_schema(schema)
  11.     .create_temporary_table("MySource1")
  12. t_env.connect(FileSystem().path("/source/path2")))
  13.     .with_format(Csv().deriveSchema())
  14.     .with_schema(schema)
  15.     .create_temporary_table("MySource2")
  16. t_env.connect(FileSystem().path("/sink/path1")))
  17.     .with_format(Csv().deriveSchema())
  18.     .with_schema(schema)
  19.     .create_temporary_table("MySink1")
  20. t_env.connect(FileSystem().path("/sink/path2")))
  21.     .with_format(Csv().deriveSchema())
  22.     .with_schema(schema)
  23.     .create_temporary_table("MySink2")
  25. stmt_set = t_env.create_statement_set()
  27. table1 = t_env.from_path("MySource1").where(col('word').like('F%'))
  28. stmt_set.add_insert("MySink1", table1)
  30. table2 = table1.union_all(t_env.from_path("MySource2"))
  31. stmt_set.add_insert("MySink2", table2)
  33. explanation = stmt_set.explain()
  34. print(explanation)

the result of multiple-sinks plan is

MultiTable Explain ↕

  1. == Abstract Syntax Tree ==
  2. LogicalLegacySink(name=[MySink1], fields=[count, word])
  3. +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  4.    +- LogicalTableScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]])
  6. LogicalLegacySink(name=[MySink2], fields=[count, word])
  7. +- LogicalUnion(all=[true])
  8.    :- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  9.    :  +- LogicalTableScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]])
  10.    +- LogicalTableScan(table=[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]])
  12. == Optimized Logical Plan ==
  13. Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')], reuse_id=[1])
  14. +- TableSourceScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]], fields=[count, word])
  16. LegacySink(name=[MySink1], fields=[count, word])
  17. +- Reused(reference_id=[1])
  19. LegacySink(name=[MySink2], fields=[count, word])
  20. +- Union(all=[true], union=[count, word])
  21.    :- Reused(reference_id=[1])
  22.    +- TableSourceScan(table=[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]], fields=[count, word])
  24. == Physical Execution Plan ==
  25. Stage 1 : Data Source
  26.     content : collect elements with CollectionInputFormat
  28.     Stage 2 : Operator
  29.         content : CsvTableSource(read fields: count, word)
  30.         ship_strategy : REBALANCE
  32.         Stage 3 : Operator
  33.             content : SourceConversion(table:Buffer(default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]), fields:(count, word))
  34.             ship_strategy : FORWARD
  36.             Stage 4 : Operator
  37.                 content : Calc(where: (word LIKE _UTF-16LE'F%'), select: (count, word))
  38.                 ship_strategy : FORWARD
  40.                 Stage 5 : Operator
  41.                     content : SinkConversionToRow
  42.                     ship_strategy : FORWARD
  44.                     Stage 6 : Operator
  45.                         content : Map
  46.                         ship_strategy : FORWARD
  48. Stage 8 : Data Source
  49.     content : collect elements with CollectionInputFormat
  51.     Stage 9 : Operator
  52.         content : CsvTableSource(read fields: count, word)
  53.         ship_strategy : REBALANCE
  55.         Stage 10 : Operator
  56.             content : SourceConversion(table:Buffer(default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]), fields:(count, word))
  57.             ship_strategy : FORWARD
  59.             Stage 12 : Operator
  60.                 content : SinkConversionToRow
  61.                 ship_strategy : FORWARD
  63.                 Stage 13 : Operator
  64.                     content : Map
  65.                     ship_strategy : FORWARD
  67.                     Stage 7 : Data Sink
  68.                         content : Sink: CsvTableSink(count, word)
  69.                         ship_strategy : FORWARD
  71.                         Stage 14 : Data Sink
  72.                             content : Sink: CsvTableSink(count, word)
  73.                             ship_strategy : FORWARD