业务说明

我们业务过程中产生了一张客户标签表,记录的是客户与客户的标签。比如00后的客户A；属狗的客户B；这种业务模型常用在推荐系统中。业务中会有客户打上新的标签,也会去掉一些标签。每天来T-1文件需要与存量表进行关联更新删除操作。
当前我们的客户量级在10亿级,标签总量为300+,表的量级在百亿+

1.1 表的结构

表结构：

spark-sql> desc indv_cust_tag_info;
cust_num --客户号,
cust_label_cd --标签号,
data_date --变更日期,
valid_status_code --标签状态(0删除,1新增)

按照标签号进行分区:

spark-sql> show partitions indv_cust_tag_info;
cust_label_cd=label001
cust_label_cd=label002
cust_label_cd=label003
cust_label_cd=label004
cust_label_cd=label005
cust_label_cd=label006

1.2 文件样例

data_date|cust_num|label(使用,隔开)|valid_status_code|
20220314|cust00001|label001,label002,label003|1|
20220314|cust00001|label004,label005|0|
20220314|cust00002|label001,label002,label003|1|
20220315|cust00001|label001,label002|0|

2.1 将文件导入临时表并将标签串炸裂开

sql_0_0="
    DROP TABLE IF EXISTS indv_cust_tag_temp;
    CREATE TABLE indv_cust_tag_temp STORED AS ORC
    SELECT 
        cust_no,
        explode(split(cust_label_cd,',')) cust_label_Cd,
        valid_status_code,
        data_date
    FROM TABLE file_2_table_temp;
"
sparl-sql --master yarn --conf spark.sql.shuffle.partitions=9 --driver-memory 6g --exector-memory 8g --num-exectors 3 -e "${sql_0_0}"

3.1 找到发生变化的标签(新增与删除)

sql_0_1="
    select cust_label_cd from indv_cust_tag_temp group by  cust_label_cd;
"
unique_info=`sparl-sql --master yarn --conf spark.sql.shuffle.partitions=9 --driver-memory 6g --exector-memory 8g --num-exectors 3 -e "${sql_0_1}"`
unique_arr=($unique_info)

4. 循环每一个分区执行

for unique_label in ${unique_arr[@]}

4.1 关联存量表,将新增与删除之后的结果保存到标签子表

sql_1_1="
   DROP TABLE IF EXISTS indv_cust_tag_info_${unique_label};
   CREATE TABLE indv_cust_tag_info_${unique_label} STORED AS ORC
   SELECT
      nvl(b.cust_no,a.cust_no) cust_no,
      nvl(b.cust_label_cd,a.cust_label_cd) cust_label_cd,
      nvl(b.data_date,a.data_date) data_date
   FROM (select * from indv_cust_tag_info where cust_label_cd = '${unique_label}') a
   FULL JOIN (select * from indv_cust_tag_temp where cust_label_cd = '${unique_label}') b
   ON a.cust_no = b.cust_no
   WHERE nvl(b.valid_status_code,1) <> '0'; 
"

sparl-sql --master yarn --conf spark.sql.shuffle.partitions=9 --driver-memory 6g --exector-memory 8g --num-exectors 3 -e "${sql_1_1}"

4.2 用标签子表，替换掉存量表中的标签分区

sql_1_2="
     set hive.exec.max.dynamic.partitions=300;
     INSERT OVERWRITE TABLE indv_cust_tag_info
     PARTITION (cust_label_cd="${unique_label}")
     SELECT a.cust_no,a.data_date 
     FROM(
         SELECT cust_no,data_date
             row_num() over(partition by cust_no order by data_date) rn
         FROM indv_cust_tag_info_${unique_label}
     ) a WHERE a.rn=1;
"
sparl-sql --master yarn --conf spark.sql.shuffle.partitions=9 --driver-memory 6g --exector-memory 8g --num-exectors 3 -e "${sql_1_2}"