#!/bin/bash

# # # # # # # # # # # # # # # # # # # # # #
# @file    : dmp_interest_tag.sh
# @author  : jinfeng.wang
# @time    : 2020-05-22 16:06:07
# # # # # # # # # # # # # # # # # # # # # #

source ../dmp_env.sh

today=${ScheduleTime:-$1}
date=$(date +"%Y%m%d" -d "-1 day $today")
date_path=$(date +%Y/%m/%d -d "-1 day $today")
expire_date=$(date +%Y%m%d -d "-4 day $today")
expire_date_path=$(date +%Y/%m/%d -d "-4 day $today")

business='14days'
source='manual'
storeSource="store"

output_path="${DMP_DEVICE_TAG_PATH}/${date_path}/${source}/${business}"
store_output_path="${DMP_DEVICE_TAG_PATH}/${date_path}/${storeSource}/${business}"
expire_path="${DMP_DEVICE_TAG_PATH}/${expire_date_path}/${source}/${business}"
expire_store_path="${DMP_DEVICE_TAG_PATH}/${expire_date_path}/${storeSource}/${business}"
input_path="${DMP_INSTALL_LIST}/${date_path}/${business}"
app_tag_path="${APP_TAG_PATH}/${date_path}"

check_await "${app_tag_path}/_SUCCESS"
check_await "${input_path}/_SUCCESS"

hadoop fs -rmr ${output_path}
hadoop fs -rmr ${store_output_path}

spark-submit --class mobvista.dmp.datasource.newtag.MatchInterestTagDailyV2 \
     --name "MatchInterestTagDailyV2.${date}.${business}" \
     --conf spark.sql.shuffle.partitions=12000 \
     --conf spark.default.parallelism=2000 \
     --conf spark.kryoserializer.buffer.max=512m \
     --conf spark.kryoserializer.buffer=64m \
     --conf spark.sql.adaptive.enabled=true \
     --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=268435456 \
     --files ${HIVE_SITE_PATH} \
     --master yarn --deploy-mode cluster --executor-memory 18g --driver-memory 6g  --executor-cores 5  --num-executors 80 \
     ../${JAR} \
     -date ${date} -manualOutput ${output_path} -business ${business} -storeOutput ${store_output_path} -coalesce 4000

if [[ $? -ne 0 ]]; then
    exit 255
fi

# 挂载分区
mount_partition "dmp_device_tag_daily" "dt='${date}', source='${source}', business='${business}'" "${output_path}"
mount_partition "dmp_device_tag_daily" "dt='${date}', source='${storeSource}', business='${business}'" "${store_output_path}"

# 补写成功标志
hadoop fs -touchz ${store_output_path}/_SUCCESS

# 删除过期的分区及删除对应路径
unmount_partition "dmp_device_tag_daily" "dt='${expire_date}', source='${source}', business='${business}'" "${expire_path}"
unmount_partition "dmp_device_tag_daily" "dt='${expire_date}', source='${storeSource}', business='${business}'" "${expire_store_path}"