spark定時提交job

本文內容：提交spark job到spark集羣，並使用crontab指定計劃任務，定時提交spark job。

任務提交

進入${SPARK_HOME}/bin目錄，有一個spark-submit的文件，就是經過腳本提交spark job的。

vim ./spark-submit 看看裏面都幹了啥。

if [ -z "${SPARK_HOME}" ]; then
  source "$(dirname "$0")"/find-spark-home
fi

# disable randomized hash for string in Python 3.3+
export PYTHONHASHSEED=0

exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"

首先找了一下環境變量，而後source了一下，這裏並無提交什麼東西，關鍵是最後一行，執行${SPARK_HOME}/bin目錄下的spark-class，帶了一個org.apache.spark.deploy.SparkSubmit參數，而且把參數傳給了spark-class，基本能夠判斷spark-submit並非真正提交job的地方。

vim ./spark-class

if [ -z "${SPARK_HOME}" ]; then
  source "$(dirname "$0")"/find-spark-home
fi

. "${SPARK_HOME}"/bin/load-spark-env.sh

# Find the java binary
if [ -n "${JAVA_HOME}" ]; then
  RUNNER="${JAVA_HOME}/bin/java"
...

# Find Spark jars.
if [ -d "${SPARK_HOME}/jars" ]; then
  SPARK_JARS_DIR="${SPARK_HOME}/jars"
else
  SPARK_JARS_DIR="${SPARK_HOME}/assembly/target/scala-$SPARK_SCALA_VERSION/jars"
fi
...

# Add the launcher build dir to the classpath if requested.
if [ -n "$SPARK_PREPEND_CLASSES" ]; then
  LAUNCH_CLASSPATH="${SPARK_HOME}/launcher/target/scala-$SPARK_SCALA_VERSION/classes:$LAUNCH_CLASSPATH"
fi


# command array and checks the value to see if the launcher succeeded.
build_command() {
  "$RUNNER" -Xmx128m -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "$@"
  printf "%d\0" $?
}

# Turn off posix mode since it does not allow process substitution
set +o posix
CMD=()
while IFS= read -d '' -r ARG; do
  CMD+=("$ARG")
done < <(build_command "$@")

...

CMD=("${CMD[@]:0:$LAST}")
exec "${CMD[@]}"

這裏只給出了部分源碼，代碼有點長，看註釋基本就能明白這裏作了什麼事：

1. 首先這裏又source了一下環境變量，可見在linux下環境變量的重要性，無時無刻都要注意環境變量。
2. 而後找到Java的環境變量。再而後去找spark中內置的jar包，這個目錄在${SPARK_HOME}/jars下，裏面包含了spark基本jar包，因此在項目打包的時候這裏面的jar包就不須要打進去了，生命週期設置爲provided。
3. 找scala的環境變量，構建一個LAUNCH_CLASSPATH。
4. 構建命令CMD，而且帶上原始輸入的參數。
5. 最後執行${CMD}。

最終是調用了org.apache.spark.launcher.Main這個類，而後把輸入的參數給了這個Main中的main(String[] argsArray)方法。繞了好大一圈仍是回到了Java，這裏的Main是Java的一個類，具體源碼自行去閱讀(真的有點複雜，由於這裏的java代碼又調用了scala...)。

提交參數說明

上問分析源碼的時候屢次看到了 $@ 這個字符，做用就是獲取全部的輸入參數，如今就來分析一下這些參數裏都有什麼。

./bin/spark-submit \
  --class <main-class> \
  --master <master-url> \
  --deploy-mode <deploy-mode> \
  --conf <key>=<value> \
  ... # other options
  <application-jar> \
  [application-arguments]

• --class 指出你須要運行的類，好比：org.apache.spark.examples.SparkPi
• --master 集羣主節點的URL，若是是提交到spark集羣，spark提供的web界面中有這個URL。而且推薦使用RESTURL。好比：spark://23.195.26.187:6066。 spark job的部署模式有多種：local[N] (N表示開啓的線程數) Spark Standalone、Mesos、Yarn、Kubernetes。
• --deploy-medo 這個參數決定了spark job的運行模式，集羣或者是本地（cluster or client）默認爲client。
• conf 自定義的一些配置，以鍵值對的形式添加。
• <application-jar> 對應spark job所在的jar包的位置。建議使用絕對路徑。
• [application-arguments] main()函數中的參數。

如下是spark官網給出的示例：

# Run application locally on 8 cores
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master local[8] \
  /path/to/examples.jar \
  100

# Run on a Spark standalone cluster in client deploy mode
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://207.184.161.138:7077 \
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

# Run on a Spark standalone cluster in cluster deploy mode with supervise
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://207.184.161.138:7077 \
  --deploy-mode cluster \
  --supervise \
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

# Run on a YARN cluster
export HADOOP_CONF_DIR=XXX
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn \
  --deploy-mode cluster \  # can be client for client mode
  --executor-memory 20G \
  --num-executors 50 \
  /path/to/examples.jar \
  1000

# Run a Python application on a Spark standalone cluster
./bin/spark-submit \
  --master spark://207.184.161.138:7077 \
  examples/src/main/python/pi.py \
  1000

# Run on a Mesos cluster in cluster deploy mode with supervise
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master mesos://207.184.161.138:7077 \
  --deploy-mode cluster \
  --supervise \
  --executor-memory 20G \
  --total-executor-cores 100 \
  http://path/to/examples.jar \
  1000

# Run on a Kubernetes cluster in cluster deploy mode
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master k8s://xx.yy.zz.ww:443 \
  --deploy-mode cluster \
  --executor-memory 20G \
  --num-executors 50 \
  http://path/to/examples.jar \
  1000

定時任務實現

http://www.javashuo.com/article/p-xrkmoeet-dg.html

這篇文章給出瞭如何指定計劃任務，這裏須要用到計劃任務。把上文中的提交job的shell寫到一個腳本中去，而後 crontab -e 新建一個計劃任務，而後設置時間定時提交job。

# 每兩個小時的第0分鐘執行一次
0 */2 * * * . /etc/profile;/bin/sh /usr/hdp/spark/tmp/submit-spark-job.sh

而後重啓crond服務：

systemctl restart crond.service
crontab -l # 查看當前計劃任務