查询 #

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);  // ingest a DataStream from an external source DataStream<Tuple3<Long, String, Integer>> ds = env.addSource(...);  // SQL query with an inlined (unregistered) table Table table = tableEnv.fromDataStream(ds, $("user"), $("product"), $("amount")); Table result = tableEnv.sqlQuery(  "SELECT SUM(amount) FROM " + table + " WHERE product LIKE '%Rubber%'");  // SQL query with a registered table // register the DataStream as view "Orders" tableEnv.createTemporaryView("Orders", ds, $("user"), $("product"), $("amount")); // run a SQL query on the Table and retrieve the result as a new Table Table result2 = tableEnv.sqlQuery(  "SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");  // create and register a TableSink final Schema schema = Schema.newBuilder()  .column("product", DataTypes.STRING())  .column("amount", DataTypes.INT())  .build();  final TableDescriptor sinkDescriptor = TableDescriptor.forConnector("filesystem")  .schema(schema)  .option("path", "/path/to/file")  .format(FormatDescriptor.forFormat("csv")  .option("field-delimiter", ",")  .build())  .build();  tableEnv.createTemporaryTable("RubberOrders", sinkDescriptor);  // run an INSERT SQL on the Table and emit the result to the TableSink tableEnv.executeSql(  "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'"); 

val env = StreamExecutionEnvironment.getExecutionEnvironment val tableEnv = StreamTableEnvironment.create(env)  // read a DataStream from an external source val ds: DataStream[(Long, String, Integer)] = env.addSource(...)  // SQL query with an inlined (unregistered) table val table = ds.toTable(tableEnv, $"user", $"product", $"amount") val result = tableEnv.sqlQuery(  s"SELECT SUM(amount) FROM $table WHERE product LIKE '%Rubber%'")  // SQL query with a registered table // register the DataStream under the name "Orders" tableEnv.createTemporaryView("Orders", ds, $"user", $"product", $"amount") // run a SQL query on the Table and retrieve the result as a new Table val result2 = tableEnv.sqlQuery(  "SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")  // create and register a TableSink val schema = Schema.newBuilder()  .column("product", DataTypes.STRING())  .column("amount", DataTypes.INT())  .build()  val sinkDescriptor = TableDescriptor.forConnector("filesystem")  .schema(schema)  .format(FormatDescriptor.forFormat("csv")  .option("field-delimiter", ",")  .build())  .build()  tableEnv.createTemporaryTable("RubberOrders", sinkDescriptor)  // run an INSERT SQL on the Table and emit the result to the TableSink tableEnv.executeSql(  "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'") 

env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env)  # SQL query with an inlined (unregistered) table # elements data type: BIGINT, STRING, BIGINT table = table_env.from_elements(..., ['user', 'product', 'amount']) result = table_env \  .sql_query("SELECT SUM(amount) FROM %s WHERE product LIKE '%%Rubber%%'" % table)  # create and register a TableSink schema = Schema.new_builder()  .column("product", DataTypes.STRING())  .column("amount", DataTypes.INT())  .build()  sink_descriptor = TableDescriptor.for_connector("filesystem")  .schema(schema)  .format(FormatDescriptor.for_format("csv")  .option("field-delimiter", ",")  .build())  .build()  t_env.create_temporary_table("RubberOrders", sink_descriptor)  # run an INSERT SQL on the Table and emit the result to the TableSink table_env \  .execute_sql("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'") 

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);  tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");  // execute SELECT statement TableResult tableResult1 = tableEnv.executeSql("SELECT * FROM Orders"); // use try-with-resources statement to make sure the iterator will be closed automatically try (CloseableIterator<Row> it = tableResult1.collect()) {  while(it.hasNext()) {  Row row = it.next();  // handle row  } }  // execute Table TableResult tableResult2 = tableEnv.sqlQuery("SELECT * FROM Orders").execute(); tableResult2.print(); 

val env = StreamExecutionEnvironment.getExecutionEnvironment() val tableEnv = StreamTableEnvironment.create(env, settings) // enable checkpointing tableEnv.getConfig  .set(CheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE) tableEnv.getConfig  .set(CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10))  tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")  // execute SELECT statement val tableResult1 = tableEnv.executeSql("SELECT * FROM Orders") val it = tableResult1.collect() try while (it.hasNext) {  val row = it.next  // handle row } finally it.close() // close the iterator to avoid resource leak  // execute Table val tableResult2 = tableEnv.sqlQuery("SELECT * FROM Orders").execute() tableResult2.print() 

env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env, settings) # enable checkpointing table_env.get_config().set("execution.checkpointing.mode", "EXACTLY_ONCE") table_env.get_config().set("execution.checkpointing.interval", "10s")  table_env.execute_sql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")  # execute SELECT statement table_result1 = table_env.execute_sql("SELECT * FROM Orders") table_result1.print()  # execute Table table_result2 = table_env.sql_query("SELECT * FROM Orders").execute() table_result2.print()