#!/bin/sh

# # # # # # # # # # # # # # # # # # # # # #
# @author  :wangjf
# @revision:2019-03-27 17:01:50
# # # # # # # # # # # # # # # # # # # # # #

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 "-32 day $today")
expire_date_path=$(date +%Y/%m/%d -d "-32 day $today")

business='dsp_req'
source='manual'
storeSource="store"
output_path="${DM_DEVICE_TAG_DAILY_PATH}/${date_path}/${source}/${business}"
store_output_path="${DM_DEVICE_TAG_DAILY_PATH}/${date_path}/${storeSource}/${business}"
expire_path="${DM_DEVICE_TAG_DAILY_PATH}/${expire_date_path}/${source}/${business}"
expire_store_path="${DM_DEVICE_TAG_DAILY_PATH}/${expire_date_path}/${storeSource}/${business}"

app_tag_path="${APP_TAG_PATH}/${date_path}"

check_await "${app_tag_path}/_SUCCESS"

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

spark-submit --class mobvista.dmp.datasource.dsp.DspReqTagDaily \
     --name "mobvista.dmp.datasource.dsp.DspReqTagDaily_wangjf_${date}" \
     --conf spark.sql.shuffle.partitions=500 \
     --conf spark.default.parallelism=200 \
     --conf spark.sql.files.maxPartitionBytes=268435456 \
     --conf spark.memory.offHeap.enabled=true \
     --conf spark.memory.offHeap.size=4294967296 \
     --master yarn --deploy-mode cluster --executor-memory 4g --driver-memory 4g  --executor-cores 2  --num-executors 150 \
     ../${JAR} \
     -date ${date} -manualOutput ${output_path} -business ${business} -storeOutput ${store_output_path} -coalesce 200

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

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

# 补写成功标志
if hadoop fs -ls ${store_output_path} > /dev/null 2>&1
then
    hadoop fs -touchz ${store_output_path}/_SUCCESS
else
    hadoop fs -mkdir -p ${store_output_path}
    hadoop fs -touchz ${store_output_path}/_SUCCESS
fi

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