diff --git a/.gitignore b/.gitignore index 00320283df..323bf20ab6 100644 --- a/.gitignore +++ b/.gitignore @@ -29,3 +29,9 @@ hits.vortex # Python venvs created by install scripts myenv +__pycache__/ + +# spark-gluten-clickhouse: from-source build tree and symlinked artifacts +gluten/ +gluten.jar +libch.so diff --git a/spark-gluten-clickhouse/README.md b/spark-gluten-clickhouse/README.md new file mode 100644 index 0000000000..013bc7bdad --- /dev/null +++ b/spark-gluten-clickhouse/README.md @@ -0,0 +1,34 @@ +This entry runs Apache Spark with the [Apache Gluten](https://gluten.apache.org/) plugin configured to use the **ClickHouse backend** ('ch'). Gluten loads `libch.so` (a fork of ClickHouse v23.1) into the Spark executor JVM and runs the columnar physical plan natively through it. See also [`spark-gluten/`](../spark-gluten/) (Velox backend), [`spark-velox/`](../spark-velox/), and the [accelerators README](../spark/README-accelerators.md). + +### Run + +`./benchmark.sh` sets a few env vars and delegates to the shared driver +[`../lib/benchmark-common.sh`](../lib/benchmark-common.sh), which runs the +per-system scripts (`install`, `load`, `query`, ...) and prints the results in +the format collected by play.clickhouse.com. `./install` builds everything from +source (no pre-built bundle is published for the CH backend). + +## Notes + +### Build + +The CH backend is not part of Apache Gluten's release tarball — only the Velox bundle is published. As a result `install` builds two things from source: + +1. **`libch.so`** — built from [Kyligence/ClickHouse](https://github.com/Kyligence/ClickHouse) at the org/branch/commit pinned in `gluten/cpp-ch/clickhouse.version`. The build uses Clang 19 / cmake / ninja (Gluten v1.4.0's CH backend requires Clang 19). Its `extern-local-engine` module links against JNI **including AWT**, so `install` uses the full `openjdk-17-jdk` (not `-headless`, which omits `jawt.h`/`libjawt.so` and makes cmake fail with `Could NOT find JNI (missing: AWT)`). +2. **The Gluten Spark plugin** — built via Maven with `-Pbackends-clickhouse -Pspark-3.5 -Pscala-2.12 -Pdelta`. The `delta` profile is required by the `backends-clickhouse` module's enforcer; `spark-3.5` pins the matching Delta version (3.2.0). JDK 8 is required at compile time (Gluten's POM); Spark itself runs under JDK 17 (see `./query`). + +Building libch.so essentially compiles ClickHouse from source: it is **memory-hungry** (Gluten's docs note that 64 GB RAM is recommended). On a c6a.4xlarge (32 GB RAM) the compile may OOM; use c6a.8xlarge or larger for a clean run. + +### Configuration + +- `spark.gluten.sql.columnar.backend.lib=ch` selects the ClickHouse backend over Velox. +- `spark.gluten.sql.columnar.libpath=` points to the native library. Gluten's wrapper cmakes into `gluten/cpp-ch/build_ch`, which drives an inner ClickHouse cmake that builds `libch.so` under `gluten/cpp-ch/build/.../extern-local-engine/`; `install` globs for it under `cpp-ch/` and symlinks it as `libch.so` in the entry directory. +- Memory is split 50/50 between Spark heap and Gluten off-heap, identical to the Velox entry — the CH backend also runs off-heap via JNI. +- `libch.so` is preloaded into the JVM via `LD_PRELOAD` (set in `query.py`). Because the library carries initial-exec-model TLS, a lazy `dlopen` from the running JVM otherwise fails with `cannot allocate memory in static TLS block`; preloading it at JVM startup — while the static TLS block still has room — avoids this. Gluten's docs use `spark.executorEnv.LD_PRELOAD` for this, but in `local[*]` mode the driver JVM is the executor and launches before that config is read, so the preload is done through the JVM's environment instead. +- Queries use ClickHouse-style regex backreferences (`\1`) rather than Spark's `$1`, since the regex evaluation happens inside libch.so. See the discussion in [`spark-gluten/README.md`](../spark-gluten/README.md) and [Gluten issue #7545](https://github.com/apache/incubator-gluten/issues/7545). + +### Links + +- [Gluten ClickHouse-backend getting started](https://gluten.apache.org/docs/get-started/ClickHouse/). +- [Gluten release page](https://gluten.apache.org/downloads/) (Velox bundles only). +- [Kyligence/ClickHouse fork](https://github.com/Kyligence/ClickHouse) (the source of libch.so). diff --git a/spark-gluten-clickhouse/benchmark.sh b/spark-gluten-clickhouse/benchmark.sh new file mode 100755 index 0000000000..fb3b4d1318 --- /dev/null +++ b/spark-gluten-clickhouse/benchmark.sh @@ -0,0 +1,9 @@ +#!/bin/bash +export BENCH_DOWNLOAD_SCRIPT="download-hits-parquet-single" +export BENCH_RESTARTABLE=no +# Single-process engine: each query forks a fresh full-machine process with no +# shared scheduler across connections, so the concurrent-QPS test only +# oversubscribes RAM rather than measuring throughput. Skip it by default; +# override BENCH_CONCURRENT_DURATION to re-enable. See issue #946. +export BENCH_CONCURRENT_DURATION="${BENCH_CONCURRENT_DURATION:-0}" +exec ../lib/benchmark-common.sh diff --git a/spark-gluten-clickhouse/check b/spark-gluten-clickhouse/check new file mode 100755 index 0000000000..45d397cda2 --- /dev/null +++ b/spark-gluten-clickhouse/check @@ -0,0 +1,8 @@ +#!/bin/bash +set -e + +# shellcheck disable=SC1091 +source myenv/bin/activate +python3 -c 'import pyspark' >/dev/null 2>&1 +[ -f gluten.jar ] +[ -f libch.so ] diff --git a/spark-gluten-clickhouse/data-size b/spark-gluten-clickhouse/data-size new file mode 100755 index 0000000000..1a34600a86 --- /dev/null +++ b/spark-gluten-clickhouse/data-size @@ -0,0 +1,4 @@ +#!/bin/bash +set -e + +du -b hits.parquet | cut -f1 diff --git a/spark-gluten-clickhouse/install b/spark-gluten-clickhouse/install new file mode 100755 index 0000000000..c0525822ed --- /dev/null +++ b/spark-gluten-clickhouse/install @@ -0,0 +1,116 @@ +#!/bin/bash +set -e + +# Spark + Apache Gluten with the ClickHouse backend ('ch'). Unlike the Velox +# backend, no pre-built bundle is published for the CH backend, so we build +# both libch.so (a ClickHouse fork) and the Gluten Spark plugin from source. +# +# The libch.so compile is essentially a ClickHouse build: RAM-hungry (Gluten's +# docs recommend >= 64 GB) and slow. Run on c6a.8xlarge or larger; c6a.4xlarge +# (32 GB) may OOM. +# +# Note: Keep in sync with spark-*/install (see README-accelerators.md). + +GLUTEN_VERSION=v1.4.0 +SPARK_PROFILE=spark-3.5 + +# Build prerequisites: +# - Java 8 (headless) to build Gluten via Maven (Gluten's pom requires JDK 8). +# - Java 17, the FULL jdk (not -headless), to run Spark AND to satisfy the +# libch.so build's JNI/AWT dependency: ClickHouse's extern-local-engine +# links against jawt.h/libjawt.so, which the -headless packages omit, so +# its cmake fails with `Could NOT find JNI (missing: AWT)` / +# `JAVA_AWT_INCLUDE_PATH-NOTFOUND`. openjdk-17-jdk ships jawt.h and pulls +# in the non-headless JRE that ships libjawt.so. +# - Clang 19, cmake, ninja, etc. to build libch.so (Gluten v1.4.0's CH +# backend requires Clang 19; see get-started/ClickHouse.md). +sudo apt-get update -y +sudo apt-get install -y python3-pip python3-venv \ + openjdk-8-jdk-headless openjdk-17-jdk \ + maven git cmake ccache ninja-build nasm yasm gawk \ + lsb-release wget software-properties-common gnupg + +# Install Clang 19 via apt.llvm.org. +if ! command -v clang-19 >/dev/null 2>&1; then + wget -O - https://apt.llvm.org/llvm.sh | sudo bash -s -- 19 +fi + +export CC=clang-19 +export CXX=clang++-19 + +ARCH=$(dpkg --print-architecture) + +# pyspark venv. +if [ ! -d myenv ]; then + python3 -m venv myenv +fi +# shellcheck disable=SC1091 +source myenv/bin/activate +pip install -q pyspark==3.5.2 psutil + +# Clone Gluten and the Kyligence ClickHouse fork that the CH backend wraps. +# The CH fork org/branch/commit are pinned in Gluten's cpp-ch/clickhouse.version. +GLUTEN_DIR="$PWD/gluten" +if [ ! -d "$GLUTEN_DIR" ]; then + git clone --depth 1 --branch "$GLUTEN_VERSION" \ + https://github.com/apache/incubator-gluten.git "$GLUTEN_DIR" +fi + +CH_VERSION_FILE="$GLUTEN_DIR/cpp-ch/clickhouse.version" +CH_ORG=$(grep '^CH_ORG=' "$CH_VERSION_FILE" | cut -d= -f2) +CH_BRANCH=$(grep '^CH_BRANCH=' "$CH_VERSION_FILE" | cut -d= -f2) +CH_DIR="$GLUTEN_DIR/cpp-ch/ClickHouse" +if [ ! -d "$CH_DIR" ]; then + git clone --recursive --shallow-submodules \ + --branch "$CH_BRANCH" \ + "https://github.com/${CH_ORG}/ClickHouse.git" "$CH_DIR" +fi + +# Build libch.so via Gluten's own wrapper. It cmakes cpp-ch into +# cpp-ch/build_ch, whose build_ch target drives an inner ClickHouse cmake that +# builds `--target libch` into cpp-ch/build/, so the artifact lands under +# cpp-ch/build/.../extern-local-engine/. Glob for it rather than hardcoding — +# the two build dirs (build_ch wrapper vs build inner) are easy to confuse. +# Honors the CC/CXX exported above; JAVA_HOME points cmake's FindJNI at the +# full JDK 17 so it locates jawt.h/libjawt.so (see the JDK note above). +export JAVA_HOME="/usr/lib/jvm/java-17-openjdk-${ARCH}" +LIBCH_SO=$(find "$GLUTEN_DIR/cpp-ch" -name libch.so -type f 2>/dev/null | head -n1) +if [ -z "$LIBCH_SO" ]; then + bash "$GLUTEN_DIR/ep/build-clickhouse/src/build_clickhouse.sh" + LIBCH_SO=$(find "$GLUTEN_DIR/cpp-ch" -name libch.so -type f 2>/dev/null | head -n1) +fi +if [ -z "$LIBCH_SO" ]; then + echo "ERROR: libch.so not found after build" >&2 + exit 1 +fi + +# Build the Gluten Spark plugin against the CH backend. JDK 8 is required at +# compile time per Gluten's pom; Spark itself runs under JDK 17 in ./query. +# pyspark wheels ship Scala 2.12 jars, so build with scala-2.12 to match. +# The delta profile is mandatory: backends-clickhouse's pom carries an +# `enforce-delta-profile` enforcer rule that fails the build with +# `"-P delta" must be set when building Gluten with ClickHouse backend` unless +# it is active. The spark-3.5 profile pins the matching delta.version (3.2.0), +# so -Pdelta needs no extra version flag. +JAVA_HOME_8="/usr/lib/jvm/java-8-openjdk-${ARCH}" +GLUTEN_JAR=$(ls "$GLUTEN_DIR"/backends-clickhouse/target/gluten-*-spark-3.5-jar-with-dependencies.jar 2>/dev/null | head -n1) +if [ -z "$GLUTEN_JAR" ]; then + ( + cd "$GLUTEN_DIR" + export MAVEN_OPTS="-Xmx8g -XX:ReservedCodeCacheSize=2g" + JAVA_HOME="$JAVA_HOME_8" PATH="$JAVA_HOME_8/bin:$PATH" \ + mvn -B clean package \ + -Pbackends-clickhouse -P"$SPARK_PROFILE" -Pscala-2.12 -Pdelta \ + -DskipTests -Dcheckstyle.skip + ) + GLUTEN_JAR=$(ls "$GLUTEN_DIR"/backends-clickhouse/target/gluten-*-spark-3.5-jar-with-dependencies.jar 2>/dev/null | head -n1) +fi +if [ -z "$GLUTEN_JAR" ]; then + echo "ERROR: could not locate built Gluten CH-backend jar" >&2 + ls "$GLUTEN_DIR/backends-clickhouse/target/" >&2 || true + exit 1 +fi + +# query.py expects the jar and native library as ./gluten.jar and ./libch.so. +ln -sf "$GLUTEN_JAR" gluten.jar +ln -sf "$LIBCH_SO" libch.so diff --git a/spark-gluten-clickhouse/load b/spark-gluten-clickhouse/load new file mode 100755 index 0000000000..27b3422e88 --- /dev/null +++ b/spark-gluten-clickhouse/load @@ -0,0 +1,6 @@ +#!/bin/bash +set -e + +# Nothing to load: query.py reads hits.parquet directly. Just flush writeback +# so the parquet the harness downloaded is durably on disk before queries run. +sync diff --git a/spark-gluten-clickhouse/queries.sql b/spark-gluten-clickhouse/queries.sql new file mode 100644 index 0000000000..31f65fc898 --- /dev/null +++ b/spark-gluten-clickhouse/queries.sql @@ -0,0 +1,43 @@ +SELECT COUNT(*) FROM hits; +SELECT COUNT(*) FROM hits WHERE AdvEngineID <> 0; +SELECT SUM(AdvEngineID), COUNT(*), AVG(ResolutionWidth) FROM hits; +SELECT AVG(UserID) FROM hits; +SELECT COUNT(DISTINCT UserID) FROM hits; +SELECT COUNT(DISTINCT SearchPhrase) FROM hits; +SELECT MIN(EventDate), MAX(EventDate) FROM hits; +SELECT AdvEngineID, COUNT(*) FROM hits WHERE AdvEngineID <> 0 GROUP BY AdvEngineID ORDER BY COUNT(*) DESC; +SELECT RegionID, COUNT(DISTINCT UserID) AS u FROM hits GROUP BY RegionID ORDER BY u DESC LIMIT 10; +SELECT RegionID, SUM(AdvEngineID), COUNT(*) AS c, AVG(ResolutionWidth), COUNT(DISTINCT UserID) FROM hits GROUP BY RegionID ORDER BY c DESC LIMIT 10; +SELECT MobilePhoneModel, COUNT(DISTINCT UserID) AS u FROM hits WHERE MobilePhoneModel <> '' GROUP BY MobilePhoneModel ORDER BY u DESC LIMIT 10; +SELECT MobilePhone, MobilePhoneModel, COUNT(DISTINCT UserID) AS u FROM hits WHERE MobilePhoneModel <> '' GROUP BY MobilePhone, MobilePhoneModel ORDER BY u DESC LIMIT 10; +SELECT SearchPhrase, COUNT(*) AS c FROM hits WHERE SearchPhrase <> '' GROUP BY SearchPhrase ORDER BY c DESC LIMIT 10; +SELECT SearchPhrase, COUNT(DISTINCT UserID) AS u FROM hits WHERE SearchPhrase <> '' GROUP BY SearchPhrase ORDER BY u DESC LIMIT 10; +SELECT SearchEngineID, SearchPhrase, COUNT(*) AS c FROM hits WHERE SearchPhrase <> '' GROUP BY SearchEngineID, SearchPhrase ORDER BY c DESC LIMIT 10; +SELECT UserID, COUNT(*) FROM hits GROUP BY UserID ORDER BY COUNT(*) DESC LIMIT 10; +SELECT UserID, SearchPhrase, COUNT(*) FROM hits GROUP BY UserID, SearchPhrase ORDER BY COUNT(*) DESC LIMIT 10; +SELECT UserID, SearchPhrase, COUNT(*) FROM hits GROUP BY UserID, SearchPhrase LIMIT 10; +SELECT UserID, extract(minute FROM EventTime) AS m, SearchPhrase, COUNT(*) FROM hits GROUP BY UserID, m, SearchPhrase ORDER BY COUNT(*) DESC LIMIT 10; +SELECT UserID FROM hits WHERE UserID = 435090932899640449; +SELECT COUNT(*) FROM hits WHERE URL LIKE '%google%'; +SELECT SearchPhrase, MIN(URL), COUNT(*) AS c FROM hits WHERE URL LIKE '%google%' AND SearchPhrase <> '' GROUP BY SearchPhrase ORDER BY c DESC LIMIT 10; +SELECT SearchPhrase, MIN(URL), MIN(Title), COUNT(*) AS c, COUNT(DISTINCT UserID) FROM hits WHERE Title LIKE '%Google%' AND URL NOT LIKE '%.google.%' AND SearchPhrase <> '' GROUP BY SearchPhrase ORDER BY c DESC LIMIT 10; +SELECT * FROM hits WHERE URL LIKE '%google%' ORDER BY EventTime LIMIT 10; +SELECT SearchPhrase FROM hits WHERE SearchPhrase <> '' ORDER BY EventTime LIMIT 10; +SELECT SearchPhrase FROM hits WHERE SearchPhrase <> '' ORDER BY SearchPhrase LIMIT 10; +SELECT SearchPhrase FROM hits WHERE SearchPhrase <> '' ORDER BY EventTime, SearchPhrase LIMIT 10; +SELECT CounterID, AVG(length(URL)) AS l, COUNT(*) AS c FROM hits WHERE URL <> '' GROUP BY CounterID HAVING COUNT(*) > 100000 ORDER BY l DESC LIMIT 25; +SELECT REGEXP_REPLACE(Referer, '^https?://(?:www\.)?([^/]+)/.*$', '\1') AS k, AVG(length(Referer)) AS l, COUNT(*) AS c, MIN(Referer) FROM hits WHERE Referer <> '' GROUP BY k HAVING COUNT(*) > 100000 ORDER BY l DESC LIMIT 25; +SELECT SUM(ResolutionWidth), SUM(ResolutionWidth + 1), SUM(ResolutionWidth + 2), SUM(ResolutionWidth + 3), SUM(ResolutionWidth + 4), SUM(ResolutionWidth + 5), SUM(ResolutionWidth + 6), SUM(ResolutionWidth + 7), SUM(ResolutionWidth + 8), SUM(ResolutionWidth + 9), SUM(ResolutionWidth + 10), SUM(ResolutionWidth + 11), SUM(ResolutionWidth + 12), SUM(ResolutionWidth + 13), SUM(ResolutionWidth + 14), SUM(ResolutionWidth + 15), SUM(ResolutionWidth + 16), SUM(ResolutionWidth + 17), SUM(ResolutionWidth + 18), SUM(ResolutionWidth + 19), SUM(ResolutionWidth + 20), SUM(ResolutionWidth + 21), SUM(ResolutionWidth + 22), SUM(ResolutionWidth + 23), SUM(ResolutionWidth + 24), SUM(ResolutionWidth + 25), SUM(ResolutionWidth + 26), SUM(ResolutionWidth + 27), SUM(ResolutionWidth + 28), SUM(ResolutionWidth + 29), SUM(ResolutionWidth + 30), SUM(ResolutionWidth + 31), SUM(ResolutionWidth + 32), SUM(ResolutionWidth + 33), SUM(ResolutionWidth + 34), SUM(ResolutionWidth + 35), SUM(ResolutionWidth + 36), SUM(ResolutionWidth + 37), SUM(ResolutionWidth + 38), SUM(ResolutionWidth + 39), SUM(ResolutionWidth + 40), SUM(ResolutionWidth + 41), SUM(ResolutionWidth + 42), SUM(ResolutionWidth + 43), SUM(ResolutionWidth + 44), SUM(ResolutionWidth + 45), SUM(ResolutionWidth + 46), SUM(ResolutionWidth + 47), SUM(ResolutionWidth + 48), SUM(ResolutionWidth + 49), SUM(ResolutionWidth + 50), SUM(ResolutionWidth + 51), SUM(ResolutionWidth + 52), SUM(ResolutionWidth + 53), SUM(ResolutionWidth + 54), SUM(ResolutionWidth + 55), SUM(ResolutionWidth + 56), SUM(ResolutionWidth + 57), SUM(ResolutionWidth + 58), SUM(ResolutionWidth + 59), SUM(ResolutionWidth + 60), SUM(ResolutionWidth + 61), SUM(ResolutionWidth + 62), SUM(ResolutionWidth + 63), SUM(ResolutionWidth + 64), SUM(ResolutionWidth + 65), SUM(ResolutionWidth + 66), SUM(ResolutionWidth + 67), SUM(ResolutionWidth + 68), SUM(ResolutionWidth + 69), SUM(ResolutionWidth + 70), SUM(ResolutionWidth + 71), SUM(ResolutionWidth + 72), SUM(ResolutionWidth + 73), SUM(ResolutionWidth + 74), SUM(ResolutionWidth + 75), SUM(ResolutionWidth + 76), SUM(ResolutionWidth + 77), SUM(ResolutionWidth + 78), SUM(ResolutionWidth + 79), SUM(ResolutionWidth + 80), SUM(ResolutionWidth + 81), SUM(ResolutionWidth + 82), SUM(ResolutionWidth + 83), SUM(ResolutionWidth + 84), SUM(ResolutionWidth + 85), SUM(ResolutionWidth + 86), SUM(ResolutionWidth + 87), SUM(ResolutionWidth + 88), SUM(ResolutionWidth + 89) FROM hits; +SELECT SearchEngineID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth) FROM hits WHERE SearchPhrase <> '' GROUP BY SearchEngineID, ClientIP ORDER BY c DESC LIMIT 10; +SELECT WatchID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth) FROM hits WHERE SearchPhrase <> '' GROUP BY WatchID, ClientIP ORDER BY c DESC LIMIT 10; +SELECT WatchID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth) FROM hits GROUP BY WatchID, ClientIP ORDER BY c DESC LIMIT 10; +SELECT URL, COUNT(*) AS c FROM hits GROUP BY URL ORDER BY c DESC LIMIT 10; +SELECT 1, URL, COUNT(*) AS c FROM hits GROUP BY 1, URL ORDER BY c DESC LIMIT 10; +SELECT ClientIP, ClientIP - 1, ClientIP - 2, ClientIP - 3, COUNT(*) AS c FROM hits GROUP BY ClientIP, ClientIP - 1, ClientIP - 2, ClientIP - 3 ORDER BY c DESC LIMIT 10; +SELECT URL, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND DontCountHits = 0 AND IsRefresh = 0 AND URL <> '' GROUP BY URL ORDER BY PageViews DESC LIMIT 10; +SELECT Title, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND DontCountHits = 0 AND IsRefresh = 0 AND Title <> '' GROUP BY Title ORDER BY PageViews DESC LIMIT 10; +SELECT URL, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 AND IsLink <> 0 AND IsDownload = 0 GROUP BY URL ORDER BY PageViews DESC LIMIT 10 OFFSET 1000; +SELECT TraficSourceID, SearchEngineID, AdvEngineID, CASE WHEN (SearchEngineID = 0 AND AdvEngineID = 0) THEN Referer ELSE '' END AS Src, URL AS Dst, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 GROUP BY TraficSourceID, SearchEngineID, AdvEngineID, Src, Dst ORDER BY PageViews DESC LIMIT 10 OFFSET 1000; +SELECT URLHash, EventDate, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 AND TraficSourceID IN (-1, 6) AND RefererHash = 3594120000172545465 GROUP BY URLHash, EventDate ORDER BY PageViews DESC LIMIT 10 OFFSET 100; +SELECT WindowClientWidth, WindowClientHeight, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 AND DontCountHits = 0 AND URLHash = 2868770270353813622 GROUP BY WindowClientWidth, WindowClientHeight ORDER BY PageViews DESC LIMIT 10 OFFSET 10000; +SELECT DATE_TRUNC('minute', EventTime) AS M, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-14' AND EventDate <= '2013-07-15' AND IsRefresh = 0 AND DontCountHits = 0 GROUP BY DATE_TRUNC('minute', EventTime) ORDER BY DATE_TRUNC('minute', EventTime) LIMIT 10 OFFSET 1000; diff --git a/spark-gluten-clickhouse/query b/spark-gluten-clickhouse/query new file mode 100755 index 0000000000..361e74c857 --- /dev/null +++ b/spark-gluten-clickhouse/query @@ -0,0 +1,13 @@ +#!/bin/bash +set -e + +# shellcheck disable=SC1091 +source myenv/bin/activate + +# Spark runs under JDK 17; Gluten was built under JDK 8 in ./install. +arch=$(dpkg --print-architecture) +export JAVA_HOME="/usr/lib/jvm/java-17-openjdk-${arch}" +export PATH="$JAVA_HOME/bin:$PATH" + +query=$(cat) +printf '%s' "$query" | python3 query.py diff --git a/spark-gluten-clickhouse/query.py b/spark-gluten-clickhouse/query.py new file mode 100755 index 0000000000..5db97c99df --- /dev/null +++ b/spark-gluten-clickhouse/query.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 + +""" +Spark + Apache Gluten using the ClickHouse backend ('ch'). The CH backend +loads libch.so (a fork of ClickHouse v23.1) into the Spark executor JVM and +runs the columnar physical plan natively. + +Reads SQL on stdin, runs it once, prints the result on stdout and the runtime +in fractional seconds as the LAST line on stderr. + +Note: Keep in sync with spark-*/query.py (see README-accelerators.md for details). +""" + +import os +import sys +import timeit + +import psutil +from pyspark.sql import SparkSession +import pyspark.sql.functions as F + + +query = sys.stdin.read() +print(query) + +# Calculate available memory to configure SparkSession (in MB). +# The CH backend runs off-heap (via JNI into libch.so), so split available +# memory between Spark's JVM heap and the off-heap pool the same way the +# Velox backend does. +ram = int(round(psutil.virtual_memory().available / (1024 ** 2) * 0.7)) +heap = ram // 2 +off_heap = ram - heap +print(f"SparkSession will use {heap} MB of heap and {off_heap} MB of off-heap memory (total {ram} MB)") + +builder = ( + SparkSession + .builder + .appName("ClickBench") + .config("spark.driver", "local[*]") # To ensure using all cores + .config("spark.driver.memory", f"{heap}m") + .config("spark.sql.parquet.binaryAsString", True) # Correct length/text results + + # Gluten + ClickHouse backend configuration + .config("spark.jars", "gluten.jar") + .config("spark.driver.extraClassPath", "gluten.jar") + .config("spark.plugins", "org.apache.gluten.GlutenPlugin") + .config("spark.shuffle.manager", "org.apache.spark.shuffle.sort.ColumnarShuffleManager") + .config("spark.gluten.sql.columnar.backend.lib", "ch") + .config("spark.gluten.sql.columnar.libpath", os.path.abspath("libch.so")) + .config("spark.memory.offHeap.enabled", "true") + .config("spark.memory.offHeap.size", f"{off_heap}m") + .config("spark.driver.extraJavaOptions", "-Dio.netty.tryReflectionSetAccessible=true") + + # Cluster-mode equivalent of the LD_PRELOAD below; a no-op in local[*] but + # kept so the config is correct if this is ever run on real executors. + .config("spark.executorEnv.LD_PRELOAD", os.path.abspath("libch.so")) +) + +# Gluten's CH backend loads libch.so into the JVM via JNI (System.load). The +# library carries initial-exec-model TLS (from its statically linked deps), so +# a lazy dlopen from the already-running JVM fails with +# java.lang.UnsatisfiedLinkError: libch.so: cannot allocate memory in static +# TLS block +# because the static TLS block is sized at process startup. Gluten's docs work +# around this with spark.executorEnv.LD_PRELOAD=, but in local[*] mode +# the driver JVM *is* the executor and is launched (by pyspark below) before any +# Spark config is read, so executorEnv never applies. Instead, preload it via +# the driver JVM's environment: setting LD_PRELOAD here does not affect this +# already-started Python process, but pyspark's launcher copies os.environ into +# the JVM it spawns, so the JVM preloads libch.so at startup while the static +# TLS block still has room. System.load() then reuses the already-loaded lib. +os.environ["LD_PRELOAD"] = os.path.abspath("libch.so") + +spark = builder.getOrCreate() + +df = spark.read.parquet("hits.parquet") +df = df.withColumn("EventTime", F.col("EventTime").cast("timestamp")) +df = df.withColumn("EventDate", F.date_add(F.lit("1970-01-01"), F.col("EventDate"))) +df.createOrReplaceTempView("hits") + +try: + start = timeit.default_timer() + result = spark.sql(query) + result.show(100) + end = timeit.default_timer() + elapsed = end - start + print(f"Time: {elapsed}") + print(f"{elapsed:.6f}", file=sys.stderr) +except Exception as e: + print(e, file=sys.stderr) + print("Failure!", file=sys.stderr) + sys.exit(1) diff --git a/spark-gluten-clickhouse/start b/spark-gluten-clickhouse/start new file mode 100755 index 0000000000..06bd986563 --- /dev/null +++ b/spark-gluten-clickhouse/start @@ -0,0 +1,2 @@ +#!/bin/bash +exit 0 diff --git a/spark-gluten-clickhouse/stop b/spark-gluten-clickhouse/stop new file mode 100755 index 0000000000..06bd986563 --- /dev/null +++ b/spark-gluten-clickhouse/stop @@ -0,0 +1,2 @@ +#!/bin/bash +exit 0 diff --git a/spark-gluten-clickhouse/template.json b/spark-gluten-clickhouse/template.json new file mode 100644 index 0000000000..6ae1ad2873 --- /dev/null +++ b/spark-gluten-clickhouse/template.json @@ -0,0 +1,13 @@ +{ + "system": "Spark (Gluten-on-ClickHouse)", + "proprietary": "no", + "hardware": "cpu", + "tuned": "no", + "tags": [ + "Java", + "C++", + "column-oriented", + "Spark derivative", + "ClickHouse" + ] +}