|
主措施包罗执行情形的界说，Source/Sink的注册以及统计查SQL的执行，详细如下：
- def main(args: Array[String]): Unit = {
- // Streaming 情形
- val env = StreamExecutionEnvironment.getExecutionEnvironment
- val tEnv = TableEnvironment.getTableEnvironment(env)
-
- // 配置EventTime
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
-
- //利便我们查出输出数据
- env.setParallelism(1)
-
- val sourceTableName = "mySource"
- // 建设自界说source数据布局
- val tableSource = new MyTableSource
-
- val sinkTableName = "csvSink"
- // 建设CSV sink 数据布局
- val tableSink = getCsvTableSink
-
- // 注册source
- tEnv.registerTableSource(sourceTableName, tableSource)
- // 注册sink
- tEnv.registerTableSink(sinkTableName, tableSink)
-
- val sql =
- "SELECT " +
- " region, " +
- " TUMBLE_START(accessTime, INTERVAL '2' MINUTE) AS winStart," +
- " TUMBLE_END(accessTime, INTERVAL '2' MINUTE) AS winEnd, COUNT(region) AS pv " +
- " FROM mySource " +
- " GROUP BY TUMBLE(accessTime, INTERVAL '2' MINUTE), region"
-
- tEnv.sqlQuery(sql).insertInto(sinkTableName);
- env.execute()
- }
4. 执行并查察运行功效 （编辑：湖南网）
【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!
|
