Flink - toAppendStream与toRetract

前言

通常我们在需要输出Table表数据时需要转换成DataStream流进行输出，然后转换流有两种模式toAppendStream追加模式、toRetractStream更新模式

toAppendStream：追加模式

• 接收端口数据，测试追加模式

代码示例

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.api.scala._

//定义样例类WaterSensor
case class WaterSensor(id:String,ts:Long,vc:Double)
object TableOutCsv {
  def main(args: Array[String]): Unit = {
    //创建流执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //创建表执行环境
    val table: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env,table)

   //接收指定端口得数据，并转换成样例类WaterSensor类型的DataStream
    val dataStream: DataStream[WaterSensor] = env.socketTextStream("192.168.95.99",7777)
      .map(a=>{
        val strings: Array[String] = a.split(",")
        WaterSensor(strings(0),strings(1).toLong,strings(2).toDouble)
      })

    //根据流创建一张Table类型得得对象
    val dataTable: Table = tableEnv.fromDataStream(dataStream)

    //调用Table API进行转换
    val dataTable2: Table = dataTable.select("id,vc").filter('id === "ws_003")

    //使用追加模式，当有数据更新时，直接在后面跟着输出
    dataTable2.toAppendStream[(String,Double)].print("append")

    //启动执行
    env.execute()
  }
}

启动端口

启动程序

测试数据

ws_001,1577844001,24.0
ws_002,1577844015,43.0
ws_003,1577844020,32.0

端口输入

程序输出

追加数据

ws_003,1577844020,23.0
ws_003,1577844020,65.0

程序输出

结论：使用toAppendStream就是当接收到新得数据时候不会影响之前得数据，而是在后面追加

toRetractStream：更新模式

• 依然是接收端口数据，只不过这次我们使用Table API对数据进行count，查看使用toRetractStream得效果

代码示例

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.api.scala._

//定义样例类WaterSensor
case class WaterSensor(id:String,ts:Long,vc:Double)
object TableOutCsv {
  def main(args: Array[String]): Unit = {
    //创建流执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //创建表执行环境
    val table: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env,table)

    //接收指定端口得数据，并转换成样例类WaterSensor类型的DataStream
    val dataStream: DataStream[WaterSensor] = env.socketTextStream("192.168.95.99",7777)
      .map(a=>{
        val strings: Array[String] = a.split(",")
        WaterSensor(strings(0),strings(1).toLong,strings(2).toDouble)
      })

    //根据流创建一张Table类型得得对象
    val dataTable: Table = tableEnv.fromDataStream(dataStream)

    //调用Table API进行转换
    val dataTable2: Table = dataTable
      .groupBy('id) //根据ID进行分组
      .select('id,'vc.count as 'countVC) //count相同ID得vc值

    //使用追加模式，当有数据更新时，直接在后面跟着输出
    dataTable2.toRetractStream[(String,Double)].print("retract")

    //启动执行
    env.execute()
  }
}

启动端口

启动程序

测试数据

ws_001,1577844001,24.0
ws_002,1577844015,43.0
ws_003,1577844020,32.0

端口输入

程序输出

追加数据

ws_003,1577844020,23.0
ws_003,1577844020,65.0

程序输出

结论：从输出得结果看，每条结果前都会有true，当接收到新得数据时会更新原先得数据，并在原先得数据前面标记false，也就是失效或者作废得意思，从而得到新得数据，到此应该也能很清晰得区分 toAppendStream与toRetractStream的区别了把