Add cluster support to v1

Author: rf972

benchmark/tpch/run_tpch.sh

@@ -7,7 +7,9 @@ if [ "$#" -lt 1 ]; then
   exit 1
 fi
 
+LOCAL=NO
 DEBUG=NO
+DEBUG_EXECUTOR=NO
 WORKERS=1
 POSITIONAL=()
 while [[ $# -gt 0 ]]
@@ -24,6 +26,14 @@ case $key in
     DEBUG=YES
     shift # past argument
     ;;
+    -de|--debug-exec)
+    DEBUG_EXECUTOR=YES
+    shift # past argument
+    ;;
+    -l|--local)
+    LOCAL=YES
+    shift # past argument
+    ;;
     *)    # unknown option
     POSITIONAL+=("$1") # save it in an array for later
     shift # past argument
@@ -34,37 +44,85 @@ set -- "${POSITIONAL[@]}" # restore positional parameters
 
 echo "DEBUG"  = "${DEBUG}"
 echo "WORKERS" = "${WORKERS}"
+echo "LOCAL" = "${LOCAL}"
+TEST=NO
+#set_speed $WORKERS
+DOCKER=sparkmaster
+DOCKER=sparklauncher
+# --conf "spark.submit.pyFiles=/build/dike.zip" \
 if [ ${DEBUG} == "YES" ]; then
   echo "Debugging"
-  docker exec -it sparkmaster spark-submit --master local \
+  docker exec -it ${DOCKER} spark-submit --master local \
   --class main.scala.TpchQuery \
   --conf "spark.jars.ivy=/build/ivy" \
   --conf "spark.driver.maxResultSize=20g" \
   --conf "spark.driver.memory=2g" \
   --conf "spark.executor.memory=2g" \
-  --conf "spark.driver.extraJavaOptions=-classpath /conf/:/build/spark-${SPARK_VERSION}/jars/*:/examples/scala/target/scala-2.12/ -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=172.18.0.3:5005" \
-  --packages com.github.scopt:scopt_2.12:4.0.0-RC2,ch.cern.sparkmeasure:spark-measure_2.12:0.17 \
+  --conf "spark.driver.extraJavaOptions=-classpath /conf/:/build/spark-${SPARK_VERSION}/jars/*: -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=10.124.48.63:5006" \
+  --packages com.github.luben:zstd-jni:1.5.0-4,org.json:json:20210307,javax.json:javax.json-api:1.1.4,org.glassfish:javax.json:1.1.4,com.github.scopt:scopt_2.12:4.0.0-RC2,ch.cern.sparkmeasure:spark-measure_2.12:0.17 \
   --jars /build/downloads/spark-sql-macros_2.12.10_0.1.0-SNAPSHOT.jar,/dikeHDFS/client/ndp-hdfs/target/ndp-hdfs-1.0.jar,/build/extra_jars/*,/pushdown-datasource/target/scala-2.12/pushdown-datasource_2.12-0.1.0.jar,/build/downloads/h2-1.4.200.jar \
   /tpch/tpch-spark/target/scala-2.12/spark-tpc-h-queries_2.12-1.0.jar $@ --workers ${WORKERS}
     #--packages com.github.scopt:scopt_2.12:4.0.0-RC2,com.amazonaws:aws-java-sdk:1.11.853,org.apache.hadoop:hadoop-aws:3.2.0,org.apache.commons:commons-csv:1.8 \
     #  --conf "spark.sql.parquet.enableVectorizedReader=false" \
-else
-  docker exec -it sparkmaster spark-submit --master local[$WORKERS] \
+elif [ ${DEBUG_EXECUTOR} == "YES" ]; then
+  echo "Debugging"
+  docker exec -it ${DOCKER} spark-submit --master local \
+  --class main.scala.TpchQuery \
+  --conf "spark.jars.ivy=/build/ivy" \
+  --conf "spark.driver.maxResultSize=20g" \
+  --conf "spark.driver.memory=2g" \
+  --conf "spark.executor.memory=2g" \
+  --conf "spark.driver.extraJavaOptions=-classpath /conf/:/build/spark-${SPARK_VERSION}/jars/*:/examples/scala/target/scala-2.12/ -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=172.169.1.10:5005" \
+  --packages com.github.luben:zstd-jni:1.5.0-4,org.json:json:20210307,javax.json:javax.json-api:1.1.4,org.glassfish:javax.json:1.1.4,com.github.scopt:scopt_2.12:4.0.0-RC2,ch.cern.sparkmeasure:spark-measure_2.12:0.17 \
+  --jars /build/downloads/spark-sql-macros_2.12.10_0.1.0-SNAPSHOT.jar,/dikeHDFS/client/ndp-hdfs/target/ndp-hdfs-1.0.jar,/build/extra_jars/*,/pushdown-datasource/target/scala-2.12/pushdown-datasource_2.12-0.1.0.jar,/build/downloads/h2-1.4.200.jar \
+  /tpch/tpch-spark/target/scala-2.12/spark-tpc-h-queries_2.12-1.0.jar $@ --workers ${WORKERS}
+elif [ ${LOCAL} == "YES" ]; then
+  echo "Local with $WORKERS workers."
+  docker exec -it ${DOCKER} spark-submit --master local[$WORKERS] \
   --conf "ivy.shared.default.root=/build/ivy_jars" \
-  --conf "spark.driver.extraClassPath=/build/extra_jars/*" \
-  --conf "spark.executor.extraClassPath=/build/extra_jars/*" \
   --conf "spark.jars.ivy=/build/ivy" \
   --conf "spark.driver.maxResultSize=20g" \
+  --conf "spark.task.maxDirectResultSize=20g" \
   --conf "spark.sql.broadcastTimeout=10000000" \
-  --conf "spark.driver.memory=32g" \
-  --conf "spark.executor.memory=32g" \
+  --conf "spark.driver.memory=2g" \
+  --conf "spark.executor.memory=2g" \
+  --conf "spark.dynamicAllocation.enabled=false" \
   --conf "spark.eventLog.enabled=true" \
   --conf "spark.eventLog.dir=/build/spark-events" \
   --conf "spark.driver.extraJavaOptions=-classpath /conf/:/build/spark-${SPARK_VERSION}/jars/*:/examples/scala/target/scala-2.12/" \
-  --packages com.github.scopt:scopt_2.12:4.0.0-RC2,ch.cern.sparkmeasure:spark-measure_2.12:0.17 \
-  --jars /build/downloads/spark-sql-macros_2.12.10_0.1.0-SNAPSHOT.jar,/dikeHDFS/client/ndp-hdfs/target/ndp-hdfs-1.0.jar,/build/extra_jars/*,/pushdown-datasource/target/scala-2.12/pushdown-datasource_2.12-0.1.0.jar,/build/downloads/h2-1.4.200.jar \
+  --packages com.github.luben:zstd-jni:1.5.0-4,org.json:json:20210307,javax.json:javax.json-api:1.1.4,org.glassfish:javax.json:1.1.4,com.github.scopt:scopt_2.12:4.0.0-RC2,ch.cern.sparkmeasure:spark-measure_2.12:0.17 \
+  --jars /dikeHDFS/client/ndp-hdfs/target/ndp-hdfs-1.0.jar,/pushdown-datasource/target/scala-2.12/pushdown-datasource_2.12-0.1.0.jar \
+  /tpch/tpch-spark/target/scala-2.12/spark-tpc-h-queries_2.12-1.0.jar $@ --workers ${WORKERS}
+elif [ ${TEST} != "YES" ]; then
+#local[$WORKERS]
+#spark://172.18.0.2:7077
+#  --conf "spark.executor.instances=1" \
+#  --conf "spark.executor.cores=1" \
+  HOST=sparkmaster
+  HOST=172.169.1.40
+  DRIVER_IP=172.169.1.40
+  docker exec -it ${DOCKER} spark-submit --total-executor-cores $WORKERS \
+              --master spark://$HOST:7077 \
+  --conf "ivy.shared.default.root=/build/ivy_jars" \
+  --conf "spark.jars.ivy=/build/ivy" \
+  --conf "spark.driver.maxResultSize=20g" \
+  --conf "spark.task.maxDirectResultSize=20g" \
+  --conf "spark.sql.broadcastTimeout=10000000" \
+  --conf "spark.driver.memory=2g" \
+  --conf "spark.executor.memory=2g" \
+  --conf "spark.dynamicAllocation.enabled=false" \
+  --conf "spark.eventLog.enabled=true" \
+  --conf "spark.eventLog.dir=/build/spark-events" \
+  --conf "spark.hadoop.dfs.client.use.datanode.hostname=true" \
+  --conf "spark.hadoop.dfs.namenode.rpc-address=172.169.1.60:9000" \
+  --conf "spark.driver.host=${DRIVER_IP}" \
+  --conf "spark.driver.bindAddress=${DRIVER_IP}" \
+  --conf "spark.driver.extraJavaOptions=-classpath /conf/:/build/spark-${SPARK_VERSION}/jars/*:/examples/scala/target/scala-2.12/" \
+  --packages com.github.luben:zstd-jni:1.5.0-4,org.json:json:20210307,javax.json:javax.json-api:1.1.4,org.glassfish:javax.json:1.1.4,com.github.scopt:scopt_2.12:4.0.0-RC2,ch.cern.sparkmeasure:spark-measure_2.12:0.17 \
+  --jars /dikeHDFS/client/ndp-hdfs/target/ndp-hdfs-1.0.jar,/pushdown-datasource/target/scala-2.12/pushdown-datasource_2.12-0.1.0.jar \
   /tpch/tpch-spark/target/scala-2.12/spark-tpc-h-queries_2.12-1.0.jar $@ --workers ${WORKERS}
 fi
+# --jars /dikeHDFS/client/ndp-hdfs/target/ndp-hdfs-1.0.jar,/pushdown-datasource/target/scala-2.12/pushdown-datasource_2.12-0.1.0.jar \
 
 #,org.dike.hdfs:ndp-hdfs:1.0 /dikeHDFS/client/ndp-hdfs/target/ndp-hdfs-1.0-jar-with-dependencies.jar,
 #--repositories file:/build/dike \

spark/bin/start-master.sh

@@ -1,11 +1,18 @@
 #!/bin/bash
-bin/spark-class org.apache.spark.deploy.master.Master > /opt/volume/logs/master.log 2>&1 &
 
+
+if [[ ! -z "${SPARK_MASTER_HOST}" ]]; then
+  echo "master ip is: $SPARK_MASTER_HOST"
+  ./sbin/start-master.sh --ip $SPARK_MASTER_HOST --port 7077 > /opt/volume/logs/master.log 2>&1 &
+else
+  bin/spark-class org.apache.spark.deploy.master.Master > /opt/volume/logs/master.log 2>&1 &
+fi
+# --properties-file /conf/spark-defaults.conf
 echo "SPARK_MASTER_READY"
 echo "SPARK_MASTER_READY" > /opt/volume/status/SPARK_MASTER_STATE
 
 echo "RUNNING_MODE $RUNNING_MODE"
 
 if [ "$RUNNING_MODE" = "daemon" ]; then
     sleep infinity
-fi
+fi

spark/bin/start-worker.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
-bin/spark-class org.apache.spark.deploy.worker.Worker spark://sparkmaster:7077 > /opt/volume/logs/worker.log 2>&1 &
-
+./sbin/start-worker.sh spark://sparkmaster:7077 > /opt/volume/logs/worker.log 2>&1 &
+# --properties-file /conf/spark-defaults.conf
 echo "SPARK_WORKER_READY"
 echo "SPARK_WORKER_READY" > /opt/volume/status/SPARK_WORKER_STATE
 

spark/docker/Dockerfile

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM ubuntu:20.04 as builder_base
+FROM ubuntu:20.04 as v1_builder_base
 LABEL version="1.0"
 LABEL maintainer="Robert Foley <robfoley972@gmail.com>"
 
@@ -82,13 +82,13 @@ WORKDIR $SPARK_HOME
 ####
 #### Spark build docker stage.
 ####
-FROM builder_base as builder
+FROM v1_builder_base as v1_builder
 LABEL version="1.0"
 
 ####
 #### Spark run docker stage.
 ####
-FROM builder_base as run
+FROM v1_builder_base as v1_run
 LABEL version="1.0"
 
 COPY entry.sh /usr/local/bin/entry.sh

spark/docker/build.sh

@@ -16,7 +16,7 @@
 # limitations under the License.
 
 ROOT_DIR=$(pwd)
-
+source setup.sh
 DOCKER_DIR=${ROOT_DIR}
 DOCKER_FILE="${DOCKER_DIR}/Dockerfile"
 
@@ -61,17 +61,17 @@ fi
 echo "User id is: $USER_ID"
 echo "Group id is: $GROUP_ID"
 
-docker build -f Dockerfile --target builder -t spark_build .
-echo "Done building spark_build docker"
+docker build -f Dockerfile --target v${DIKE_VERSION}_builder -t v${DIKE_VERSION}_spark_build .
+echo "Done building v${DIKE_VERSION}_spark_build docker"
 
-docker build -f Dockerfile -t spark_run .
-echo "Done building spark_run docker"
+docker build -f Dockerfile -t v${DIKE_VERSION}_spark_run .
+echo "Done building v${DIKE_VERSION}_spark_run docker"
 
 # Set the home directory in the Docker container.
 DOCKER_HOME_DIR=${DOCKER_HOME_DIR:-/home/${USER_NAME}}
 
-docker build -t "spark-build-${USER_NAME}" - <<UserSpecificDocker
-FROM spark_build
+docker build -t "v${DIKE_VERSION}-spark-build-${USER_NAME}" - <<UserSpecificDocker
+FROM v${DIKE_VERSION}_spark_build
 RUN rm -f /var/log/faillog /var/log/lastlog
 RUN groupadd --non-unique -g ${GROUP_ID} ${USER_NAME}
 RUN useradd -g ${GROUP_ID} -u ${USER_ID} -k /root -m ${USER_NAME} -d "${DOCKER_HOME_DIR}"
@@ -86,10 +86,10 @@ RUN chmod 0600 ~/.ssh/authorized_keys
 EXPOSE 22
 
 UserSpecificDocker
-echo "Done building spark-build-${USER_NAME}"
+echo "Done building v${DIKE_VERSION}-spark-build-${USER_NAME}"
 
-docker build -t "spark-run-${USER_NAME}" - <<UserSpecificDocker
-FROM spark_run
+docker build -t "v${DIKE_VERSION}-spark-run-${USER_NAME}" - <<UserSpecificDocker
+FROM v${DIKE_VERSION}_spark_run
 RUN rm -f /var/log/faillog /var/log/lastlog
 RUN groupadd --non-unique -g ${GROUP_ID} ${USER_NAME}
 RUN useradd -g ${GROUP_ID} -u ${USER_ID} -k /root -m ${USER_NAME} -d "${DOCKER_HOME_DIR}"
@@ -104,6 +104,6 @@ RUN chmod 0600 ~/.ssh/authorized_keys
 EXPOSE 22
 
 UserSpecificDocker
-echo "Done building spark-run-${USER_NAME}"
+echo "Done building v${DIKE_VERSION}-spark-run-${USER_NAME}"
 
 echo "Done building dockers"

spark/docker/setup.sh

@@ -14,7 +14,8 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
+DIKE_VERSION=1
+echo "VERSION: ${DIKE_VERSION}"
 ROOT_DIR=$(pwd)
 DOCKER_DIR=docker
 DOCKER_FILE="${DOCKER_DIR}/Dockerfile"

spark/docker/start-launcher-host.sh

@@ -0,0 +1,74 @@
+#!/bin/bash
+
+# Include the setup for our cached local directories. (.m2, .ivy2, etc)
+source docker/setup.sh
+
+mkdir -p "${ROOT_DIR}/volume/logs"
+rm -f "${ROOT_DIR}/volume/logs/master*.log"
+
+mkdir -p "${ROOT_DIR}/volume/status"
+rm -f "${ROOT_DIR}/volume/status/MASTER*"
+
+CMD="sleep 365d"
+RUNNING_MODE="daemon"
+
+DOCKER_HOSTS="$(cat spark.config | grep DOCKER_HOSTS)"
+IFS='=' read -a IP_ARRAY <<< "$DOCKER_HOSTS"
+DOCKER_HOSTS=${IP_ARRAY[1]}
+HOSTS=""
+IFS=',' read -a IP_ARRAY <<< "$DOCKER_HOSTS"
+for i in "${IP_ARRAY[@]}"
+do
+  HOSTS="$HOSTS --add-host=$i"
+done
+DOCKER_HOSTS=$HOSTS
+echo "Docker Hosts: $DOCKER_HOSTS"
+
+LAUNCHER_IP="$(cat spark.config | grep LAUNCHER_IP)"
+IFS='=' read -a IP_ARRAY <<< "$LAUNCHER_IP"
+LAUNCHER_IP=${IP_ARRAY[1]}
+echo "LAUNCHER_IP: $LAUNCHER_IP"
+
+if [ $RUNNING_MODE = "interactive" ]; then
+  DOCKER_IT="-i -t"
+fi
+#  --cpuset-cpus="9-12" \
+DOCKER_RUN="docker run ${DOCKER_IT} --rm \
+  -p 5006:5006 \
+  --name sparklauncher \
+  --network host ${DOCKER_HOSTS} \
+  -e MASTER=spark://sparkmaster:7077 \
+  -e SPARK_CONF_DIR=/conf \
+  -e SPARK_PUBLIC_DNS=localhost \
+  -e SPARK_MASTER="spark://sparkmaster:7077" \
+  -e SPARK_DRIVER_HOST=${LAUNCHER_IP} \
+  --mount type=bind,source=$(pwd)/spark,target=/spark \
+  --mount type=bind,source=$(pwd)/build,target=/build \
+  --mount type=bind,source=$(pwd)/examples,target=/examples \
+  --mount type=bind,source=$(pwd)/../dikeHDFS,target=/dikeHDFS \
+  --mount type=bind,source=$(pwd)/../benchmark/tpch,target=/tpch \
+  --mount type=bind,source=$(pwd)/../data,target=/tpch-data \
+  --mount type=bind,source=$(pwd)/../pushdown-datasource/pushdown-datasource,target=/pushdown-datasource \
+  -v $(pwd)/conf/master:/conf  \
+  -v ${ROOT_DIR}/build/.m2:${DOCKER_HOME_DIR}/.m2 \
+  -v ${ROOT_DIR}/build/.gnupg:${DOCKER_HOME_DIR}/.gnupg \
+  -v ${ROOT_DIR}/build/.sbt:${DOCKER_HOME_DIR}/.sbt \
+  -v ${ROOT_DIR}/build/.cache:${DOCKER_HOME_DIR}/.cache \
+  -v ${ROOT_DIR}/build/.ivy2:${DOCKER_HOME_DIR}/.ivy2 \
+  -v ${ROOT_DIR}/volume/status:/opt/volume/status \
+  -v ${ROOT_DIR}/volume/logs:/opt/volume/logs \
+  -v ${ROOT_DIR}/bin/:${DOCKER_HOME_DIR}/bin \
+  -e RUNNING_MODE=${RUNNING_MODE} \
+  -u ${USER_ID} \
+  v${DIKE_VERSION}-spark-run-${USER_NAME} ${CMD}"
+
+if [ $RUNNING_MODE = "interactive" ]; then
+  eval "${DOCKER_RUN}"
+else
+  eval "${DOCKER_RUN}" &
+  while [ ! -f "${ROOT_DIR}/volume/status/SPARK_MASTER_STATE" ]; do
+    sleep 1
+  done
+
+  cat "${ROOT_DIR}/volume/status/SPARK_MASTER_STATE"
+fi