转载 Flink SQL join INTERVAL 代码案例

悠悠 2021-12-04 00:25 285阅读 0赞

1，下面的案例就是 在下单之后，在一个小时之内付款的数据才是有效数据，使用  JOIN INTERVAL  来实现，这可是知识盲点，不过真的是好用。
    
    package com.coder.flink.core.table_sql.sql_function
    
    import java.sql.Timestamp
    
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.table.api.TableEnvironment
    import org.apache.flink.table.api.scala._
    import org.apache.flink.types.Row
    
    import scala.collection.mutable
    
    object SimpleTimeIntervalJoin {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val tEnv = TableEnvironment.getTableEnvironment(env)
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        // 构造订单数据
        val ordersData = new mutable.MutableList[(String, String, Timestamp)]
        ordersData.+=(("001", "iphone", new Timestamp(1545800002000L)))
        ordersData.+=(("002", "mac", new Timestamp(1545800003000L)))
        ordersData.+=(("003", "book", new Timestamp(1545800004000L)))
        ordersData.+=(("004", "cup", new Timestamp(1545800018000L)))
    
        // 构造付款表
        val paymentData = new mutable.MutableList[(String, String, Timestamp)]
        paymentData.+=(("001", "alipay", new Timestamp(1545803501000L)))
        paymentData.+=(("002", "card", new Timestamp(1545803602000L)))
        paymentData.+=(("003", "card", new Timestamp(1545803610000L)))
        paymentData.+=(("004", "alipay", new Timestamp(1545803611000L)))
        val orders = env
          .fromCollection(ordersData)
          .assignTimestampsAndWatermarks(new TimestampExtractor[String, String]())
          .toTable(tEnv, 'orderId, 'productName, 'orderTime.rowtime)
        val ratesHistory = env
          .fromCollection(paymentData)
          .assignTimestampsAndWatermarks(new TimestampExtractor[String, String]())
          .toTable(tEnv, 'orderId, 'payType, 'payTime.rowtime)
    
        tEnv.registerTable("Orders", orders)
        tEnv.registerTable("Payment", ratesHistory)
    
        var sqlQuery =
          """
            |SELECT
            |  o.orderId,
            |  o.productName,
            |  p.payType,
            |  o.orderTime,
            |  cast(payTime as timestamp) as payTime
            |FROM
            |  Orders AS o JOIN Payment AS p ON o.orderId = p.orderId AND
            | p.payTime BETWEEN orderTime AND orderTime + INTERVAL '1' HOUR
            |""".stripMargin
        tEnv.registerTable("TemporalJoinResult", tEnv.sqlQuery(sqlQuery))
    
        val result = tEnv.scan("TemporalJoinResult").toAppendStream[Row]
        result.print()
        env.execute()
      }
    
    }
    
    class TimestampExtractor[T1, T2]
      extends BoundedOutOfOrdernessTimestampExtractor[(T1, T2, Timestamp)](Time.seconds(10)) {
      override def extractTimestamp(element: (T1, T2, Timestamp)): Long = {
        element._3.getTime
      }
    }

完整案例地址：[https://enjoyment.cool/2019/03/22/Apache%20Flink%20%E6%BC%AB%E8%B0%88%E7%B3%BB%E5%88%97%20-%20Time%20Interval%20JOIN/\#more][https_enjoyment.cool_2019_03_22_Apache_20Flink_20_E6_BC_AB_E8_B0_88_E7_B3_BB_E5_88_97_20-_20Time_20Interval_20JOIN_more]

[https_enjoyment.cool_2019_03_22_Apache_20Flink_20_E6_BC_AB_E8_B0_88_E7_B3_BB_E5_88_97_20-_20Time_20Interval_20JOIN_more]: https://enjoyment.cool/2019/03/22/Apache%20Flink%20%E6%BC%AB%E8%B0%88%E7%B3%BB%E5%88%97%20-%20Time%20Interval%20JOIN/#more