EnvironmentSettings settings = EnvironmentSettings.inStreamingMode(); TableEnvironment tEnv = TableEnvironment.create(settings);  tEnv.executeSql("CREATE TABLE transactions (\n" +     "    account_id  BIGINT,\n" +     "    amount      BIGINT,\n" +     "    transaction_time TIMESTAMP(3),\n" +     "    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '5' SECOND\n" +     ") WITH (\n" +     "    'connector' = 'kafka',\n" +     "    'topic'     = 'transactions',\n" +     "    'properties.bootstrap.servers' = 'kafka:9092',\n" +     "    'format'    = 'csv'\n" +     ")");  tEnv.executeSql("CREATE TABLE spend_report (\n" +     "    account_id BIGINT,\n" +     "    log_ts     TIMESTAMP(3),\n" +     "    amount     BIGINT\n," +     "    PRIMARY KEY (account_id, log_ts) NOT ENFORCED" +     ") WITH (\n" +     "   'connector'  = 'jdbc',\n" +     "   'url'        = 'jdbc:mysql://mysql:3306/sql-demo',\n" +     "   'table-name' = 'spend_report',\n" +     "   'driver'     = 'com.mysql.jdbc.Driver',\n" +     "   'username'   = 'sql-demo',\n" +     "   'password'   = 'demo-sql'\n" +     ")");  Table transactions = tEnv.from("transactions"); report(transactions).executeInsert("spend_report");

Breaking Down The Code  #

The Execution Environment  #

The first two lines set up your  TableEnvironment. The table environment is how you can set properties for your Job, specify whether you are writing a batch or a streaming application, and create your sources. This walkthrough creates a standard table environment that uses the streaming execution.

EnvironmentSettings settings = EnvironmentSettings.inStreamingMode(); TableEnvironment tEnv = TableEnvironment.create(settings);

Registering Tables  #

Next, tables are registered in the current  catalog that you can use to