当前位置：网站首页>（5） Flink's table API and SQL update mode and Kafka connector case

（5） Flink's table API and SQL update mode and Kafka connector case

2022-07-02 15:41:00 【wx5ba7ab4695f27】

List of articles

Update mode
Application example

Update mode

about stream Type of Table data , Marking is required because INSERT,UPDATE,DELETE Which operation updates the data , stay Table API Pass through Update Modes Specify the data update type , By specifying different Update Modes Determine which update operation data interacts with external systems

       
        .connect(...)
        
.inAppendMode()   // Interaction INSERT Operation update data 
        
.inUpsertMode()   //INSERT,UPDATE,DELETE Operation update data 
        
.inRetractMode()  // Interaction INSERT and DELETE Operation update data 
       
       
        1.
        2.
        3.
        4.

Application example

       
        import org.apache.flink.api.scala.typeutils.Types
        
import org.apache.flink.streaming.api.scala._
        
import org.apache.flink.table.api.scala._
        
import org.apache.flink.table.descriptors.{Json, Kafka, Rowtime, Schema}
        

        
object ConnectorDemo {
        
  def main(args: Array[String]): Unit = {
        
    val env = StreamExecutionEnvironment.getExecutionEnvironment
        
    val tabEnv = StreamTableEnvironment.create(env)
        
    tabEnv.connect(
        
      new Kafka()
        
        .version("0.10")
        
        .topic("test")
        
        .property("zookeeper.connect","note01:2181,note02:2181,note03:2181")
        
        .property("bootstrap.servers","note01:9092,note02:9092,note03:9092")
        
    )
        
    // Appoint Table Format Information 
        
      .withFormat(
        
        new Json()
        
          .failOnMissingField(true)
        
          .jsonSchema(
        
            """
        
              |{
        
              |type:'object',
        
              |properties:{
        
              |id:{
        
              |type:'number'
        
              |},
        
              |name:{
        
              |type:'string'
        
              |},
        
              |timestamp:{
        
              |type:'string'      
        
              |format:'date-time'
        
              |}
        
              |}
        
              |}
        
              |""".stripMargin)
        
      ).withSchema(
        
      new Schema()
        
        .field("id",Types.INT)
        
        .field("name",Types.STRING)
        
        .field("rowtime",Types.SQL_TIMESTAMP)
        
        .rowtime(
        
          new Rowtime()
        
            .timestampsFromField("timestamp")
        
            .watermarksPeriodicBounded(60000) //1 minutes
        
        )
        
    )
        
    // Specify data update mode 
        
      .inAppendMode()
        
      .registerTableSource("kafkaTable")
        
  }
        
}
       
       
        1.
        2.
        3.
        4.
        5.
        6.
        7.
        8.
        9.
        10.
        11.
        12.
        13.
        14.
        15.
        16.
        17.
        18.
        19.
        20.
        21.
        22.
        23.
        24.
        25.
        26.
        27.
        28.
        29.
        30.
        31.
        32.
        33.
        34.
        35.
        36.
        37.
        38.
        39.
        40.
        41.
        42.
        43.
        44.
        45.
        46.
        47.
        48.
        49.
        50.
        51.
        52.
        53.
        54.