Apache Flink - API de table et SQL

L'API de table est une API relationnelle avec un langage d'expression similaire à SQL. Cette API peut effectuer à la fois le traitement par lots et par flux. Il peut être intégré avec les API Java et Scala Dataset et Datastream. Vous pouvez créer des tables à partir d'ensembles de données et de flux de données existants ou à partir de sources de données externes. Grâce à cette API relationnelle, vous pouvez effectuer des opérations telles que la jointure, l'agrégation, la sélection et le filtrage. Que l'entrée soit par lot ou par flux, la sémantique de la requête reste la même.

Voici un exemple de programme d'API Table -

// for batch programs use ExecutionEnvironment instead of StreamExecutionEnvironment
val env = StreamExecutionEnvironment.getExecutionEnvironment

// create a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// register a Table
tableEnv.registerTable("table1", ...) // or
tableEnv.registerTableSource("table2", ...) // or
tableEnv.registerExternalCatalog("extCat", ...)

// register an output Table
tableEnv.registerTableSink("outputTable", ...);
// create a Table from a Table API query
val tapiResult = tableEnv.scan("table1").select(...)
// Create a Table from a SQL query
val sqlResult = tableEnv.sqlQuery("SELECT ... FROM table2 ...")

// emit a Table API result Table to a TableSink, same for SQL result
tapiResult.insertInto("outputTable")

// execute
env.execute()