SqlJob { public JobExecutionResult start() throws Exception { log.info("sql job {} staring...", sqlConfig.getJobName()); StreamExecutionEnvironment env = new LocalStreamEnvironment(); EnvironmentSettings settings = EnvironmentSettings.newInstance() .useBlinkPlanner().inStreamingMode().build(); TableEnvironmentImpl tEnv = (TableEnvironmentImpl) StreamTableEnvironment.create(env, settings); SqlParser sqlParser = SqlParser.create(sqlConfig.getSql(), PlinkSqlParser.sqlParserConfig); SqlNodeList sqlNodes = sqlParser.parseStmtList(); sqlNodes.forEach(sqlNode -> { if (sqlNode instanceof SqlSetOption) { String name = ((SqlSetOption) sqlNode).getName().getSimple(); String value = ((SqlSetOption) sqlNode).getValue().toString(); tEnv.getConfig().getConfiguration().setString(name, value); } else { String sql = sqlNode.toSqlString(SkipAnsiCheckSqlDialect.DEFAULT).getSql(); tEnv.sqlUpdate(sql); } }); return tEnv.execute(sqlConfig.getJobName()); } SqlJob(SqlConfig sqlConfig); JobExecutionResult start();  }
@Test public void sqlJobWatermarkTest() { List<JSONObject> sourceData = Stream.of( new JSONObject().fluentPut("data_time", "2020-01-01 12:00:01"), new JSONObject().fluentPut("data_time", "2020-01-01 12:00:02"), new JSONObject().fluentPut("data_time", "2020-01-01 12:00:03"), new JSONObject().fluentPut("data_time", "2020-01-01 12:01:01") ).collect(Collectors.toList()); String sql = "create table t1( " + "data_time STRING, " + "row1_time AS to_timestamp(data_time)," + "WATERMARK FOR row1_time AS row1_time - INTERVAL '5' SECOND " + ") with ( 'connector' = 'collection','data'='" + JSON.toJSONString(sourceData) + "');" + "create table t2(stime TIMESTAMP(3),cnt bigint) with ( 'connector' = 'print');" + "insert into t2 select TUMBLE_START(row1_time, INTERVAL '1' MINUTE) as stime,count(1) cnt from t1 group by TUMBLE(row1_time, INTERVAL '1' MINUTE);;"; Exception exception = null; try { SqlConfig config = SqlConfig.builder().sql(sql).jobName("sql_job_test").build(); SqlJob sqlJob = new SqlJob(config); sqlJob.start(); } catch (Exception e) { log.error("sqlJobTest error", e); exception = e; } assert exception == null; }