#!/bin/bash
# 启动的主类参数信息,启动版本信息
APP_NAME=stream.forward.ne_
APP_VERSION=2020.01.17
JAR_PATH=../lib/uml-stream-forward-sxcg-1.0.0.jar
MAIN_CLASS=com.uml.tech.SparkApp

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


# 第三方依赖 jar
for jar in ../lib/*.jar
do
  if [ ${jar} != ${JAR_PATH} ] ; then
    LIBJARS=${jar},${LIBJARS}
  fi
done

# 检验上次 job 启动信息
appId=`yarn application -list | grep ${APP_NAME} | awk '{print $1}'`

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

echo "start submit spark application ......"

NUM_EXECUTORS=8

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --class ${MAIN_CLASS} \
  --jars ${LIBJARS} \
  --driver-memory ${DRIVER_MEMORY} \
  --num-executors ${NUM_EXECUTORS} \
  --executor-cores ${EXECUTOR_CORES} \
  --executor-memory ${EXECUTOR_MEMORY} \
  --conf spark.storage.memoryFraction=0.3 \
  --conf spark.app.name=${APP_NAME}${APP_VERSION} \
  --conf spark.streaming.kafka.maxRatePerPartition=${MAX_RATE_PER_PARTITION} \
  --conf spark.streaming.stopGracefullyOnShutdown=true \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  ${JAR_PATH} \
  ${WINDOW_DURATIONS}