#!/bin/bash

APP_NAME=uml-stream-to-tdengine-farm-can
APP_VERSION=2022.11.01
APP_JAR=../lib/uml-stream-to-tdengine-farm-can-1.0.0.jar
MAIN_CLASS=com.uml.tech.spark.StreamingApp
DURATION=10
# 插入tdengine一个批次
BATCH_SIZE=100
# 0是earliest 1是latest 其他是current
OFFSET_FLAG=0
# 分区
PARTITION_NUM=8
cp ../lib/guava-29.0-jre.jar ./
GUAVA_JAR_PATH=guava-29.0-jre.jar

APP_ID=`yarn application -list |grep ${APP_NAME} |awk '{print $1}'`

if [ "${APP_ID}" != "" ] ; then
  echo `yarn application -kill ${APP_ID}`
fi

echo "Start Spark Application ${APP_NAME}${APP_VERSION} ...."

for jar in ../lib/*.jar
do
  if [ ${jar} != ${APP_JAR} ] ; then
    LIBJARS=${jar},${LIBJARS}
  fi
done

for file in ../conf/*; do
  if [ -f ${file} ]; then
    RESOURCES_FILES=${file},${RESOURCES_FILES}
  fi
done

spark-submit \
  --class ${MAIN_CLASS} \
  --master yarn \
  --deploy-mode cluster \
  --jars ${LIBJARS} \
  --conf spark.app.name=${APP_NAME}${APP_VERSION} \
  --conf spark.driver.cores=1 \
  --conf spark.driver.memory=2g \
  --conf spark.driver.maxResultSize=0 \
  --conf spark.executor.cores=1 \
  --conf spark.executor.instances=8 \
  --conf spark.executor.memory=4g \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.dynamicAllocation.enabled=false \
  --conf spark.shuffle.service.enabled=true \
  --conf spark.scheduler.mode=FIFO \
  --conf spark.streaming.concurrentJobs=1 \
  --conf spark.streaming.backpressure.enabled=false \
  --conf spark.streaming.kafka.maxRatePerPartition=10000 \
  --conf spark.streaming.stopGracefullyOnShutdown=true \
  --conf spark.executor.extraClassPath=${GUAVA_JAR_PATH} \
  $APP_JAR \
  ${DURATION} ${BATCH_SIZE} ${OFFSET_FLAG} ${PARTITION_NUM}