# !/bin/sh
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 -d "$ScheduleTime 91 days ago" +"%Y%m%d")
expire_date_path=$(date -d "$ScheduleTime 91 days ago" +"%Y/%m/%d")
business='ss'

input_path="${EVENT_SS_PATH}/${date_path}"

check_await "${input_path}/virginia/23/_SUCCESS"
## CAMPAIGN_LIST_DATE=$(get_recently_date "${ETL_CAMPAIN_LIST_PATH}" "${date}" "")
OUTPUT_PATH="${ODS_DMP_EVENT_ORG}/${date_path}/${business}"
EXPIRE_PATH="${ODS_DMP_EVENT_ORG}/${expire_date_path}/${business}"

# dwh.ods_dmp_event_org
hadoop fs -rmr ${OUTPUT_PATH}

spark-submit --class mobvista.dmp.datasource.event_tag.DmpSsEventTag \
      --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
      --conf spark.shuffle.file.buffer=256k \
      --conf spark.reducer.maxSizeInFlight=100m \
      --conf spark.shuffle.io.maxRetries=60  \
      --conf spark.shuffle.io.retryWait=60s \
      --conf spark.network.timeout=1200 \
      --conf spark.shuffle.compress=true \
      --conf spark.driver.maxResultSize=2024M  \
      --conf spark.kryoserializer.buffer.max=300M \
      --conf spark.shuffle.spill.compress=true \
      --name dmp_ss_event_tag \
      --conf spark.locality.wait=0 \
      --conf spark.shuffle.io.numConnectionsPerPeer=10 \
      --conf spark.shuffle.consolidateFiles=true \
      --conf spark.dynamicAllocation.enabled=false \
      --conf spark.akka.frameSize=100 \
     --master yarn --deploy-mode cluster --driver-memory 4g --executor-cores 2 --executor-memory 4G --num-executors 20 \
     ../${JAR} \
      -today ${date}  -input "${input_path}/*/*" -output  ${OUTPUT_PATH}  -coalesce 100

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

mount_partition "ods_dmp_event_org" "dt='${date}',business='ss'" "$OUTPUT_PATH"
# 删除过期的分区及删除对应路径
unmount_partition "ods_dmp_event_org" "dt='${expire_date}',business='ss'" "${EXPIRE_PATH}"

hadoop fs -touchz ${OUTPUT_PATH}/_SUCCESS

