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

Reviewed By: excelle08

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
@@ -454,10 +454,10 @@ the better, Spark benchmark also reports `queries_per_hour` which is 3600 divide
 by the execution time. `score` denotes the relative Sparkbench performance to
 DCPerf's baseline.
 For CPU performance analysis, it is
-also helpful to use `execution_time_test_93586-stage-2.0` because Stage 2.0
+also helpful to use `execution_time_test_93586-stage-4.0` because Stage 4.0
 is a compute intensive phase and is much less influenced by I/O. We expect the
 average CPU utilization during the entire benchmark to be around 55~75%. The
-CPU utilization during Stage 2.0 full batch period could reach nearly 100%.
+CPU utilization during Stage 4.0 period could reach nearly 100%.
 
 ```
 {
@@ -493,8 +493,7 @@ CPU utilization during Stage 2.0 full batch period could reach nearly 100%.
     "execution_time_test_93586": 288.3,
     "execution_time_test_93586-stage-0.0": 10.0,
     "execution_time_test_93586-stage-1.0": 67.0,
-    "execution_time_test_93586-stage-2.0": 205.0,
-    "execution_time_test_93586-stage-2.0-fullbatch": 181.0,
+    "execution_time_test_93586-stage-4.0": 205.0,
     "queries_per_hour": 12.4869927159,
     "score": 3.121748179,
     "worker_cores": 172,
@@ -506,7 +505,7 @@ CPU utilization during Stage 2.0 full batch period could reach nearly 100%.
 ```
 
 In the reported metrics, `execution_time_test_93586` is the overall execution time,
-`execution_time_test_93586-stage-2.0` is the execution time of Spark's
+`execution_time_test_93586-stage-4.0` is the execution time of Spark's
 compute-intensive phase.
 
 Spark benchmark will also put its runtime logs into `benchmark_metrics_<run_id>/work` folder.

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,25 @@ 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"),
+            )
+        log4j2_srcfile = joinpath(CONF_PATH, "log4j2.properties")
+        if args.real:
+            shutil.copy(
+                log4j2_srcfile, joinpath(SPARK_HOME, "conf", "log4j2.properties")
+            )
 
 
 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:

packages/spark_standalone/templates/proj_root/settings/log4j2.properties

@@ -0,0 +1,56 @@
+#
+# 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.
+#
+# Set everything to be logged to the console
+rootLogger.level = info
+rootLogger.appenderRef.stdout.ref = console
+# In the pattern layout configuration below, we specify an explicit `%ex` conversion
+# pattern for logging Throwables. If this was omitted, then (by default) Log4J would
+# implicitly add an `%xEx` conversion pattern which logs stacktraces with additional
+# class packaging information. That extra information can sometimes add a substantial
+# performance overhead, so we disable it in our default logging config.
+# For more information, see SPARK-39361.
+appender.console.type = Console
+appender.console.name = console
+appender.console.target = SYSTEM_ERR
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n%ex
+# Set the default spark-shell/spark-sql log level to WARN. When running the
+# spark-shell/spark-sql, the log level for these classes is used to overwrite
+# the root logger's log level, so that the user can have different defaults
+# for the shell and regular Spark apps.
+logger.repl.name = org.apache.spark.repl.Main
+logger.repl.level = info
+logger.thriftserver.name = org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver
+logger.thriftserver.level = info
+# Settings to quiet third party logs that are too verbose
+logger.jetty1.name = org.sparkproject.jetty
+logger.jetty1.level = warn
+logger.jetty2.name = org.sparkproject.jetty.util.component.AbstractLifeCycle
+logger.jetty2.level = error
+logger.replexprTyper.name = org.apache.spark.repl.SparkIMain$exprTyper
+logger.replexprTyper.level = info
+logger.replSparkILoopInterpreter.name = org.apache.spark.repl.SparkILoop$SparkILoopInterpreter
+logger.replSparkILoopInterpreter.level = info
+logger.parquet1.name = org.apache.parquet
+logger.parquet1.level = error
+logger.parquet2.name = parquet
+logger.parquet2.level = error
+# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
+logger.RetryingHMSHandler.name = org.apache.hadoop.hive.metastore.RetryingHMSHandler
+logger.RetryingHMSHandler.level = fatal
+logger.FunctionRegistry.name = org.apache.hadoop.hive.ql.exec.FunctionRegistry
+logger.FunctionRegistry.level = error