Flink SQL Temporal Join (快照Join)

Temporal Join的使用条件

快照Join原理图解

快照Join Flink SQL语法

示例代码 temporal Join – inner / left Join

temporal Join – inner Join

package com.imooc.scala.sqljoin

import java.time.ZoneId

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

/**
 * 时态/快照Join（Temporal Join）之 Inner Join
 * Created by xuwei
 */
object TemporalJoin_InnerJoin {

  def main(args: Array[String]): Unit = {
    //创建执行环境
    val settings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()
    val tEnv = TableEnvironment.create(settings)

    //指定国内的时区
    tEnv.getConfig.setLocalTimeZone(ZoneId.of("Asia/Shanghai"))

    //设置Source自动置为idle的时间--如果数据源可以一直正常产生数据，则不需要配置此参数！！！
    tEnv.getConfig.set("table.exec.source.idle-timeout","10s")

    //订单表--仅追加表
    val OrdersTableSql =
      """
        |CREATE TABLE orders(
        |  order_id BIGINT,
        |  price DECIMAL(32,2),
        |  currency STRING,
        |  ts BIGINT,
        |  order_time AS TO_TIMESTAMP_LTZ(ts,3),
        |  WATERMARK FOR order_time AS order_time
        |)WITH(
        |  'connector' = 'kafka',
        |  'topic' = 'orders',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'properties.group.id' = 'gid-sql-orders',
        |  'scan.startup.mode' = 'latest-offset',
        |  'format' = 'json',
        |  'json.fail-on-missing-field' = 'false',
        |  'json.ignore-parse-errors' = 'true'
        |)
        |""".stripMargin
    tEnv.executeSql(OrdersTableSql)

    //汇率表-版本表
    val CurrencyRatesTableSql =
      """
        |CREATE TABLE currency_rates(
        |  currency STRING,
        |  rate DECIMAL(32,2),
        |  ts BIGINT,
        |  update_time AS TO_TIMESTAMP_LTZ(ts,3),
        |  WATERMARK FOR update_time AS update_time,
        |  PRIMARY KEY (currency) NOT ENFORCED
        |)WITH(
        |  'connector' = 'upsert-kafka',
        |  'topic' = 'currency_rates',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'properties.group.id' = 'gid-sql-rates',
        |  'key.format' = 'json',
        |  'value.format' = 'json'
        |)
        |""".stripMargin
    tEnv.executeSql(CurrencyRatesTableSql)


    //结果表
    val resTableSql =
      """
        |CREATE TABLE orders_rates(
        |  order_id BIGINT,
        |  price DECIMAL(32,2),
        |  currency STRING,
        |  rate DECIMAL(32,2),
        |  order_time TIMESTAMP_LTZ(3)
        |)WITH(
        |  'connector' = 'kafka',
        |  'topic' = 'orders_rates',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'format' = 'json',
        |  'sink.partitioner' = 'default'
        |)
        |""".stripMargin
    tEnv.executeSql(resTableSql)

    //关联订单表和汇率表
    val joinSql =
      """
        |INSERT INTO orders_rates
        |SELECT
        |  order_id,
        |  price,
        |  orders.currency,
        |  rate,
        |  order_time
        |FROM orders
        |INNER JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time
        |ON orders.currency = currency_rates.currency
        |""".stripMargin

    //关联订单表和汇率表--SQL中给表起别名
    val joinSqlWithAs =
      """
        |INSERT INTO orders_rates
        |SELECT
        |  os.order_id,
        |  os.price,
        |  os.currency,
        |  cr.rate,
        |  os.order_time
        |FROM orders AS os
        |INNER JOIN currency_rates FOR SYSTEM_TIME AS OF os.order_time AS cr
        |ON os.currency = cr.currency
        |""".stripMargin
    tEnv.executeSql(joinSql)

  }

}

temporal Join – left Join

package com.imooc.scala.sqljoin

import java.time.ZoneId

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

/**
 * 时态/快照Join（Temporal Join）之 Left Join
 * Created by xuwei
 */
object TemporalJoin_LeftJoin {

  def main(args: Array[String]): Unit = {
    //创建执行环境
    val settings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()
    val tEnv = TableEnvironment.create(settings)

    //指定国内的时区
    tEnv.getConfig.setLocalTimeZone(ZoneId.of("Asia/Shanghai"))

    //设置Source自动置为idle的时间--如果数据源可以一直正常产生数据，则不需要配置此参数！！！
    tEnv.getConfig.set("table.exec.source.idle-timeout","10s")

    //订单表--仅追加表
    val OrdersTableSql =
      """
        |CREATE TABLE orders(
        |  order_id BIGINT,
        |  price DECIMAL(32,2),
        |  currency STRING,
        |  ts BIGINT,
        |  order_time AS TO_TIMESTAMP_LTZ(ts,3),
        |  WATERMARK FOR order_time AS order_time
        |)WITH(
        |  'connector' = 'kafka',
        |  'topic' = 'orders',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'properties.group.id' = 'gid-sql-orders',
        |  'scan.startup.mode' = 'latest-offset',
        |  'format' = 'json',
        |  'json.fail-on-missing-field' = 'false',
        |  'json.ignore-parse-errors' = 'true'
        |)
        |""".stripMargin
    tEnv.executeSql(OrdersTableSql)

    //汇率表-版本表
    val CurrencyRatesTableSql =
      """
        |CREATE TABLE currency_rates(
        |  currency STRING,
        |  rate DECIMAL(32,2),
        |  ts BIGINT,
        |  update_time AS TO_TIMESTAMP_LTZ(ts,3),
        |  WATERMARK FOR update_time AS update_time,
        |  PRIMARY KEY (currency) NOT ENFORCED
        |)WITH(
        |  'connector' = 'upsert-kafka',
        |  'topic' = 'currency_rates',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'properties.group.id' = 'gid-sql-rates',
        |  'key.format' = 'json',
        |  'value.format' = 'json'
        |)
        |""".stripMargin
    tEnv.executeSql(CurrencyRatesTableSql)


    //结果表
    val resTableSql =
      """
        |CREATE TABLE orders_rates(
        |  order_id BIGINT,
        |  price DECIMAL(32,2),
        |  currency STRING,
        |  rate DECIMAL(32,2),
        |  order_time TIMESTAMP_LTZ(3)
        |)WITH(
        |  'connector' = 'kafka',
        |  'topic' = 'orders_rates',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'format' = 'json',
        |  'sink.partitioner' = 'default'
        |)
        |""".stripMargin
    tEnv.executeSql(resTableSql)

    //关联订单表和汇率表
    val joinSql =
      """
        |INSERT INTO orders_rates
        |SELECT
        |  order_id,
        |  price,
        |  orders.currency,
        |  rate,
        |  order_time
        |FROM orders
        |LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time
        |ON orders.currency = currency_rates.currency
        |""".stripMargin

    tEnv.executeSql(joinSql)

  }

}

使用版本表视图实现快照JOIN代码示例

package com.imooc.scala.sqljoin

import java.time.ZoneId

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

/**
 * 版本表视图JOIN
 * Created by xuwei
 */
object TemporalJoin_LeftJoin_VersionedView {

  def main(args: Array[String]): Unit = {
    //创建执行环境
    val settings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()
    val tEnv = TableEnvironment.create(settings)

    //指定国内的时区
    tEnv.getConfig.setLocalTimeZone(ZoneId.of("Asia/Shanghai"))

    //设置Source自动置为idle的时间--如果数据源可以一直正常产生数据，则不需要配置此参数！！！
    tEnv.getConfig.set("table.exec.source.idle-timeout","10s")

    //订单表--仅追加表
    val OrdersTableSql =
      """
        |CREATE TABLE orders(
        |  order_id BIGINT,
        |  price DECIMAL(32,2),
        |  currency STRING,
        |  ts BIGINT,
        |  order_time AS TO_TIMESTAMP_LTZ(ts,3),
        |  WATERMARK FOR order_time AS order_time
        |)WITH(
        |  'connector' = 'kafka',
        |  'topic' = 'orders',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'properties.group.id' = 'gid-sql-orders',
        |  'scan.startup.mode' = 'latest-offset',
        |  'format' = 'json',
        |  'json.fail-on-missing-field' = 'false',
        |  'json.ignore-parse-errors' = 'true'
        |)
        |""".stripMargin
    tEnv.executeSql(OrdersTableSql)

    //汇率表-仅追加表
    val CurrencyRatesTableSql =
      """
        |CREATE TABLE currency_rates(
        |  currency STRING,
        |  rate DECIMAL(32,2),
        |  ts BIGINT,
        |  update_time AS TO_TIMESTAMP_LTZ(ts,3),
        |  WATERMARK FOR update_time AS update_time
        |)WITH(
        |  'connector' = 'kafka',
        |  'topic' = 'currency_rates',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'properties.group.id' = 'gid-sql-rates',
        |  -- 为了便于演示，在这使用latest-offset，每次启动都使用最新的数据
        |  'scan.startup.mode' = 'latest-offset',
        |  'format' = 'json',
        |  'json.fail-on-missing-field' = 'false',
        |  'json.ignore-parse-errors' = 'true'
        |)
        |""".stripMargin
    tEnv.executeSql(CurrencyRatesTableSql)

    //定义版本表视图(间接实现版本表的效果)
    val VersionedRatesViewSql =
      """
        |CREATE VIEW versioned_rates AS
        |SELECT currency, rate, update_time -- 1:指定update_time为时间字段
        |  FROM (
        |      SELECT *,
        |      ROW_NUMBER() OVER (PARTITION BY currency -- 2:指定currency为主键
        |        ORDER BY update_time DESC      -- 3:ORDER BY中必须指定TIMESTAMP格式的时间字段
        |      ) AS row_num
        |      FROM currency_rates
        |  )
        |WHERE row_num = 1
        |""".stripMargin
    tEnv.executeSql(VersionedRatesViewSql)


    //结果表
    val resTableSql =
      """
        |CREATE TABLE orders_rates(
        |  order_id BIGINT,
        |  price DECIMAL(32,2),
        |  currency STRING,
        |  rate DECIMAL(32,2),
        |  order_time TIMESTAMP_LTZ(3)
        |)WITH(
        |  'connector' = 'kafka',
        |  'topic' = 'orders_rates',
        |  'properties.bootstrap.servers' = 'bigdata01:9092,bigdata02:9092,bigdata03:9092',
        |  'format' = 'json',
        |  'sink.partitioner' = 'default'
        |)
        |""".stripMargin
    tEnv.executeSql(resTableSql)

    //关联订单表和汇率表
    val joinSql =
      """
        |INSERT INTO orders_rates
        |SELECT
        |  order_id,
        |  price,
        |  orders.currency,
        |  rate,
        |  order_time
        |FROM orders
        |LEFT JOIN versioned_rates FOR SYSTEM_TIME AS OF orders.order_time
        |ON orders.currency = versioned_rates.currency
        |""".stripMargin

    tEnv.executeSql(joinSql)

  }

}