#!/bin/bash

APP_NAME=stream-fence
APP_VERSION=1.0.0
APP_JAR=../lib/${APP_NAME}-1.0.0.jar
MAIN_CLASS=uml.tech.bigdata.spark.StreamingApp
KAFKA_JAR_PATH=kafka-clients-0.10.2.2.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 ../resources/*; do
  if [ -f ${file} ]; then
    RESOURCES_FILES=${file},${RESOURCES_FILES}
  fi
done

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=1 \
  --conf spark.driver.memory=512m \
  --conf spark.driver.maxResultSize=0 \
  --conf spark.executor.cores=1 \
  --conf spark.executor.instances=1 \
  --conf spark.executor.memory=1g \
  --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=100000 \
  --conf spark.streaming.stopGracefullyOnShutdown=true \
      --conf spark.driver.extraClassPath=${KAFKA_JAR_PATH} \
      --conf spark.executor.extraClassPath=${KAFKA_JAR_PATH} \
  $APP_JAR \
  10000 8 curr
# 窗口周期 分区数 kafka offset从哪读取可选first/last/curr
