GBase Lookup

一、介绍

GBase维表，支持全量和异步方式
全量缓存:将维表数据全部加载到内存中，建议数据量不大，且数据不经常变动的场景使用。
异步缓存:使用异步方式查询数据，并将查询到的数据使用lru缓存到内存中，建议数据量大使用。

二、支持版本

GBase8a(8.6.2.43)

三、插件名称

SQL	gbase-x

四、参数说明

• connector

  • 描述：gbase-x
  • 必选：是
  • 参数类型：String
  • 默认值：无
    

• url

  • 描述：jdbc:gbase://localhost:9042/test，GBase jdbc 链接
  • 必选：是
  • 参数类型：String
  • 默认值：无
    

• schema-name

  • 描述：GBase 数据库
  • 必选：是
  • 参数类型：String
  • 默认值：
    

• table-name

  • 描述：表名
  • 必选：是
  • 参数类型：String
  • 默认值：无：
    

• username

  • 描述：username
  • 必选：是
  • 参数类型：String
  • 默认值：无
    

• password

  • 描述：password
  • 必选：是
  • 参数类型：String
  • 默认值：无
    

• lookup.cache-type

  • 描述：维表缓存类型(NONE、LRU、ALL)，默认LRU
  • 必选：否
  • 参数类型：string
  • 默认值：LRU
    

• lookup.cache-period

  • 描述：ALL维表每隔多久加载一次数据，默认3600000毫秒(一个小时)
  • 必选：否
  • 参数类型：string
  • 默认值：3600000
    

• lookup.cache.max-rows

  • 描述：lru维表缓存数据的条数，默认10000条
  • 必选：否
  • 参数类型：string
  • 默认值：10000
    

• lookup.cache.ttl

  • 描述：lru维表缓存数据的时间，默认60000毫秒(一分钟)
  • 必选：否
  • 参数类型：string
  • 默认值：60000
    

• lookup.fetch-size

  • 描述：ALL维表每次从数据库加载的条数，默认1000条
  • 必选：否
  • 参数类型：string
  • 默认值：1000
    

• lookup.parallelism

  • 描述：维表并行度
  • 必选：否
  • 参数类型：string
  • 默认值：无
    

五、数据类型

支持	BOOLEAN、TINYINT、SMALLINT、INT、BIGINT、FLOAT、DOUBLE、DECIMAL、STRING、VARCHAR、CHAR、TIMESTAMP、DATE、BINARY
暂不支持	ARRAY、MAP、STRUCT、UNION

六、脚本示例

CREATE TABLE source
(
    id   INT,
    name STRING,
    age  tinyint,
    PROCTIME AS PROCTIME()
) WITH (
      'connector' = 'kafka-x'
      ,'topic' = 'tiezhu_in_one'
      ,'properties.bootstrap.servers' = 'kudu1:9092'
      ,'properties.group.id' = 'luna_g'
      ,'scan.startup.mode' = 'earliest-offset'
      -- ,'scan.startup.mode' = 'latest-offset'
      ,'format' = 'json'
      ,'json.timestamp-format.standard' = 'SQL'
      );

CREATE TABLE side
(
    id             int,
    name           varchar,
    price          decimal,
    money          double,
    message        string,
    age            tinyint,
    todayTimestamp timestamp,
    todayDate      date,
    todayTime      time,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
      'connector' = 'gbase-x',
      'url' = 'jdbc:gbase://gbase:5258/dev_db',
      'table-name' = 'sink',
      'schema-name' = 'dev_db',
      'username' = 'dev',
      'password' = 'dev123',
--       'lookup.cache-type' = 'lru'
      'lookup.cache-type' = 'all'
      );

CREATE TABLE sink
(
    id             int,
    name           varchar,
    price          decimal,
    money          double,
    message        string,
    age            tinyint,
    todayDate      date,
    todayTime      time,
    todayTimestamp timestamp
) WITH (
      'connector' = 'stream-x'
      );

create
TEMPORARY view view_out
  as
select u.id             AS id,
       u.name           AS name,
       s.price          AS price,
       s.money          AS money,
       s.name           AS message,
       u.age            AS age,
       s.todayDate      AS todayDate,
       s.todayTime      AS todayTime,
       s.todayTimestamp AS todayTimestamp
from source u
         left join side FOR SYSTEM_TIME AS OF u.PROCTIME AS s
                   on u.id = s.id;

insert into sink
select *
from view_out;