Apache Flink 漫谈系列 - SQL概览

我们简单的将计算结果写入到Apache Flink内置支持的CSVSink中，定义Sink如下：

def getCsvTableSink: TableSink[Row] = { 
val tempFile = File.createTempFile("csv_sink_", "tem") 
// 打印sink的文件路径，方便我们查看运行结果 
println("Sink path : " + tempFile) 
if (tempFile.exists()) { 
tempFile.delete() 
} 
new CsvTableSink(tempFile.getAbsolutePath).configure( 
Array[String]("region", "winStart", "winEnd", "pv"), 
Array[TypeInformation[_]](Types.STRING, Types.SQL_TIMESTAMP, Types.SQL_TIMESTAMP, Types.LONG))} 

3. 构建主程序

（编辑：常州站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!