class 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());
    }

    public  SqlJob(SqlConfig sqlConfig);

}