HBase 维表

简述

Easystream 支持使用 HBase 表作为维表进行 join。

示例

CREATE TABLE user_log (
    user_id VARCHAR,
    item_id VARCHAR,
    category_id VARCHAR,
    behavior VARCHAR,
    ts TIMESTAMP
) WITH (
    'connector.type' = 'kafka',
    'connector.version' = 'universal',
    'connector.topic' = 'user_behavior',
    'connector.startup-mode' = 'latest-offset',
    'connector.properties.zookeeper.connect' = 'xxx',
    'connector.properties.bootstrap.servers' = 'xxx',
    'connector.properties.group.id' = 'xxx',
    'update-mode' = 'append',
    'format.type' = 'json',
    'format.derive-schema' = 'true'
);
--dim
CREATE TABLE dim_table (
    user_id VARCHAR,
    base_info row<age VARCHAR, tall varchar>,
    info row<gener VARCHAR, children varchar>
) WITH (
    'connector.type' = 'hbase',
    'connector.version' = '1.4.3',
    'connector.property.version' = '2',
    'connector.table-name' = 'user_dim',
    'connector.zookeeper.quorum' = '*',
    'connector.zookeeper.znode.parent' = '*',
    'connector.lookup.cache.type' = 'lru'
);
-- sink
CREATE TABLE pvuvage_sink (
    dt VARCHAR,
    age INT,
    zhibiao1 BIGINT,
    zhibiao2 BIGINT
) WITH (
    'connector.type' = 'jdbc',
    'connector.url' = 'jdbc:mysql://10.122.173.167:3306/flink-test',
    'connector.table' = 'pvuv_age_sink_hbase',
    'connector.username' = '*',
    'connector.password' = '*',
    'connector.write.flush.max-rows' = '1'
);
INSERT INTO pvuvage_sink
select DATE_FORMAT(d.ts, 'yyyy-MM-dd HH:00') as dt, d.age, sum(d.tall) as zhibiao1, sum(d.childNum) as zhibiao2
from (
SELECT
  u.ts,
  cast(w.age as int)  as age,
  cast(w.tall as bigint) as tall,
  cast(w.children as bigint) as childNum,
  u.user_id,
  u.behavior
FROM (select user_id,item_id,category_id,behavior,ts,PROCTIME() as proc from user_log) as u
left join dim_table for system_time as of u.proc as w
on u.user_id = w.user_id
where w.age > 10
) as d
GROUP BY DATE_FORMAT(d.ts, 'yyyy-MM-dd HH:00') , d.age;

• Flink HBase Connector 官方文档

• Flink 维表 join 官方文档

With 参数