#!/bin/bash

APP_NAME=stream.command.farm_
APP_VERSION=2020.8.24
APP_JAR=../lib/uml-stream-command-farm-2.0.0.jar
MAIN_CLASS=com.uml.tech.spark.StreamingApp

# 窗口周期
WINDOW_DURATIONS=5000
# 项目启动时，初始化executor的数量
INITIAL_EXECUTORS=1
# 最小executor数量
MIN_EXECUTORS=1
# 最大executor数量
MAX_EXECUTORS=10
# driver核心数
DRIVER_CORES=1
# driver内存
DRIVER_MEMORY=2g
# executor核心数
EXECUTOR_CORES=1
# executor内存
EXECUTOR_MEMORY=2g
# executor空闲多久没有收到任务就被杀掉
EXECUTOR_IDLE_TIMEOUT=10s
# executor执行任务超过多久，就创建新的executor，最大创建到 MAX_EXECUTORS 个，指数型个数创建，1、2、4、8这种方式创建
SCHEDULER_BACKLOG_TIMEOUT=5s
# 再次触发创建新executor的间隔时间
SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT=60s
# 每批从kafka分区中拉取多少数据
MAX_RATE_PER_PARTITION=10000

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 ../resources/*; do
  RESOURCES_FILES=${file},${RESOURCES_FILES}
done

RESOURCES_FILES=../resources/application.properties

spark-submit \
  --class ${MAIN_CLASS} \
  --master yarn \
  --deploy-mode cluster \
  --jars ${LIBJARS} \
  --files ${RESOURCES_FILES} \
  --conf spark.app.name=${APP_NAME}${APP_VERSION} \
  --conf spark.driver.cores=${DRIVER_CORES} \
  --conf spark.driver.memory=${DRIVER_MEMORY} \
  --conf spark.driver.maxResultSize=0 \
  --conf spark.executor.cores=${EXECUTOR_CORES} \
  --conf spark.executor.memory=${EXECUTOR_MEMORY} \
  --conf spark.executor.extraJavaOptions=" -XX:+UseG1GC" \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.initialExecutors=${INITIAL_EXECUTORS} \
  --conf spark.dynamicAllocation.minExecutors=${MIN_EXECUTORS} \
  --conf spark.dynamicAllocation.maxExecutors=${MAX_EXECUTORS} \
  --conf spark.dynamicAllocation.executorIdleTimeout=${EXECUTOR_IDLE_TIMEOUT} \
  --conf spark.dynamicAllocation.schedulerBacklogTimeout=${SCHEDULER_BACKLOG_TIMEOUT} \
  --conf spark.dynamicAllocation.sustainedSchedulerBacklogTimeout=${SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT} \
  --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=${MAX_RATE_PER_PARTITION} \
  --conf spark.streaming.stopGracefullyOnShutdown=true \
  $APP_JAR \
  ${WINDOW_DURATIONS}
