总览： import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
object sqlQueryTable {
def main(args: Array[String]): Unit = {
//create env
val env = StreamExecutionEnvironment.getExecutionEnvironment
//set parallelism
env.setParallelism(1)
//env setting
val envSettings = EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build()
//create table env
val tableEnv = StreamTableEnvironment.create(env, envSettings)
//create flink table
val flink_table_sql =
"""
|create table student2_flink_table
|(
| code varchar(20) null,
| name varchar(20) null,
| score int null
|)with(
|'connector.type'='jdbc',
|'connector.url'='jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8&useSSL=true&serverTimezone=UTC',
|'connector.table'='student2',
|'connector.driver'='com.mysql.jdbc.Driver',
|'connector.username'='root',
|'connector.password'='123456'
|)
|""".stripMargin
//execute create flink table sql
tableEnv.executeSql(flink_table_sql)
//register table
val myStudent_Flink_Table = tableEnv.from("student2_flink_table")
//query table
val result = tableEnv.sqlQuery(s"select * from $myStudent_Flink_Table where code < 10")
//print
//result.toRetractStream[(String,String)].print()
result.toAppendStream[(String,String,Int)].print()
//create view
tableEnv.createTemporaryView("student2_flink_view", myStudent_Flink_Table)
//query use sql
val querySQL =
"""
|select code,
|name,
|sum(score) as total_score
|from student2_flink_view
|group by code, name
|""".stripMargin
//query
val total_score_resut = tableEnv.sqlQuery(querySQL)
//print
total_score_resut.toRetractStream[(String,String,Int)].print("view-")
//execute
env.execute()
}
}
toAppendStream 和 toRetractStream区别
toAppendStream 的输出结果
toRetractStream的输出结果
感觉 toRetractStream支持更新文章来源：https://www.toymoban.com/news/detail-422033.html
从输出得结果看，每条结果前都会有true，当接收到新得数据时会更新原先得数据，并在原先得数据前面标记false，也就是失效或者作废得意思，从而得到新得数据，到此应该也能很清晰得区分 toAppendStream与toRetractStream的区别了文章来源地址https://www.toymoban.com/news/detail-422033.html
到了这里，关于flink学习35：flinkSQL查询mysql的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
