Flink学习5-使用rowtime且分窗，Connector读取Kafka写入MySQL例子

喜欢ヅ旅行 2023-06-10 04:29 24阅读 0赞

# Flink学习5-使用rowtime且分窗，Connector读取Kafka写入MySQL例子 #

# Flink系列文章 #

*  更多Flink系列文章请点击[Flink系列文章][Flink]
 *  更多大数据文章请点击[大数据好文推荐][Link 1]

## 摘要 ##

本文本来应该写在[Flink学习3-API介绍][Flink_3-API]的Connector章节，但因为笔者在做这个功能时翻遍全网都没找到例子，怕各位搜不到，所以在这里写了。

熟悉我的朋友都知道，我喜欢把知识点都写到一篇文章，原谅我凑了一篇文章 - -。

注意，本文基于Flink 1.9.0。

## 例子 ##

我们的场景就是从Kafka读出无界流数据，进行SQL查询，随后写入MySQL中。这里我们采用的是事件时间event\_time，即rowtime。

难点主要是：

*  官网例子太简略，不知道到底咋写？
 *  Rowtime定义很麻烦，必须是原始数据就是timestamp，否则没法转？
 *  Flink类型如何转换?
 *  嵌套类型如何处理?
 *  如何用sql分窗？
 *  如何解决Flink时间是utc时间，无法修改，与中国时间差8小时问题？

来吧，我们直接上例子，上述问题答案都在下面。这里用scala。

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.scala.StreamTableEnvironment
    import org.apache.flink.table.api.{ EnvironmentSettings}
    import org.apache.flink.table.descriptors.{ Json, Kafka, Rowtime, Schema}
    import org.apache.flink.api.common.typeinfo.Types
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.table.sources.tsextractors.{ ExistingField}
    
    object readkafka { 
      def main(args: Array[String]):Unit ={ 
        // **********************
        // BLINK STREAMING QUERY
        // **********************
        // get streaming runtime
        val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
        // set event time as stream time
        bsEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        // get blink streaming settings
        val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
        // get streaming table runtime
        val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
    
        bsTableEnv
          .connect(new Kafka()
            .version("0.10")
            .topic("student")
            .property("bootstrap.servers","192.168.1.1:9092")
            .property("group.id","test_20191025")
            .property("zookeeper.connect","192.168.1.1:2222")
            .startFromEarliest())
          .withFormat(new Json()
            .failOnMissingField(false)
              .schema(Types.ROW_NAMED(
                Array("stu_id", "name", "time", "register_time", "detail", "mf"),
                Types.STRING,
                Types.STRING,
                Types.LONG,
                Types.LONG,
                Types.ROW_NAMED(
                  Array("age", "grade"),
                  Types.STRING,
                  Types.STRING),
                Types.STRING)))
          .withSchema(new Schema()
            .field("stu_id", Types.STRING)
            .field("name", Types.STRING)
            .field("rowtime", Types.SQL_TIMESTAMP)
              .rowtime(new Rowtime()
                .timestampsFromExtractor(new ExistingField("time"))
                .watermarksPeriodicBounded(5000))
            .field("register_time", Types.LONG)
            .field("detail", Types.ROW_NAMED(Array("age","grade"), Types.INT, Types.INT))
            .field("gender",Types.STRING).from("mf")
            )
          .inAppendMode()
          .registerTableSource("testKafkaSource")
    
    
        bsTableEnv.sqlUpdate("CREATE TABLE testMysqlSink(" +
              "stu_num bigint, avg_age double, tumble_start TIMESTAMP(3), tumble_end TIMESTAMP(3)" +
              ") WITH (" +
              "'connector.type' = 'jdbc'," +
              "'connector.url' = 'jdbc:mysql://192.168.1.1:3306/test?autoReconnect=true'," +
              "'connector.table' = 'stu_20191023'," +
              "'connector.driver' = 'com.mysql.jdbc.Driver'," +
              "'connector.username' = '111111', " +
              "'connector.password' = '111111'," +
              "'connector.write.flush.max-rows' = '1'" +
              ")")
    
        bsTableEnv.sqlUpdate("INSERT INTO testMysqlSink " +
          "select " +
          "count(dinstinct stu_id) as stu_num" +
          ",avg(age) as avg_age" +
          ",cast(from_unixtime(unix_timestamp(cast(TUMBLE_START(rowtime, INTERVAL '1' MINUTE) as varchar)) + 8*60*60,'yyyy-MM-dd HH:mm:ss') as timestamp) as tumble_start" +
          ",cast(from_unixtime(unix_timestamp(cast(TUMBLE_END(rowtime, INTERVAL '1' MINUTE) as varchar)) + 8*60*60,'yyyy-MM-dd HH:mm:ss') as timestamp) as tumble_end " +
          "from testKafkaSource " +
          "group by TUMBLE(rowtime, INTERVAL '1' MINUTE)")
    
    	// set flink job name
        bsTableEnv.execute("test_1028_1")
      }
    }

## 更多好文 ##

*  [重要|Flink SQL与kafka整合的那些事儿][Flink SQL_kafka]

[Flink]: https://blog.csdn.net/baichoufei90/article/details/105674793
[Link 1]: https://blog.csdn.net/baichoufei90/article/details/90264272
[Flink_3-API]: https://blog.csdn.net/baichoufei90/article/details/82891909
[Flink SQL_kafka]: https://zhuanlan.zhihu.com/p/68116734