Flink 的 Scala API 怎么指定 时间字段

需要的pom依赖：

<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>${flink.version}</version>

<groupId>org.apache.flink</groupId>
<artifactId>flink-table_2.11</artifactId>
<version>${flink.version}</version>

<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>${flink.version}</version>

<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.8_2.11</artifactId>
<version>${flink.version}</version>

从kafka消费数据，转换为table，然后进行sql查询。
用scala开发，需要导入的包，不要漏掉，否则会有问题。

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.api.scala._
1
2
3
4
5
6
下面是完整代码：
package com.ddxygq.bigdata.flink.sql

import java.util.Properties

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.api.scala._

/**

• @ Author: keguang
• @ Date: 2019/2/22 16:13
• @ version: v1.0.0
• @ description:
  */

object TableDemo {
def main(args: Array[String]): Unit = {

demo

}

def demo2(): Unit ={

val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env)

val input:DataSet[WC] = env.fromElements(WC("hello", 1), WC("hello", 1), WC("ciao", 1))
val input2:DataSet[WC] = env.fromElements(WC("hello", 1), WC("hello", 1))
val table = input.toTable(tEnv, 'word, 'frequency)
val table2 = input2.toTable(tEnv, 'word2, 'frequency2)
val result = table.join(table2).where('word == 'word2).select('word, 'frequency)
result.toDataSet[(String, Long)].print()

}

def demo: Unit ={

val sEnv = StreamExecutionEnvironment.getExecutionEnvironment
val sTableEnv = TableEnvironment.getTableEnvironment(sEnv)

// 连接kafka
val ZOOKEEPER_HOST = "qcloud-test-hadoop01:2181"
val KAFKA_BROKERS = "qcloud-test-hadoop01:9092,qcloud-test-hadoop02:9092,qcloud-test-hadoop03:9092"
val TRANSACTION_GROUP = "transaction"
val kafkaProps = new Properties()
kafkaProps.setProperty("zookeeper.connect",ZOOKEEPER_HOST)
kafkaProps.setProperty("bootstrap.servers", KAFKA_BROKERS)
kafkaProps.setProperty("group.id",TRANSACTION_GROUP)
val input = sEnv.addSource(
  new FlinkKafkaConsumer08[String]("flink-test", new SimpleStringSchema(), kafkaProps)
)
  .flatMap(x => x.split(" "))
  .map(x => (x, 1L))

val table = sTableEnv.registerDataStream("Words", input, 'word, 'frequency)

val result = sTableEnv
  .scan("Words")
  .groupBy("word")
  .select('word, 'frequency.sum as 'cnt)
sTableEnv.toRetractStream[(String, Long)](result).print()

sTableEnv.sqlQuery("select * from Words").toAppendStream[(String, Long)].print()

sEnv.execute("TableDemo")

}
}

这里有两个地方：
1、这里举例用了table的算子，和标准的sql查询语法，为了演示table的基本用法。

val result = sTableEnv

  .scan("Words")
  .groupBy("word")
  .select('word, 'frequency.sum as 'cnt)

这个分组聚合统计其实可以替换成：

val result = sTableEnv.sqlQuery("select word,sum(frequency) as cnt from Words group by word")

// 打印到控制台
sTableEnv.toRetractStream(String, Long).print()
那么这个与下面的查询结果有什么区别呢？

sTableEnv.sqlQuery("select * from Words").toAppendStream[(String, Long)].print()

区别很明显，这里消费kafka的实时数据，那么Words表是一个动态的流表，数据在不断append，一个是group by的分组聚合，结果需要不断更新，比如当前是(hello,4)，这时候又来了一个词语hello，就需要update结果为(hello,5)，如果有新词，还需要insert，而后者是select * from Words，只是追加结果。

所以，这里只是展示打印到控制台的写法不同，前者调用的是toRetractStream方法，而后者是调用toAppendStream。