diff --git a/.github/workflows/RunBenchmark.yml b/.github/workflows/RunBenchmark.yml index 385bd1f0..3539d720 100644 --- a/.github/workflows/RunBenchmark.yml +++ b/.github/workflows/RunBenchmark.yml @@ -100,12 +100,14 @@ jobs: fi - name: name new branch + if: always() shell: bash run: | echo "new_branch_name=results-`date +%Y-%m-%d-%Hh%Mm`" >> $GITHUB_ENV echo ${{ env.new_branch_name }} - name: Commit updates + if: always() shell: bash working-directory: /var/lib/mount/db-benchmark-metal run: | diff --git a/_launcher/launcher.R b/_launcher/launcher.R index ae92dbdf..456afd96 100644 --- a/_launcher/launcher.R +++ b/_launcher/launcher.R @@ -203,7 +203,21 @@ launch = function(dt, mockup, out_dir="out") { if (file.exists(ret_file)) file.remove(ret_file) } cmd = sprintf("%s > %s 2> %s", solution.cmd(s, t, d), out_file, err_file) # ./_launcher/solution.R ... > out 2> err - shcmd = sprintf("/bin/bash -c \"%s%s\"", venv, cmd) # this is needed to source python venv + inner = sprintf("/bin/bash -c \"%s%s\"", venv, cmd) # this is needed to source python venv + # Run each solution in its own transient systemd scope so the OOM killer + # targets only that scope and cannot propagate to the R launcher process. + # MemoryMax is set to 90% of total physical RAM. + mem_max = tryCatch({ + meminfo = readLines("/proc/meminfo") + total_kb = as.integer(sub(".*:\\s*(\\d+)\\s*kB", "\\1", meminfo[grepl("^MemTotal", meminfo)])) + as.integer(total_kb * 1024 * 0.9) + }, error = function(e) NA_integer_) + cat(sprintf("systemd-run MemoryMax: %s\n", if (is.na(mem_max)) "NA (/proc/meminfo unavailable, scope wrapping disabled)" else sprintf("%d bytes (%.1f GB)", mem_max, mem_max/1024^3))) + if (!is.na(mem_max) && nzchar(Sys.which("systemd-run"))) { + shcmd = sprintf("systemd-run --scope -p MemoryMax=%d %s", mem_max, inner) + } else { + shcmd = inner + } if (mockup) { cat(cmd) cat(shcmd) diff --git a/clickhouse/exec.sh b/clickhouse/exec.sh index e1eb7dc4..cd097c77 100755 --- a/clickhouse/exec.sh +++ b/clickhouse/exec.sh @@ -52,7 +52,9 @@ if [ $1 == 'groupby' ]; then fi clickhouse-client --user db_benchmark --query "INSERT INTO $SRC_DATANAME FROM INFILE 'data/${SRC_DATANAME}.csv'" # confirm all data loaded - echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' + if [[ $TEST_RUN != "true" ]]; then + echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' + fi export THREADS elif [ $1 == 'join' ]; then RHS=$(join_to_tbls $SRC_DATANAME) @@ -106,10 +108,12 @@ elif [ $1 == 'join' ]; then tail -n+2 data/$RHS3.csv | clickhouse-client --user db_benchmark --query "INSERT INTO $RHS3 SELECT * FROM input('id1 Nullable(Int32), id2 Nullable(Int32), id3 Nullable(Int32), id4 Nullable(String), id5 Nullable(String), id6 Nullable(String), v2 Nullable(Float64)') FORMAT CSV" # validate - echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' - echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS1'\n$(echo $RHS1 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' - echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS2'\n$(echo $RHS2 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' - echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS3'\n$(echo $RHS3 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' + if [[ $TEST_RUN != "true" ]]; then + echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' + echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS1'\n$(echo $RHS1 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' + echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS2'\n$(echo $RHS2 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' + echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS3'\n$(echo $RHS3 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)' + fi export RHS1 export RHS2 diff --git a/dask/groupby_dask.py b/dask/groupby_dask.py index 3ab86942..aeb6660b 100755 --- a/dask/groupby_dask.py +++ b/dask/groupby_dask.py @@ -63,7 +63,6 @@ def load_dataset( x = dd.read_csv( src_grp, dtype={"id1":"category","id2":"category","id3":"category","id4":"Int32","id5":"Int32","id6":"Int32","v1":"Int32","v2":"Int32","v3":"float64"}, - engine="pyarrow", **kwargs ) x = x.persist() @@ -139,7 +138,7 @@ class QuerySix(Query): @staticmethod def query(x: dd.DataFrame) -> dd.DataFrame: - ans = x.groupby(['id4','id5'], dropna=False, observed=True).agg({'v3': ['median','std']}, shuffle='p2p').compute() + ans = x.groupby(['id4','id5'], dropna=False, observed=True).agg({'v3': ['median','std']}, shuffle='tasks').compute() ans.reset_index(inplace=True) # #68 return ans @@ -166,7 +165,7 @@ class QueryEight(Query): @staticmethod def query(x: dd.DataFrame) -> dd.DataFrame: - ans = x[~x['v3'].isna()][['id6','v3']].groupby('id6', dropna=False, observed=True).apply(lambda x: x.nlargest(2, columns='v3'), meta={'id6':'Int64', 'v3':'float64'})[['v3']].compute() + ans = x[~x['v3'].isna()][['id6','v3']].groupby('id6', dropna=False, observed=True).apply(lambda x: x.nlargest(2, columns='v3'), meta={'v3':'float64'}, include_groups=False)[['v3']].compute() ans.reset_index(level='id6', inplace=True) ans.reset_index(drop=True, inplace=True) # drop because nlargest creates some extra new index field return ans @@ -180,7 +179,7 @@ class QueryNine(Query): @staticmethod def query(x: dd.DataFrame) -> dd.DataFrame: - ans = x[['id2','id4','v1','v2']].groupby(['id2','id4'], dropna=False, observed=True).apply(lambda x: pd.Series({'r2': x.corr()['v1']['v2']**2}), meta={'r2':'float64'}).compute() + ans = x[['id2','id4','v1','v2']].groupby(['id2','id4'], dropna=False, observed=True).apply(lambda x: pd.Series({'r2': x.corr()['v1']['v2']**2}), meta={'r2':'float64'}, include_groups=False).compute() ans.reset_index(inplace=True) return ans diff --git a/dask/join_dask.py b/dask/join_dask.py index cb256dc7..9641d6e3 100755 --- a/dask/join_dask.py +++ b/dask/join_dask.py @@ -153,14 +153,14 @@ def load_datasets( logging.info("Loading dataset: %s" % data_name) logging.info("Reading source: %s" % src_jn_x) - x = dd.read_csv(src_jn_x, engine="pyarrow", **kwargs).persist() + x = dd.read_csv(src_jn_x, **kwargs).persist() logging.info("Reading source: %s" % src_jn_y[0]) - small = dd.read_csv(src_jn_y[0], engine="pyarrow", **kwargs).persist() + small = dd.read_csv(src_jn_y[0], **kwargs).persist() logging.info("Reading source: %s" % src_jn_y[1]) - medium = dd.read_csv(src_jn_y[1], engine="pyarrow", **kwargs).persist() + medium = dd.read_csv(src_jn_y[1], **kwargs).persist() logging.info("Reading source: %s" % src_jn_y[2]) - big = dd.read_csv(src_jn_y[2], engine="pyarrow", **kwargs).persist() + big = dd.read_csv(src_jn_y[2], **kwargs).persist() return [ x, diff --git a/dask/ver-dask.sh b/dask/ver-dask.sh index 66f23520..ac25b586 100755 --- a/dask/ver-dask.sh +++ b/dask/ver-dask.sh @@ -2,4 +2,4 @@ set -e source ./dask/py-dask/bin/activate -python3 -c 'import dask as dk; open("dask/VERSION","w").write(dk.__version__); open("dask/REVISION","w").write(dk.__git_revision__);' > /dev/null +python3 -c 'import dask as dk; open("dask/VERSION","w").write(dk.__version__); open("dask/REVISION","w").write(dk.__git_revision__ or "");' > /dev/null diff --git a/path.env b/path.env index 7a87b42f..8ad8a879 100644 --- a/path.env +++ b/path.env @@ -1,4 +1,8 @@ -export JULIA_HOME=/opt/julia-1.9.2 +if [ -d "$HOME/.juliaup/bin" ]; then + export JULIA_HOME=$HOME/.juliaup +else + export JULIA_HOME=/opt/julia-1.9.2 +fi export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64 export PATH=$PATH:$JULIA_HOME/bin export MOUNT_POINT=/var/lib/mount diff --git a/run.sh b/run.sh index c2408bb2..d0361767 100755 --- a/run.sh +++ b/run.sh @@ -98,7 +98,13 @@ if [[ "$RUN_SOLUTIONS" =~ "haskell" ]]; then ./haskell/ver-haskell.sh; fi; # run if [[ -f ./stop ]]; then echo "# Benchmark run $BATCH has been interrupted after $(($(date +%s)-$BATCH))s due to 'stop' file" && rm -f ./stop && rm -f ./run.lock && exit; fi; echo "# Running benchmark scripts launcher" +set +e Rscript ./_launcher/launch.R +LAUNCH_EXIT=$? +set -e +if [ $LAUNCH_EXIT -ne 0 ]; then + echo "# WARNING: launcher exited with code $LAUNCH_EXIT (possible OOM or crash), partial results may have been recorded" +fi if [[ -f ./stop ]]; then echo "# Benchmark run $BATCH has been interrupted after $(($(date +%s)-$BATCH))s due to 'stop' file" && rm -f ./stop && rm -f ./run.lock && exit; fi; # publish report for all tasks diff --git a/spark/groupby-spark.py b/spark/groupby-spark.py index 1228a075..9c3b8d60 100755 --- a/spark/groupby-spark.py +++ b/spark/groupby-spark.py @@ -25,10 +25,10 @@ print("loading dataset %s" % data_name, flush=True) mem_usage = "240g" -if "TEST_RUN" in os.environ: - mem_usage = "2g" if machine_type == 'c6id.4xlarge': mem_usage = "30g" +if "TEST_RUN" in os.environ: + mem_usage = "2g" from pyspark.conf import SparkConf diff --git a/spark/join-spark.py b/spark/join-spark.py index 319f80f3..eef901f6 100755 --- a/spark/join-spark.py +++ b/spark/join-spark.py @@ -27,10 +27,10 @@ raise Exception("Something went wrong in preparing files used for join") mem_usage = "220g" -if "TEST_RUN" in os.environ: - mem_usage = "2g" if machine_type == 'c6id.4xlarge': mem_usage = "30g" +if "TEST_RUN" in os.environ: + mem_usage = "2g" from pyspark.conf import SparkConf spark = SparkSession.builder \