Hive 拉链表实践

CREATE TABLE member(
            member_id VARCHAR ( 64 ),
            phoneno VARCHAR ( 20 ),
            create_time datetime,
            update_time datetime );

CREATE TABLE member_delta
          (member_id string,
            phoneno string,
            create_time string,
            update_time string)
PARTITIONED BY (DAY string);

CREATE TABLE member_his_tmp
          (member_id string,
            phoneno string,
            effective_date date,
            expire_date date
            );

CREATE TABLE member_his
          (member_id string,
            phoneno string,
            effective_date date,
            expire_date date);

INSERT overwrite TABLE member_his
SELECT
  member_id,
  phoneno,
  to_date ( create_time ) AS effective_date,
 '3000-12-31'
FROM
member_delta
WHERE
DAY = '2019-11-08'

INSERT overwrite TABLE member_his_tmp
SELECT *
FROM
  (
-- 2019-11-09增量数据，代表最新的状态，该数据的生效时间是2019-11-09，过期时间为3000-12-31
-- 这些增量的数据需要被全部加载到历史拉链表中
SELECT member_id,
       phoneno,
       '2019-11-09' effective_date,
                    '3000-12-31' expire_date
   FROM member_delta
   WHERE DAY='2019-11-09'
   UNION ALL 
-- 用当前为生效状态的拉链数据，去left join 增量数据，
-- 如果匹配得上，则表示该数据已发生了更新，
-- 此时，需要将发生更新的数据的过期时间更改为当前时间.
-- 如果匹配不上，则表明该数据没有发生更新，此时过期时间不变
SELECT a.member_id,
       a.phoneno,
       a.effective_date,
       if(b.member_id IS NULL, to_date(a.expire_date), to_date(b.day)) expire_date
   FROM
     (SELECT *
      FROM member_his
      WHERE expire_date='3000-12-31') a
   LEFT JOIN
     (SELECT *
      FROM member_delta
      WHERE DAY='2019-11-09') b ON a.member_id=b.member_id)his

INSERT overwrite TABLE member_his
SELECT *
FROM member_his_tmp

INSERT overwrite TABLE member_his_tmp
SELECT *
FROM
(
-- 2019-11-10增量数据，代表最新的状态，该数据的生效时间是2019-11-10，过期时间为3000-12-31
-- 这些增量的数据需要被全部加载到历史拉链表中
SELECT member_id,
      phoneno,
      '2019-11-10' effective_date,
                   '3000-12-31' expire_date
  FROM member_delta
  WHERE DAY='2019-11-10'
  UNION ALL
-- 用当前为生效状态的拉链数据，去left join 增量数据，
-- 如果匹配得上，则表示该数据已发生了更新，
-- 此时，需要将发生更新的数据的过期时间更改为当前时间.
-- 如果匹配不上，则表明该数据没有发生更新，此时过期时间不变
SELECT a.member_id,
      a.phoneno,
      a.effective_date,
      if(b.member_id IS NULL, to_date(a.expire_date), to_date(b.day)) expire_date
  FROM
    (SELECT *
    FROM member_his
    WHERE expire_date='3000-12-31') a
  LEFT JOIN
    (SELECT *
    FROM member_delta
    WHERE DAY='2019-11-10') b ON a.member_id=b.member_id)his

#!/bin/bash

#如果是输入的日期按照取输入日期；如果没输入日期取当前时间的前一天
if [ -n "$1" ] ;then
    do_date=$1
else
    do_date=`date -d "-1 day" +%F`
fi

sql="

INSERT overwrite TABLE member_his_tmp
SELECT *
FROM
  (
-- 2019-11-10增量数据，代表最新的状态，该数据的生效时间是2019-11-10，过期时间为3000-12-31
-- 这些增量的数据需要被全部加载到历史拉链表中
SELECT member_id,
       phoneno,
       '$do_date' effective_date,
       '3000-12-31' expire_date
   FROM member_delta
   WHERE DAY='$do_date'
   UNION ALL
-- 用当前为生效状态的拉链数据，去left join 增量数据，
-- 如果匹配得上，则表示该数据已发生了更新，
-- 此时，需要将发生更新的数据的过期时间更改为当前时间.
-- 如果匹配不上，则表明该数据没有发生更新，此时过期时间不变
SELECT a.member_id,
       a.phoneno,
       a.effective_date,
       if(b.member_id IS NULL, to_date(a.expire_date), to_date(b.day)) expire_date
   FROM
     (SELECT *
      FROM member_his
      WHERE expire_date='3000-12-31') a
   LEFT JOIN
     (SELECT *
      FROM member_delta
      WHERE DAY='$do_date') b ON a.member_id=b.member_id)his;
"

$hive -e "$sql"