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.

Main Differences Between the Two Planners

  • Blink treats batch jobs as a special case of streaming. As such, the conversion between Table and DataSet is also not supported, and batch jobs will not be translated into DateSet programs but translated into DataStream programs, the same as the streaming jobs.
  • The Blink planner does not support BatchTableSource, use bounded StreamTableSource instead of it.
  • The Blink planner only support the brand new Catalog and does not support ExternalCatalog which is deprecated.
  • The implementations of FilterableTableSource for the old planner and the Blink planner are incompatible. The old planner will push down PlannerExpressions into FilterableTableSource, while the Blink planner will push down Expressions.
  • String based key-value config options (Please see the documentation about Configuration for details) are only used for the Blink planner.
  • The implementation(CalciteConfig) of PlannerConfig in two planners is different.
  • The Blink planner will optimize multiple-sinks into one DAG (supported only on TableEnvironment, not on StreamTableEnvironment). The old planner will always optimize each sink into a new DAG, where all DAGs are independent of each other.
  • The old planner does not support catalog statistics now, while the Blink planner does.

Structure of Table API and SQL Programs

All Table API and SQL programs for batch and streaming follow the same pattern. The following code example shows the common structure of Table API and SQL programs.

  1. // create a TableEnvironment for specific planner batch or streaming
  2. TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // create a Table
  5. tableEnv.connect(...).createTemporaryTable("table1");
  6. // register an output Table
  7. tableEnv.connect(...).createTemporaryTable("outputTable");
  9. // create a Table object from a Table API query
  10. Table tapiResult = tableEnv.from("table1").select(...);
  11. // create a Table object from a SQL query
  12. Table sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");
  14. // emit a Table API result Table to a TableSink, same for SQL result
  15. tapiResult.insertInto("outputTable");
  17. // execute
  18. tableEnv.execute("java_job");
  1. // create a TableEnvironment for specific planner batch or streaming
  2. val tableEnv = ... // see "Create a TableEnvironment" section
  4. // create a Table
  5. tableEnv.connect(...).createTemporaryTable("table1")
  6. // register an output Table
  7. tableEnv.connect(...).createTemporaryTable("outputTable")
  9. // create a Table from a Table API query
  10. val tapiResult = tableEnv.from("table1").select(...)
  11. // create a Table from a SQL query
  12. val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ...")
  14. // emit a Table API result Table to a TableSink, same for SQL result
  15. tapiResult.insertInto("outputTable")
  17. // execute
  18. tableEnv.execute("scala_job")
  1. # create a TableEnvironment for specific planner batch or streaming
  2. table_env = ... # see "Create a TableEnvironment" section
  4. # register a Table
  5. table_env.connect(...).create_temporary_table("table1")
  7. # register an output Table
  8. table_env.connect(...).create_temporary_table("outputTable")
  10. # create a Table from a Table API query
  11. tapi_result = table_env.from_path("table1").select(...)
  12. # create a Table from a SQL query
  13. sql_result  = table_env.sql_query("SELECT ... FROM table1 ...")
  15. # emit a Table API result Table to a TableSink, same for SQL result
  16. tapi_result.insert_into("outputTable")
  18. # execute
  19. table_env.execute("python_job")

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

Create a TableEnvironment

The TableEnvironment is a central concept of the Table API and SQL integration. It 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 a DataStream or DataSet into a Table
  • Holding a reference to an ExecutionEnvironment or StreamExecutionEnvironment

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 BatchTableEnvironment.create() or StreamTableEnvironment.create() method with a StreamExecutionEnvironment or an ExecutionEnvironment and an optional TableConfig. The TableConfig can be used to configure the TableEnvironment or to customize the query optimization and translation process (see Query Optimization).

Make sure to choose the specific planner BatchTableEnvironment/StreamTableEnvironment that matches your programming language.

If both planner jars are on the classpath (the default behavior), you should explicitly set which planner to use in the current program.

  1. // **********************
  2. // FLINK STREAMING QUERY
  3. // **********************
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.EnvironmentSettings;
  6. import org.apache.flink.table.api.java.StreamTableEnvironment;
  8. EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
  9. StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
  10. StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);
  11. // or TableEnvironment fsTableEnv = TableEnvironment.create(fsSettings);
  13. // ******************
  14. // FLINK BATCH QUERY
  15. // ******************
  16. import org.apache.flink.api.java.ExecutionEnvironment;
  17. import org.apache.flink.table.api.java.BatchTableEnvironment;
  19. ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
  20. BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);
  22. // **********************
  23. // BLINK STREAMING QUERY
  24. // **********************
  25. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  26. import org.apache.flink.table.api.EnvironmentSettings;
  27. import org.apache.flink.table.api.java.StreamTableEnvironment;
  29. StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
  30. EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  31. StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
  32. // or TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);
  34. // ******************
  35. // BLINK BATCH QUERY
  36. // ******************
  37. import org.apache.flink.table.api.EnvironmentSettings;
  38. import org.apache.flink.table.api.TableEnvironment;
  40. EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
  41. TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);
  1. // **********************
  2. // FLINK STREAMING QUERY
  3. // **********************
  4. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  5. import org.apache.flink.table.api.EnvironmentSettings
  6. import org.apache.flink.table.api.scala.StreamTableEnvironment
  8. val fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
  9. val fsEnv = StreamExecutionEnvironment.getExecutionEnvironment
  10. val fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings)
  11. // or val fsTableEnv = TableEnvironment.create(fsSettings)
  13. // ******************
  14. // FLINK BATCH QUERY
  15. // ******************
  16. import org.apache.flink.api.scala.ExecutionEnvironment
  17. import org.apache.flink.table.api.scala.BatchTableEnvironment
  19. val fbEnv = ExecutionEnvironment.getExecutionEnvironment
  20. val fbTableEnv = BatchTableEnvironment.create(fbEnv)
  22. // **********************
  23. // BLINK STREAMING QUERY
  24. // **********************
  25. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  26. import org.apache.flink.table.api.EnvironmentSettings
  27. import org.apache.flink.table.api.scala.StreamTableEnvironment
  29. val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
  30. val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
  31. val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
  32. // or val bsTableEnv = TableEnvironment.create(bsSettings)
  34. // ******************
  35. // BLINK BATCH QUERY
  36. // ******************
  37. import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
  39. val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
  40. val bbTableEnv = TableEnvironment.create(bbSettings)
  1. # **********************
  2. # FLINK STREAMING QUERY
  3. # **********************
  4. from pyflink.datastream import StreamExecutionEnvironment
  5. from pyflink.table import StreamTableEnvironment, EnvironmentSettings
  7. f_s_env = StreamExecutionEnvironment.get_execution_environment()
  8. f_s_settings = EnvironmentSettings.new_instance().use_old_planner().in_streaming_mode().build()
  9. f_s_t_env = StreamTableEnvironment.create(f_s_env, environment_settings=f_s_settings)
  11. # ******************
  12. # FLINK BATCH QUERY
  13. # ******************
  14. from pyflink.dataset import ExecutionEnvironment
  15. from pyflink.table import BatchTableEnvironment
  17. f_b_env = ExecutionEnvironment.get_execution_environment()
  18. f_b_t_env = BatchTableEnvironment.create(f_b_env, table_config)
  20. # **********************
  21. # BLINK STREAMING QUERY
  22. # **********************
  23. from pyflink.datastream import StreamExecutionEnvironment
  24. from pyflink.table import StreamTableEnvironment, EnvironmentSettings
  26. b_s_env = StreamExecutionEnvironment.get_execution_environment()
  27. b_s_settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
  28. b_s_t_env = StreamTableEnvironment.create(b_s_env, environment_settings=b_s_settings)
  30. # ******************
  31. # BLINK BATCH QUERY
  32. # ******************
  33. from pyflink.table import EnvironmentSettings, BatchTableEnvironment
  35. b_b_settings = EnvironmentSettings.new_instance().use_blink_planner().in_batch_mode().build()
  36. b_b_t_env = BatchTableEnvironment.create(environment_settings=b_b_settings)

Note: If there is only one planner jar in /lib directory, you can use useAnyPlanner (use_any_planner for python) to create specific EnvironmentSettings.

Create Tables in the Catalog

A TableEnvironment maintains a map of catalogs of tables which are created with an identifier. Eachidentifier consists of 3 parts: catalog name, database name and object name. If a catalog or database is notspecified, 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 anexisting Table object, usually the result of a Table API or SQL query. TABLES describeexternal 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 Flinksession that is connected to the catalog and will continue to exist until the table is explicitlydropped.

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

Shadowing

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

This might be useful for experimentation. It allows running exactly the same query first against atemporary table that e.g. has just a subset of data, or the data is obfuscated. Once verified thatthe 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 logicalquery plan. It can be created in a catalog as follows:

  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);
  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)
  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 databasesystems, i.e., the query that defines the Table is not optimized but will be inlined when anotherquery 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 theregistered 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 Apacha Kafka or a regular file system can be declared here.

  1. tableEnvironment
  2.   .connect(...)
  3.   .withFormat(...)
  4.   .withSchema(...)
  5.   .inAppendMode()
  6.   .createTemporaryTable("MyTable")
  1. tableEnvironment
  2.   .connect(...)
  3.   .withFormat(...)
  4.   .withSchema(...)
  5.   .inAppendMode()
  6.   .createTemporaryTable("MyTable")
  1. table_environment \
  2.     .connect(...) \
  3.     .with_format(...) \
  4.     .with_schema(...) \
  5.     .in_append_mode() \
  6.     .create_temporary_table("MyTable")
  1. tableEnvironment.sqlUpdate("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 viatable API or SQL.

Identifiers follow SQL requirements which means that they can be escaped with a backtick character (`).Additionally all SQL reserved keywords must be escaped.

  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 'View' in the catalog named 'custom_catalog' in the
  16. // database named 'custom_database'. 'View' is a reserved keyword and must be escaped.  
  17. tableEnv.createTemporaryView("`View`", table);
  19. // register the view named 'example.View' in the catalog named 'custom_catalog'
  20. // in the database named 'custom_database' 
  21. tableEnv.createTemporaryView("`example.View`", table);
  23. // register the view named 'exampleView' in the catalog named 'other_catalog'
  24. // in the database named 'other_database' 
  25. tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table);
  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 'View' in the catalog named 'custom_catalog' in the
  17. // database named 'custom_database'. 'View' is a reserved keyword and must be escaped.  
  18. tableEnv.createTemporaryView("`View`", table)
  20. // register the view named 'example.View' in the catalog named 'custom_catalog'
  21. // in the database named 'custom_database' 
  22. tableEnv.createTemporaryView("`example.View`", table)
  24. // register the view named 'exampleView' in the catalog named 'other_catalog'
  25. // in the database named 'other_database' 
  26. 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:

  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 === 'FRANCE'")
  11.   .groupBy("cID, cName")
  12.   .select("cID, cName, revenue.sum AS revSum");
  14. // emit or convert Table
  15. // execute query
  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 Symbols, which start with a single tick (') to reference the attributes of a Table. The Table API uses Scala implicits. Make sure to import org.apache.flink.api.scala. and org.apache.flink.table.api.scala. in order to use Scala implicit conversions.

  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("cCountry === 'FRANCE'") \
  11.     .group_by("cID, cName") \
  12.     .select("cID, cName, revenue.sum AS 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.

  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
  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
  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.

  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.sqlUpdate(
  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.   );
  16. // execute query
  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.sqlUpdate("""
  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)
  16. // execute query
  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.sql_update(
  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. )
  16. # execute query

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 TableSink.

The Table.insertInto(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:

  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.LONG());
  10. tableEnv.connect(new FileSystem("/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 = ...
  17. // emit the result Table to the registered TableSink
  18. result.insertInto("CsvSinkTable");
  20. // execute the program
  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.LONG())
  10. tableEnv.connect(new FileSystem("/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.insertInto("CsvSinkTable")
  21. // execute the program
  1. # get a TableEnvironment
  2. table_env = ... # see "Create a TableEnvironment" section
  4. # create a TableSink
  5. t_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.insert_into("CsvSinkTable")
  21. # execute the program

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 or DataSet programs depending on whether their input is a streaming or batch input. A query is internally represented as a logical query plan and is translated in two phases:

  • Optimization of the logical plan
  • Translation into a DataStream or DataSet programA Table API or SQL query is translated when:
  • a Table is emitted to a TableSink, i.e., when Table.insertInto() is called.
  • a SQL update query is specified, i.e., when TableEnvironment.sqlUpdate() is called.
  • a Table is converted into a DataStream or DataSet (see Integration with DataStream and DataSet API).

Once translated, a Table API or SQL query is handled like a regular DataStream or DataSet program and is executed when StreamExecutionEnvironment.execute() or ExecutionEnvironment.execute() is called.

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:

  • Optimization of the logical plan,
  • Translation into a DataStream program.The behavior of translating a query is different for TableEnvironment and StreamTableEnvironment.

For TableEnvironment, a Table API or SQL query is translated when TableEnvironment.execute() is called, because TableEnvironment will optimize multiple-sinks into one DAG.

While for StreamTableEnvironment, a Table API or SQL query is translated when:

  • a Table is emitted to a TableSink, i.e., when Table.insertInto() is called.
  • a SQL update query is specified, i.e., when TableEnvironment.sqlUpdate() is called.
  • a Table is converted into a DataStream.

Once translated, a Table API or SQL query is handled like a regular DataStream program and is executed when TableEnvironment.execute() or StreamExecutionEnvironment.execute() is called.

Integration with DataStream and DataSet API

Both planners on stream can integrate with the DataStream API. Only old planner can integrate with the DataSet API, Blink planner on batch could not be combined with both.Note: The DataSet API discussed below is only relevant for the old planner on batch.

Table API and SQL queries can be easily integrated with and embedded into DataStream and DataSet programs. For instance, it is possible to query an external table (for example from a RDBMS), do some pre-processing, such as filtering, projecting, aggregating, or joining with meta data, and then further process the data with either the DataStream or DataSet API (and any of the libraries built on top of these APIs, such as CEP or Gelly). Inversely, a Table API or SQL query can also be applied on the result of a DataStream or DataSet program.

This interaction can be achieved by converting a DataStream or DataSet into a Table and vice versa. In this section, we describe how these conversions are done.

Implicit Conversion for Scala

The Scala Table API features implicit conversions for the DataSet, DataStream, and Table classes. These conversions are enabled by importing the package org.apache.flink.table.api.scala. in addition to org.apache.flink.api.scala. for the Scala DataStream API.

Create a View from a DataStream or DataSet

A DataStream or DataSet can be registered in a TableEnvironment as a View. The schema of the resulting view depends on the data type of the registered DataStream or DataSet. Please check the section about mapping of data types to table schema for details.

Note: Views created from a DataStream or DataSet can be registered as temporary views only.

  1. // get StreamTableEnvironment
  2. // registration of a DataSet in a BatchTableEnvironment is equivalent
  3. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  5. DataStream<Tuple2<Long, String>> stream = ...
  7. // register the DataStream as View "myTable" with fields "f0", "f1"
  8. tableEnv.createTemporaryView("myTable", stream);
  10. // register the DataStream as View "myTable2" with fields "myLong", "myString"
  11. tableEnv.createTemporaryView("myTable2", stream, "myLong, myString");
  1. // get TableEnvironment 
  2. // registration of a DataSet is equivalent
  3. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  5. val stream: DataStream[(Long, String)] = ...
  7. // register the DataStream as View "myTable" with fields "f0", "f1"
  8. tableEnv.createTemporaryView("myTable", stream)
  10. // register the DataStream as View "myTable2" with fields "myLong", "myString"
  11. tableEnv.createTemporaryView("myTable2", stream, 'myLong, 'myString)

Convert a DataStream or DataSet into a Table

Instead of registering a DataStream or DataSet in a TableEnvironment, it can also be directly converted into a Table. This is convenient if you want to use the Table in a Table API query.

  1. // get StreamTableEnvironment
  2. // registration of a DataSet in a BatchTableEnvironment is equivalent
  3. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  5. DataStream<Tuple2<Long, String>> stream = ...
  7. // Convert the DataStream into a Table with default fields "f0", "f1"
  8. Table table1 = tableEnv.fromDataStream(stream);
  10. // Convert the DataStream into a Table with fields "myLong", "myString"
  11. Table table2 = tableEnv.fromDataStream(stream, "myLong, myString");
  1. // get TableEnvironment
  2. // registration of a DataSet is equivalent
  3. val tableEnv = ... // see "Create a TableEnvironment" section
  5. val stream: DataStream[(Long, String)] = ...
  7. // convert the DataStream into a Table with default fields '_1, '_2
  8. val table1: Table = tableEnv.fromDataStream(stream)
  10. // convert the DataStream into a Table with fields 'myLong, 'myString
  11. val table2: Table = tableEnv.fromDataStream(stream, 'myLong, 'myString)

Convert a Table into a DataStream or DataSet

A Table can be converted into a DataStream or DataSet. In this way, custom DataStream or DataSet programs can be run on the result of a Table API or SQL query.

When converting a Table into a DataStream or DataSet, you need to specify the data type of the resulting DataStream or DataSet, i.e., the data type into which the rows of the Table are to be converted. Often the most convenient conversion type is Row. The following list gives an overview of the features of the different options:

  • Row: fields are mapped by position, arbitrary number of fields, support for null values, no type-safe access.
  • POJO: fields are mapped by name (POJO fields must be named as Table fields), arbitrary number of fields, support for null values, type-safe access.
  • Case Class: fields are mapped by position, no support for null values, type-safe access.
  • Tuple: fields are mapped by position, limitation to 22 (Scala) or 25 (Java) fields, no support for null values, type-safe access.
  • Atomic Type: Table must have a single field, no support for null values, type-safe access.

Convert a Table into a DataStream

A Table that is the result of a streaming query will be updated dynamically, i.e., it is changing as new records arrive on the query’s input streams. Hence, the DataStream into which such a dynamic query is converted needs to encode the updates of the table.

There are two modes to convert a Table into a DataStream:

  • Append Mode: This mode can only be used if the dynamic Table is only modified by INSERT changes, i.e, it is append-only and previously emitted results are never updated.
  • Retract Mode: This mode can always be used. It encodes INSERT and DELETE changes with a boolean flag.
  1. // get StreamTableEnvironment. 
  2. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // Table with two fields (String name, Integer age)
  5. Table table = ...
  7. // convert the Table into an append DataStream of Row by specifying the class
  8. DataStream<Row> dsRow = tableEnv.toAppendStream(table, Row.class);
  10. // convert the Table into an append DataStream of Tuple2<String, Integer> 
  11. //   via a TypeInformation
  12. TupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(
  13.   Types.STRING(),
  14.   Types.INT());
  15. DataStream<Tuple2<String, Integer>> dsTuple = 
  16.   tableEnv.toAppendStream(table, tupleType);
  18. // convert the Table into a retract DataStream of Row.
  19. //   A retract stream of type X is a DataStream<Tuple2<Boolean, X>>. 
  20. //   The boolean field indicates the type of the change. 
  21. //   True is INSERT, false is DELETE.
  22. DataStream<Tuple2<Boolean, Row>> retractStream = 
  23.   tableEnv.toRetractStream(table, Row.class);
  1. // get TableEnvironment. 
  2. // registration of a DataSet is equivalent
  3. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  5. // Table with two fields (String name, Integer age)
  6. val table: Table = ...
  8. // convert the Table into an append DataStream of Row
  9. val dsRow: DataStream[Row] = tableEnv.toAppendStream[Row](table)
  11. // convert the Table into an append DataStream of Tuple2[String, Int]
  12. val dsTuple: DataStream[(String, Int)] dsTuple = 
  13.   tableEnv.toAppendStream[(String, Int)](table)
  15. // convert the Table into a retract DataStream of Row.
  16. //   A retract stream of type X is a DataStream[(Boolean, X)]. 
  17. //   The boolean field indicates the type of the change. 
  18. //   True is INSERT, false is DELETE.
  19. val retractStream: DataStream[(Boolean, Row)] = tableEnv.toRetractStream[Row](table)

Note: A detailed discussion about dynamic tables and their properties is given in the Dynamic Tables document.

Convert a Table into a DataSet

A Table is converted into a DataSet as follows:

  1. // get BatchTableEnvironment
  2. BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);
  4. // Table with two fields (String name, Integer age)
  5. Table table = ...
  7. // convert the Table into a DataSet of Row by specifying a class
  8. DataSet<Row> dsRow = tableEnv.toDataSet(table, Row.class);
  10. // convert the Table into a DataSet of Tuple2<String, Integer> via a TypeInformation
  11. TupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(
  12.   Types.STRING(),
  13.   Types.INT());
  14. DataSet<Tuple2<String, Integer>> dsTuple = 
  15.   tableEnv.toDataSet(table, tupleType);
  1. // get TableEnvironment 
  2. // registration of a DataSet is equivalent
  3. val tableEnv = BatchTableEnvironment.create(env)
  5. // Table with two fields (String name, Integer age)
  6. val table: Table = ...
  8. // convert the Table into a DataSet of Row
  9. val dsRow: DataSet[Row] = tableEnv.toDataSet[Row](table)
  11. // convert the Table into a DataSet of Tuple2[String, Int]
  12. val dsTuple: DataSet[(String, Int)] = tableEnv.toDataSet[(String, Int)](table)

Mapping of Data Types to Table Schema

Flink’s DataStream and DataSet APIs support very diverse types. Composite types such as Tuples (built-in Scala and Flink Java tuples), POJOs, Scala case classes, and Flink’s Row type allow for nested data structures with multiple fields that can be accessed in table expressions. Other types are treated as atomic types. In the following, we describe how the Table API converts these types into an internal row representation and show examples of converting a DataStream into a Table.

The mapping of a data type to a table schema can happen in two ways: based on the field positions or based on the field names.

Position-based Mapping

Position-based mapping can be used to give fields a more meaningful name while keeping the field order. This mapping is available for composite data types with a defined field order as well as atomic types. Composite data types such as tuples, rows, and case classes have such a field order. However, fields of a POJO must be mapped based on the field names (see next section). Fields can be projected out but can’t be renamed using an alias as.

When defining a position-based mapping, the specified names must not exist in the input data type, otherwise the API will assume that the mapping should happen based on the field names. If no field names are specified, the default field names and field order of the composite type are used or f0 for atomic types.

  1. // get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
  2. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section;
  4. DataStream<Tuple2<Long, Integer>> stream = ...
  6. // convert DataStream into Table with default field names "f0" and "f1"
  7. Table table = tableEnv.fromDataStream(stream);
  9. // convert DataStream into Table with field "myLong" only
  10. Table table = tableEnv.fromDataStream(stream, "myLong");
  12. // convert DataStream into Table with field names "myLong" and "myInt"
  13. Table table = tableEnv.fromDataStream(stream, "myLong, myInt");
  1. // get a TableEnvironment
  2. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  4. val stream: DataStream[(Long, Int)] = ...
  6. // convert DataStream into Table with default field names "_1" and "_2"
  7. val table: Table = tableEnv.fromDataStream(stream)
  9. // convert DataStream into Table with field "myLong" only
  10. val table: Table = tableEnv.fromDataStream(stream, 'myLong)
  12. // convert DataStream into Table with field names "myLong" and "myInt"
  13. val table: Table = tableEnv.fromDataStream(stream, 'myLong, 'myInt)

Name-based Mapping

Name-based mapping can be used for any data type including POJOs. It is the most flexible way of defining a table schema mapping. All fields in the mapping are referenced by name and can be possibly renamed using an alias as. Fields can be reordered and projected out.

If no field names are specified, the default field names and field order of the composite type are used or f0 for atomic types.

  1. // get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
  2. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. DataStream<Tuple2<Long, Integer>> stream = ...
  6. // convert DataStream into Table with default field names "f0" and "f1"
  7. Table table = tableEnv.fromDataStream(stream);
  9. // convert DataStream into Table with field "f1" only
  10. Table table = tableEnv.fromDataStream(stream, "f1");
  12. // convert DataStream into Table with swapped fields
  13. Table table = tableEnv.fromDataStream(stream, "f1, f0");
  15. // convert DataStream into Table with swapped fields and field names "myInt" and "myLong"
  16. Table table = tableEnv.fromDataStream(stream, "f1 as myInt, f0 as myLong");
  1. // get a TableEnvironment
  2. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  4. val stream: DataStream[(Long, Int)] = ...
  6. // convert DataStream into Table with default field names "_1" and "_2"
  7. val table: Table = tableEnv.fromDataStream(stream)
  9. // convert DataStream into Table with field "_2" only
  10. val table: Table = tableEnv.fromDataStream(stream, '_2)
  12. // convert DataStream into Table with swapped fields
  13. val table: Table = tableEnv.fromDataStream(stream, '_2, '_1)
  15. // convert DataStream into Table with swapped fields and field names "myInt" and "myLong"
  16. val table: Table = tableEnv.fromDataStream(stream, '_2 as 'myInt, '_1 as 'myLong)

Atomic Types

Flink treats primitives (Integer, Double, String) or generic types (types that cannot be analyzed and decomposed) as atomic types. A DataStream or DataSet of an atomic type is converted into a Table with a single attribute. The type of the attribute is inferred from the atomic type and the name of the attribute can be specified.

  1. // get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
  2. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. DataStream<Long> stream = ...
  6. // convert DataStream into Table with default field name "f0"
  7. Table table = tableEnv.fromDataStream(stream);
  9. // convert DataStream into Table with field name "myLong"
  10. Table table = tableEnv.fromDataStream(stream, "myLong");
  1. // get a TableEnvironment
  2. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  4. val stream: DataStream[Long] = ...
  6. // convert DataStream into Table with default field name "f0"
  7. val table: Table = tableEnv.fromDataStream(stream)
  9. // convert DataStream into Table with field name "myLong"
  10. val table: Table = tableEnv.fromDataStream(stream, 'myLong)

Tuples (Scala and Java) and Case Classes (Scala only)

Flink supports Scala’s built-in tuples and provides its own tuple classes for Java. DataStreams and DataSets of both kinds of tuples can be converted into tables. Fields can be renamed by providing names for all fields (mapping based on position). If no field names are specified, the default field names are used. If the original field names (f0, f1, … for Flink Tuples and _1, _2, … for Scala Tuples) are referenced, the API assumes that the mapping is name-based instead of position-based. Name-based mapping allows for reordering fields and projection with alias (as).

  1. // get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
  2. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. DataStream<Tuple2<Long, String>> stream = ...
  6. // convert DataStream into Table with default field names "f0", "f1"
  7. Table table = tableEnv.fromDataStream(stream);
  9. // convert DataStream into Table with renamed field names "myLong", "myString" (position-based)
  10. Table table = tableEnv.fromDataStream(stream, "myLong, myString");
  12. // convert DataStream into Table with reordered fields "f1", "f0" (name-based)
  13. Table table = tableEnv.fromDataStream(stream, "f1, f0");
  15. // convert DataStream into Table with projected field "f1" (name-based)
  16. Table table = tableEnv.fromDataStream(stream, "f1");
  18. // convert DataStream into Table with reordered and aliased fields "myString", "myLong" (name-based)
  19. Table table = tableEnv.fromDataStream(stream, "f1 as 'myString', f0 as 'myLong'");
  1. // get a TableEnvironment
  2. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  4. val stream: DataStream[(Long, String)] = ...
  6. // convert DataStream into Table with renamed default field names '_1, '_2
  7. val table: Table = tableEnv.fromDataStream(stream)
  9. // convert DataStream into Table with field names "myLong", "myString" (position-based)
  10. val table: Table = tableEnv.fromDataStream(stream, 'myLong, 'myString)
  12. // convert DataStream into Table with reordered fields "_2", "_1" (name-based)
  13. val table: Table = tableEnv.fromDataStream(stream, '_2, '_1)
  15. // convert DataStream into Table with projected field "_2" (name-based)
  16. val table: Table = tableEnv.fromDataStream(stream, '_2)
  18. // convert DataStream into Table with reordered and aliased fields "myString", "myLong" (name-based)
  19. val table: Table = tableEnv.fromDataStream(stream, '_2 as 'myString, '_1 as 'myLong)
  21. // define case class
  22. case class Person(name: String, age: Int)
  23. val streamCC: DataStream[Person] = ...
  25. // convert DataStream into Table with default field names 'name, 'age
  26. val table = tableEnv.fromDataStream(streamCC)
  28. // convert DataStream into Table with field names 'myName, 'myAge (position-based)
  29. val table = tableEnv.fromDataStream(streamCC, 'myName, 'myAge)
  31. // convert DataStream into Table with reordered and aliased fields "myAge", "myName" (name-based)
  32. val table: Table = tableEnv.fromDataStream(stream, 'age as 'myAge, 'name as 'myName)

POJO (Java and Scala)

Flink supports POJOs as composite types. The rules for what determines a POJO are documented here.

When converting a POJO DataStream or DataSet into a Table without specifying field names, the names of the original POJO fields are used. The name mapping requires the original names and cannot be done by positions. Fields can be renamed using an alias (with the as keyword), reordered, and projected.

  1. // get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
  2. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // Person is a POJO with fields "name" and "age"
  5. DataStream<Person> stream = ...
  7. // convert DataStream into Table with default field names "age", "name" (fields are ordered by name!)
  8. Table table = tableEnv.fromDataStream(stream);
  10. // convert DataStream into Table with renamed fields "myAge", "myName" (name-based)
  11. Table table = tableEnv.fromDataStream(stream, "age as myAge, name as myName");
  13. // convert DataStream into Table with projected field "name" (name-based)
  14. Table table = tableEnv.fromDataStream(stream, "name");
  16. // convert DataStream into Table with projected and renamed field "myName" (name-based)
  17. Table table = tableEnv.fromDataStream(stream, "name as myName");
  1. // get a TableEnvironment
  2. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  4. // Person is a POJO with field names "name" and "age"
  5. val stream: DataStream[Person] = ...
  7. // convert DataStream into Table with default field names "age", "name" (fields are ordered by name!)
  8. val table: Table = tableEnv.fromDataStream(stream)
  10. // convert DataStream into Table with renamed fields "myAge", "myName" (name-based)
  11. val table: Table = tableEnv.fromDataStream(stream, 'age as 'myAge, 'name as 'myName)
  13. // convert DataStream into Table with projected field "name" (name-based)
  14. val table: Table = tableEnv.fromDataStream(stream, 'name)
  16. // convert DataStream into Table with projected and renamed field "myName" (name-based)
  17. val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName)

Row

The Row data type supports an arbitrary number of fields and fields with null values. Field names can be specified via a RowTypeInfo or when converting a Row DataStream or DataSet into a Table. The row type supports mapping of fields by position and by name. Fields can be renamed by providing names for all fields (mapping based on position) or selected individually for projection/ordering/renaming (mapping based on name).

  1. // get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
  2. StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
  4. // DataStream of Row with two fields "name" and "age" specified in `RowTypeInfo`
  5. DataStream<Row> stream = ...
  7. // convert DataStream into Table with default field names "name", "age"
  8. Table table = tableEnv.fromDataStream(stream);
  10. // convert DataStream into Table with renamed field names "myName", "myAge" (position-based)
  11. Table table = tableEnv.fromDataStream(stream, "myName, myAge");
  13. // convert DataStream into Table with renamed fields "myName", "myAge" (name-based)
  14. Table table = tableEnv.fromDataStream(stream, "name as myName, age as myAge");
  16. // convert DataStream into Table with projected field "name" (name-based)
  17. Table table = tableEnv.fromDataStream(stream, "name");
  19. // convert DataStream into Table with projected and renamed field "myName" (name-based)
  20. Table table = tableEnv.fromDataStream(stream, "name as myName");
  1. // get a TableEnvironment
  2. val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
  4. // DataStream of Row with two fields "name" and "age" specified in `RowTypeInfo`
  5. val stream: DataStream[Row] = ...
  7. // convert DataStream into Table with default field names "name", "age"
  8. val table: Table = tableEnv.fromDataStream(stream)
  10. // convert DataStream into Table with renamed field names "myName", "myAge" (position-based)
  11. val table: Table = tableEnv.fromDataStream(stream, 'myName, 'myAge)
  13. // convert DataStream into Table with renamed fields "myName", "myAge" (name-based)
  14. val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName, 'age as 'myAge)
  16. // convert DataStream into Table with projected field "name" (name-based)
  17. val table: Table = tableEnv.fromDataStream(stream, 'name)
  19. // convert DataStream into Table with projected and renamed field "myName" (name-based)
  20. val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName)

Query Optimization

Apache Flink leverages Apache Calcite to optimize and translate queries. The optimization currently performed include projection and filter push-down, subquery decorrelation, and other kinds of query rewriting. Old planner does not yet optimize the order of joins, but executes them in the same order as defined in the query (order of Tables in the FROM clause and/or order of join predicates in the WHERE clause).

It is possible to tweak the set of optimization rules which are applied in different phases by providing a CalciteConfig object. This can be created via a builder by calling CalciteConfig.createBuilder()) and is provided to the TableEnvironment by calling tableEnv.getConfig.setPlannerConfig(calciteConfig).

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 TableEnvironment.explain(table) method or TableEnvironment.explain() method. explain(table) returns the plan of a given Table. explain() returns the result of a multiple sinks plan and is mainly used for the Blink planner. It returns a String describing three plans:

  • the Abstract Syntax Tree of the relational query, i.e., the unoptimized logical query plan,
  • the optimized logical query plan, and
  • the physical execution plan.The following code shows an example and the corresponding output for given Table using explain(table):
  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. Table table1 = tEnv.fromDataStream(stream1, "count, word");
  8. Table table2 = tEnv.fromDataStream(stream2, "count, word");
  9. Table table = table1
  10.   .where("LIKE(word, 'F%')")
  11.   .unionAll(table2);
  13. String explanation = tEnv.explain(table);
  14. System.out.println(explanation);
  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. val explanation: String = tEnv.explain(table)
  11. println(explanation)
  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("LIKE(word, 'F%')") \
  8.     .union_all(table2)
  10. explanation = t_env.explain(table)
  11. print(explanation)
  1. == Abstract Syntax Tree ==
  2. LogicalUnion(all=[true])
  3.   LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  4.     FlinkLogicalDataStreamScan(id=[1], fields=[count, word])
  5.   FlinkLogicalDataStreamScan(id=[2], fields=[count, word])
  7. == Optimized Logical Plan ==
  8. DataStreamUnion(all=[true], union all=[count, word])
  9.   DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  10.     DataStreamScan(id=[1], fields=[count, word])
  11.   DataStreamScan(id=[2], fields=[count, word])
  13. == Physical Execution Plan ==
  14. Stage 1 : Data Source
  15.     content : collect elements with CollectionInputFormat
  17. Stage 2 : Data Source
  18.     content : collect elements with CollectionInputFormat
  20.     Stage 3 : Operator
  21.         content : from: (count, word)
  22.         ship_strategy : REBALANCE
  24.         Stage 4 : Operator
  25.             content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
  26.             ship_strategy : FORWARD
  28.             Stage 5 : Operator
  29.                 content : from: (count, word)
  30.                 ship_strategy : REBALANCE
  1. == Abstract Syntax Tree ==
  2. LogicalUnion(all=[true])
  3.   LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  4.     FlinkLogicalDataStreamScan(id=[1], fields=[count, word])
  5.   FlinkLogicalDataStreamScan(id=[2], fields=[count, word])
  7. == Optimized Logical Plan ==
  8. DataStreamUnion(all=[true], union all=[count, word])
  9.   DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  10.     DataStreamScan(id=[1], fields=[count, word])
  11.   DataStreamScan(id=[2], fields=[count, word])
  13. == Physical Execution Plan ==
  14. Stage 1 : Data Source
  15.     content : collect elements with CollectionInputFormat
  17. Stage 2 : Data Source
  18.     content : collect elements with CollectionInputFormat
  20.     Stage 3 : Operator
  21.         content : from: (count, word)
  22.         ship_strategy : REBALANCE
  24.         Stage 4 : Operator
  25.             content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
  26.             ship_strategy : FORWARD
  28.             Stage 5 : Operator
  29.                 content : from: (count, word)
  30.                 ship_strategy : REBALANCE
  1. == Abstract Syntax Tree ==
  2. LogicalUnion(all=[true])
  3.   LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  4.     FlinkLogicalDataStreamScan(id=[3], fields=[count, word])
  5.   FlinkLogicalDataStreamScan(id=[6], fields=[count, word])
  7. == Optimized Logical Plan ==
  8. DataStreamUnion(all=[true], union all=[count, word])
  9.   DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  10.     DataStreamScan(id=[3], fields=[count, word])
  11.   DataStreamScan(id=[6], fields=[count, word])
  13. == Physical Execution Plan ==
  14. Stage 1 : Data Source
  15.     content : collect elements with CollectionInputFormat
  17.     Stage 2 : Operator
  18.         content : Flat Map
  19.         ship_strategy : FORWARD
  21.         Stage 3 : Operator
  22.             content : Map
  23.             ship_strategy : FORWARD
  25. Stage 4 : Data Source
  26.     content : collect elements with CollectionInputFormat
  28.     Stage 5 : Operator
  29.         content : Flat Map
  30.         ship_strategy : FORWARD
  32.         Stage 6 : Operator
  33.             content : Map
  34.             ship_strategy : FORWARD
  36.             Stage 7 : Operator
  37.                 content : Map
  38.                 ship_strategy : FORWARD
  40.                 Stage 8 : Operator
  41.                     content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
  42.                     ship_strategy : FORWARD
  44.                     Stage 9 : Operator
  45.                         content : Map
  46.                         ship_strategy : FORWARD

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

  1. EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  2. TableEnvironment tEnv = TableEnvironment.create(settings);
  4. final Schema schema = new Schema()
  5.     .field("count", DataTypes.INT())
  6.     .field("word", DataTypes.STRING());
  8. tEnv.connect(new FileSystem("/source/path1"))
  9.     .withFormat(new Csv().deriveSchema())
  10.     .withSchema(schema)
  11.     .createTemporaryTable("MySource1");
  12. tEnv.connect(new FileSystem("/source/path2"))
  13.     .withFormat(new Csv().deriveSchema())
  14.     .withSchema(schema)
  15.     .createTemporaryTable("MySource2");
  16. tEnv.connect(new FileSystem("/sink/path1"))
  17.     .withFormat(new Csv().deriveSchema())
  18.     .withSchema(schema)
  19.     .createTemporaryTable("MySink1");
  20. tEnv.connect(new FileSystem("/sink/path2"))
  21.     .withFormat(new Csv().deriveSchema())
  22.     .withSchema(schema)
  23.     .createTemporaryTable("MySink2");
  25. Table table1 = tEnv.from("MySource1").where("LIKE(word, 'F%')");
  26. table1.insertInto("MySink1");
  28. Table table2 = table1.unionAll(tEnv.from("MySource2"));
  29. table2.insertInto("MySink2");
  31. String explanation = tEnv.explain(false);
  32. System.out.println(explanation);
  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("/source/path1"))
  9.     .withFormat(new Csv().deriveSchema())
  10.     .withSchema(schema)
  11.     .createTemporaryTable("MySource1")
  12. tEnv.connect(new FileSystem("/source/path2"))
  13.     .withFormat(new Csv().deriveSchema())
  14.     .withSchema(schema)
  15.     .createTemporaryTable("MySource2")
  16. tEnv.connect(new FileSystem("/sink/path1"))
  17.     .withFormat(new Csv().deriveSchema())
  18.     .withSchema(schema)
  19.     .createTemporaryTable("MySink1")
  20. tEnv.connect(new FileSystem("/sink/path2"))
  21.     .withFormat(new Csv().deriveSchema())
  22.     .withSchema(schema)
  23.     .createTemporaryTable("MySink2")
  25. val table1 = tEnv.from("MySource1").where("LIKE(word, 'F%')")
  26. table1.insertInto("MySink1")
  28. val table2 = table1.unionAll(tEnv.from("MySource2"))
  29. table2.insertInto("MySink2")
  31. val explanation = tEnv.explain(false)
  32. println(explanation)
  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. table1 = t_env.from_path("MySource1").where("LIKE(word, 'F%')")
  26. table1.insert_into("MySink1")
  28. table2 = table1.union_all(t_env.from_path("MySource2"))
  29. table2.insert_into("MySink2")
  31. explanation = t_env.explain()
  32. print(explanation)

the result of multiple-sinks plan is

  1. == Abstract Syntax Tree ==
  2. LogicalSink(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. LogicalSink(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. Sink(name=[MySink1], fields=[count, word])
  17. +- Reused(reference_id=[1])
  19. Sink(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