Updating Spark JVM to Graal VM and to 4.0-spark-hadoop (#210)

Summary:

Spark now runs with GraalVM Java 17 and Spark is updated to 4.0-hadoop

Differential Revision: D81242805

Author: ahmadelyoussef

packages/spark_standalone/README.md

@@ -148,7 +148,7 @@ to run your systems exclusively with IPv6. If your systems only support IPv4,
 we now provide a flag to enable support for it.
 
 * `JAVA_HOME`: You may need to manually set the environment variable `JAVA_HOME`
-to be the path of the JDK if Spark benchmark fails.
+to be the path of the JDK: /usr/lib/jvm/graalvm-jdk-17.0.12+8.1/ if Spark benchmark fails.
 
 ## Also Note
 
@@ -398,7 +398,7 @@ please remove the data from previous runs so that SparkBench can rebuild databas
 
 ```bash
 rm -rf /flash23/warehouse
-rm -rf <benchpressPath>/benchmarks/spark_standalone/spark-2.4.5-bin-hadoop2.7/metastore_db
+rm -rf <benchpressPath>/benchmarks/spark_standalone/spark-4.0.0-bin-hadoop3/metastore_db
 ```
 
 4. Create the `/flash23` folder. Copy `bpc_t93586_s2_synthetic_5GB`
@@ -408,7 +408,7 @@ Note that SparkBench mini does not require the high I/O throughput
 
 5. Run `spark_standalone_remote_mini` job on a real machine.
 This will create data in `/flash23/warehouse`
-and `<benchpressPath>/benchmarks/spark_standalone/spark-2.4.5-bin-hadoop2.7/metastore_db`.
+and `<benchpressPath>/benchmarks/spark_standalone/spark-4.0.0-bin-hadoop3/metastore_db`.
 Create a backup of these two folders. By default, this job uses the 5GB dataset.
 If you want to use the 1GB dataset, run the job with specifying the
 `dataset_name` parameter like this:
@@ -425,12 +425,12 @@ with the same commands and options.
 Building database takes a considerable amount of time, so it's advisable to consider
 using the same set of storage nodes and NVMe drives when running SparkBench on another
 compute node server. If you choose to reuse the Spark database in `/flash23/warehouse`,
-please also make sure to copy the folder `metastore_db` under `benchmarks/spark_standalone/spark-2.4.5-bin-hadoop2.7`
+please also make sure to copy the folder `metastore_db` under `benchmarks/spark_standalone/spark-4.0.0-bin-hadoop3`
 to the new machine's same location, for example:
 
 ```
 # Under the DCPerf folder
-rsync -a benchmarks/spark_standalone/spark-2.4.5-bin-hadoop2.7/metastore_db root@<target-hostname>:~/DCPerf/benchmarks/spark_standalone/spark-2.4.5-bin-hadoop2.7/
+rsync -a benchmarks/spark_standalone/spark-4.0.0-bin-hadoop3/metastore_db root@<target-hostname>:~/DCPerf/benchmarks/spark_standalone/spark-4.0.0-bin-hadoop3/
 ```
 
 If you do not copy over the `metastore_db` folder, you will see errors like the following

packages/spark_standalone/install_spark_standalone.sh

@@ -11,13 +11,25 @@ SPARK_PKG_ROOT="$(dirname "$(readlink -f "$0")")"
 TEMPLATES_DIR="${SPARK_PKG_ROOT}/templates"
 LINUX_DIST_ID="$(awk -F "=" '/^ID=/ {print $2}' /etc/os-release | tr -d '"')"
 
+# Detect architecture
+ARCH=$(uname -m)
+GRAALVM_ARCH="x64"
+if [ "$ARCH" = "aarch64" ]; then
+  GRAALVM_ARCH="aarch64"
+fi
+
 # Install system dependencies
 if [ "$LINUX_DIST_ID" = "ubuntu" ]; then
-  apt install -y openjdk-8-jdk fio
-  apt install -y git-lfs
+  apt install -y git-lfs fio
+  wget https://download.oracle.com/graalvm/17/archive/graalvm-jdk-17.0.12_linux-${GRAALVM_ARCH}_bin.tar.gz
+  mkdir -p /usr/lib/jvm/
+  tar -xzf graalvm-jdk-17.0.12_linux-${GRAALVM_ARCH}_bin.tar.gz -C /usr/lib/jvm/
+
 elif [ "$LINUX_DIST_ID" = "centos" ]; then
-  dnf install -y java-1.8.0-openjdk fio
-  dnf install -y git-lfs
+  dnf install -y git-lfs fio
+  wget https://download.oracle.com/graalvm/17/archive/graalvm-jdk-17.0.12_linux-${GRAALVM_ARCH}_bin.tar.gz
+  mkdir -p /usr/lib/jvm/
+  tar -xzf graalvm-jdk-17.0.12_linux-${GRAALVM_ARCH}_bin.tar.gz -C /usr/lib/jvm/
 fi
 
 # copy over directory
@@ -30,10 +42,10 @@ fi
 
 # download spark
 pushd "${OUT}" || exit 1
-if [ ! -f spark-2.4.5-bin-hadoop2.7.tgz ]; then
-  wget https://archive.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz
+if [ ! -f spark-4.0.0-bin-hadoop3.tgz ]; then
+  wget https://archive.apache.org/dist/spark/spark-4.0.0/spark-4.0.0-bin-hadoop3.tgz
 fi
-tar xzf spark-2.4.5-bin-hadoop2.7.tgz
+tar xzf spark-4.0.0-bin-hadoop3.tgz
 popd || exit 1
 
 # create sub directories

packages/spark_standalone/templates/proj_root/scripts/run_perf_common.py

@@ -55,7 +55,7 @@ def run_spark_sql(
     if database:
         cmd.extend(["--database", database])
     log_file = sql_file.replace(".sql", ".log")
-    env = {}
+    env = key_environ.copy()
     if node0_only:
         cmd = ["numactl", "--cpunodebind=0", "--membind=0"] + cmd
     run_cmd(cmd, SPARK_HOME, log_file, env, for_real)
@@ -248,7 +248,7 @@ def start(args) -> None:
     # driver
     cmd = ["sbin/start-master.sh"]
     log_file = joinpath(WORK_PATH, "start_master.log")
-    env = {}
+    env = key_environ.copy()
     write_spark_env(args, worker_idx=-1)
     if args.numa == "node0_only":
         cmd = ["numactl", "--cpunodebind=0", "--membind=0"] + cmd
@@ -301,27 +301,28 @@ def start(args) -> None:
     # shuffle server
     cmd = ["sbin/start-shuffle-service.sh"]
     log_file = joinpath(WORK_PATH, "start_shuffle_service.log")
-    env = {}
+    env = key_environ.copy()
     if args.numa == "node0_only":
         cmd = ["numactl", "--cpunodebind=0", "--membind=0"] + cmd
     run_cmd(cmd, SPARK_HOME, log_file, env, args.real)
 
 
 def stop(args) -> None:
     # workers
-    cmd = ["sbin/stop-slave.sh"]
-    log_file = joinpath(WORK_PATH, "stop_slave.log")
-    env = {"SPARK_WORKER_INSTANCES": str(args.num_workers)}
+    cmd = ["sbin/stop-worker.sh"]
+    log_file = joinpath(WORK_PATH, "stop-worker.log")
+    env = key_environ.copy()
+    env["SPARK_WORKER_INSTANCES"] = str(args.num_workers)
     run_cmd(cmd, SPARK_HOME, log_file, env, args.real, check=False)
     # shuffle server
     cmd = ["sbin/stop-shuffle-service.sh"]
     log_file = joinpath(WORK_PATH, "stop_shuffle_service.log")
-    env = {}
+    env = key_environ.copy()
     run_cmd(cmd, SPARK_HOME, log_file, env, args.real, check=False)
     # drvier
     cmd = ["sbin/stop-master.sh"]
     log_file = joinpath(WORK_PATH, "stop_master.log")
-    env = {}
+    env = key_environ.copy()
     run_cmd(cmd, SPARK_HOME, log_file, env, args.real, check=False)
 
 
@@ -405,6 +406,8 @@ def setup(args, init: bool = False) -> None:
     SPARK_CONFIGS = get_standalone_configs(platform)
     # add warehouse directory
     SPARK_CONFIGS["spark.sql.warehouse.dir"] = os.path.abspath(args.database_location)
+    # Add legacy assignment policy for Spark 4.0.0 compatibility
+    SPARK_CONFIGS["spark.sql.storeAssignmentPolicy"] = "LEGACY"
     if init:
         print(f"set database location at {os.path.abspath(args.database_location)}/")
     # update total core count
@@ -430,6 +433,20 @@ def setup(args, init: bool = False) -> None:
             shutil.copy(
                 start_slave_srcfile, joinpath(SPARK_HOME, "sbin", "start-slave-fb.sh")
             )
+        # copy custom shuffle service scripts
+        start_shuffle_srcfile = joinpath(
+            PROJ_ROOT, "scripts", "start-shuffle-service.sh"
+        )
+        stop_shuffle_srcfile = joinpath(PROJ_ROOT, "scripts", "stop-shuffle-service.sh")
+        if args.real:
+            shutil.copy(
+                start_shuffle_srcfile,
+                joinpath(SPARK_HOME, "sbin", "start-shuffle-service.sh"),
+            )
+            shutil.copy(
+                stop_shuffle_srcfile,
+                joinpath(SPARK_HOME, "sbin", "stop-shuffle-service.sh"),
+            )
 
 
 def init_parser():

packages/spark_standalone/templates/proj_root/scripts/start-shuffle-service.sh

@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+#
+
+# Starts the external shuffle server on the machine this script is executed on.
+#
+# Usage: start-shuffle-server.sh
+#
+# Use the SPARK_SHUFFLE_OPTS environment variable to set shuffle server configuration.
+#
+
+if [ -z "${SPARK_HOME}" ]; then
+  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"
+fi
+
+. "${SPARK_HOME}/sbin/spark-config.sh"
+. "${SPARK_HOME}/bin/load-spark-env.sh"
+
+exec "${SPARK_HOME}/sbin"/spark-daemon.sh start org.apache.spark.deploy.ExternalShuffleService 1

packages/spark_standalone/templates/proj_root/scripts/stop-shuffle-service.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+#
+
+# Stops the external shuffle service on the machine this script is executed on.
+
+if [ -z "${SPARK_HOME}" ]; then
+  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"
+fi
+
+"${SPARK_HOME}/sbin"/spark-daemon.sh stop org.apache.spark.deploy.ExternalShuffleService 1

packages/spark_standalone/templates/proj_root/scripts/utils.py

@@ -98,35 +98,8 @@ def read_sys_configs() -> Dict[str, int]:
 
 
 def find_java_home() -> str:
-    # Try finding a home path for java 8
-    candidates = [
-        "/usr/lib/jvm/java-1.8.0-openjdk",
-        "/usr/lib/jvm/java-1.8.0-jre",
-        "/usr/lib/jvm/java-8-openjdk",
-        "/usr/lib/jvm/java-8-jre",
-        "/usr/lib/jvm/openjdk-8",
-        "/usr/lib/jvm/jre-1.8.0",
-        "/usr/lib/jvm/jre-1.8.0-openjdk",
-    ]
-    archname = platform.machine()
-    if archname == "x86_64":
-        archname = "amd64"
-    elif archname == "aarch64":
-        archname = "arm64"
-    for path in candidates:
-        if os.path.exists(f"{path}/bin/java"):
-            return path
-        path_with_arch = f"{path}-{archname}"
-        if os.path.exists(f"{path_with_arch}/bin/java"):
-            return path_with_arch
-    # If none of the candidate exists, try find through `java` command
-    try:
-        java_path = subprocess.check_output(["which", "java"], text=True).strip()
-        java_home = str(pathlib.Path(os.path.realpath(java_path)).parents[1])
-    except subprocess.CalledProcessError:
-        java_home = ""
-
-    return java_home
+    # Always use the specific GraalVM JDK 17.0.12 path
+    return "/usr/lib/jvm/graalvm-jdk-17.0.12+8.1"
 
 
 def read_environ() -> Dict[str, str]:
@@ -135,7 +108,7 @@ def read_environ() -> Dict[str, str]:
     env_vars["PROJ_ROOT"] = "/".join(os.path.abspath(__file__).split("/")[:-2])
     env_vars["JAVA_HOME"] = find_java_home()
     env_vars["SPARK_HOME"] = os.path.join(
-        env_vars["PROJ_ROOT"], "spark-2.4.5-bin-hadoop2.7"
+        env_vars["PROJ_ROOT"], "spark-4.0.0-bin-hadoop3"
     )
     # read from actual environment
     for k in env_vars: