set mapreduce.job.name=Build Hive Global Dict - extract distinct value;
USE NightlyBuild;
set hive.exec.compress.output=false;
set hive.mapred.mode=unstrict;

-- create hive global dictionary table
CREATE TABLE IF NOT EXISTS NightlyBuild.ValidationCube_global_dict
( dict_key STRING COMMENT '', 
   dict_val INT COMMENT '' 
) 
COMMENT 'Hive Global Dictionary' 
PARTITIONED BY (dict_column string) 
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' 
STORED AS TEXTFILE; 

-- create distinct value table
DROP TABLE IF EXISTS kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f__group_by; 
CREATE TABLE IF NOT EXISTS kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f__group_by 
( 
   dict_key STRING COMMENT '' 
) 
COMMENT '' 
PARTITIONED BY (dict_column string) 
STORED AS TEXTFILE 
;

-- create segment level dictionary table
DROP TABLE IF EXISTS kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f_global_dict; 
CREATE TABLE IF NOT EXISTS kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f_global_dict 
( 
   dict_key STRING COMMENT '' , 
   dict_val STRING COMMENT '' 
) 
COMMENT '' 
PARTITIONED BY (dict_column string) 
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' 
STORED AS TEXTFILE 
;

-- insert data into distinct value table
INSERT OVERWRITE TABLE kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f__group_by 
PARTITION (dict_column = 'USERACTIONLOGSAMPLE_PV_ID') 
SELECT a.DICT_KEY FROM (
  SELECT 
   USERACTIONLOGSAMPLE_PV_ID as DICT_KEY 
  FROM kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f
  GROUP BY USERACTIONLOGSAMPLE_PV_ID) a 
    LEFT JOIN 
  (SELECT DICT_KEY FROM NightlyBuild.ValidationCube_global_dict    
   WHERE DICT_COLUMN = 'USERACTIONLOGSAMPLE_PV_ID' ) b 
ON a.DICT_KEY = b.DICT_KEY 
WHERE b.DICT_KEY IS NULL 
;
INSERT OVERWRITE TABLE kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f__group_by 
PARTITION (dict_column = 'USERACTIONLOGSAMPLE_PLAY_ID') 
SELECT a.DICT_KEY FROM (
  SELECT 
   USERACTIONLOGSAMPLE_PLAY_ID as DICT_KEY 
  FROM kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f
  GROUP BY USERACTIONLOGSAMPLE_PLAY_ID) a 
    LEFT JOIN 
  (SELECT DICT_KEY FROM NightlyBuild.ValidationCube_global_dict    
   WHERE DICT_COLUMN = 'USERACTIONLOGSAMPLE_PLAY_ID' ) b 
ON a.DICT_KEY = b.DICT_KEY 
WHERE b.DICT_KEY IS NULL 
;

-- calculate max dict id 
INSERT OVERWRITE TABLE  kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f__group_by PARTITION (DICT_COLUMN = 'KYLIN_MAX_DISTINCT_COUNT') 
SELECT CONCAT_WS(',', tc.dict_column, cast(tc.total_distinct_val AS String), if(tm.max_dict_val is null, '0', cast(max_dict_val as string))) 
FROM (
    SELECT dict_column, count(1) total_distinct_val
    FROM NightlyBuild.kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f__group_by
    WHERE DICT_COLUMN != 'KYLIN_MAX_DISTINCT_COUNT'
    GROUP BY dict_column) tc 
LEFT JOIN (
    SELECT dict_column, if(max(dict_val) is null, 0, max(dict_val)) as max_dict_val 
    FROM NightlyBuild.ValidationCube_global_dict
    GROUP BY dict_column) tm 
ON tc.dict_column = tm.dict_column;

set mapreduce.job.name=Build Hive Global Dict - merge to dict table;
USE NightlyBuild;
set hive.mapred.mode=unstrict;

-- data is prepared in previous two MR jobs, create partition for segment level dictionary base on prepared files
ALTER TABLE kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f_global_dict ADD IF NOT EXISTS PARTITION (dict_column='USERACTIONLOGSAMPLE_PV_ID'); 
ALTER TABLE kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f_global_dict ADD IF NOT EXISTS PARTITION (dict_column='USERACTIONLOGSAMPLE_PLAY_ID'); 


-- merge segment level dictionary into global dictionary
INSERT OVERWRITE TABLE NightlyBuild.ValidationCube_global_dict 
PARTITION (dict_column = 'USERACTIONLOGSAMPLE_PV_ID') 
 SELECT dict_key, dict_val FROM NightlyBuild.ValidationCube_global_dict 
 WHERE dict_column = 'USERACTIONLOGSAMPLE_PV_ID' 
  UNION ALL 
 SELECT dict_key, dict_val FROM kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f_global_dict 
 WHERE dict_column = 'USERACTIONLOGSAMPLE_PV_ID' ;

INSERT OVERWRITE TABLE NightlyBuild.ValidationCube_global_dict 
PARTITION (dict_column = 'USERACTIONLOGSAMPLE_PLAY_ID') 
 SELECT dict_key, dict_val FROM NightlyBuild.ValidationCube_global_dict 
 WHERE dict_column = 'USERACTIONLOGSAMPLE_PLAY_ID' 
  UNION ALL 
 SELECT dict_key, dict_val FROM kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f_global_dict 
 WHERE dict_column = 'USERACTIONLOGSAMPLE_PLAY_ID' ;

set mapreduce.job.name=Build Hive Global Dict - replace intermediate table;
USE NightlyBuild;
set hive.exec.compress.output=false; set hive.mapred.mode=unstrict;

-- encode/replace flat table for specific column
INSERT OVERWRITE TABLE NightlyBuild.kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f 
SELECT 
a.USERACTIONLOGSAMPLE_UID 
,a.USERACTIONLOGSAMPLE_ACT_TYPE 
,a.USERACTIONLOGSAMPLE_PAGE_ID 
,a.USERACTIONLOGSAMPLE_DEVICE_BRAND 
,a.USERACTIONLOGSAMPLE_ITEM_TYPE_ID 
,a.USERACTIONLOGSAMPLE_CITY 
,a.USERACTIONLOGSAMPLE_PART_DT 
,b.dict_val 
,a.USERACTIONLOGSAMPLE_PLAY_ID 
FROM NightlyBuild.kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f a 
LEFT OUTER JOIN 
 (SELECT dict_key, dict_val FROM NightlyBuild.ValidationCube_global_dict WHERE dict_column = 'USERACTIONLOGSAMPLE_PV_ID') b 
ON a.USERACTIONLOGSAMPLE_PV_ID = b.dict_key;


INSERT OVERWRITE TABLE NightlyBuild.kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f 
SELECT 
a.USERACTIONLOGSAMPLE_UID 
,a.USERACTIONLOGSAMPLE_ACT_TYPE 
,a.USERACTIONLOGSAMPLE_PAGE_ID 
,a.USERACTIONLOGSAMPLE_DEVICE_BRAND 
,a.USERACTIONLOGSAMPLE_ITEM_TYPE_ID 
,a.USERACTIONLOGSAMPLE_CITY 
,a.USERACTIONLOGSAMPLE_PART_DT 
,a.USERACTIONLOGSAMPLE_PV_ID 
,b.dict_val 
FROM NightlyBuild.kylin_intermediate_validationcube_bb2091dc_82ba_f77f_86b9_6f83a1af639f a 
LEFT OUTER JOIN 
 (SELECT dict_key, dict_val FROM NightlyBuild.ValidationCube_global_dict WHERE dict_column = 'USERACTIONLOGSAMPLE_PLAY_ID') b 
ON a.USERACTIONLOGSAMPLE_PLAY_ID = b.dict_key;