tuplex · bgivertz · Feb 4, 2022 · Feb 9, 2022 · Feb 9, 2022 · Feb 10, 2022
diff --git a/benchmarks/incremental/.gitignore b/benchmarks/incremental/.gitignore
@@ -0,0 +1,6 @@
+data/*.csv
+tuplex_output/
+python_code_pipeline_stage_*.py
+transform_stage_*.txt
+results_dirty_zillow@10G/
+tuplex_config.json
diff --git a/benchmarks/incremental/README.md b/benchmarks/incremental/README.md
@@ -0,0 +1,25 @@
+
+## Incremental Exception Resolution Experiment
+
+In this experiment, we run a sequence of six pipelines over the dirty zillow dataset.
+- The first pipeline contains no ignore or resolver operations
+- The final pipeline contains 5 unique ignore and resolver operations
+- Each of the pipelines in between incrementally adds on an additional resolver until all are present for the final pipeline
+
+We compare the following conditions, for a total of 8 experimental trials
+- Plain vs Incremental Resolution
+- Single vs Multi-threaded
+- Merge in order vs Merge without order
+
+In order to get 10GB of input data, replicate dirty zillow data 1460x (or use 1500x for simplicity).
+
+### Setup
+To replicate the original data, create the 10G files with the following settings:
+```
+python3 replicate-data.py -s 1460 -o data/zillow_dirty@10G.csv
+```
+Note that both files have the same number of rows, but the synthetic version is slightly larger.
+
+### Running the benchmark
+Use
+`nohup perflock ./benchmark.sh -hwloc &`
diff --git a/benchmarks/incremental/benchmark-synthetic.sh b/benchmarks/incremental/benchmark-synthetic.sh
@@ -0,0 +1,96 @@
+#!/usr/bin/env bash
+
+# Parse HWLOC settings
+HWLOC=""
+if [ $# -ne 0 ] && [ $# -ne 1 ]; then # check nmber of inputs
+  echo "usage: ./benchmark.sh [-hwloc]"
+  exit 1
+fi
+
+if [ $# -eq 1 ]; then # check if hwloc
+  if [ "$1" != "-hwloc" ]; then # check flag
+    echo -e "invalid flag: $1\nusage: ./benchmark.sh [-hwloc]"
+    exit 1
+  fi
+  HWLOC="hwloc-bind --cpubind node:1 --membind node:1 --cpubind node:2 --membind node:2"
+fi
+
+# use 10 runs (3 for very long jobs) and a timeout after 180min/3h
+NUM_RUNS=1
+NUM_STEPS=10
+TIMEOUT=14400
+
+RESDIR='results_synthetic'
+DATA_PATH='/hot/scratch/bgivertz/data/synthetic/synth'
+INCREMENTAL_OUT_PATH='/hot/scratch/bgivertz/output/incremental'
+COMMIT_OUT_PATH='/hot/scratch/bgivertz/output/commit'
+PLAIN_OUT_PATH='/hot/scratch/bgivertz/output/plain'
+
+rm -rf $RESDIR
+rm -rf $INCREMENTAL_OUT_PATH
+rm -rf $PLAIN_OUT_PATH
+rm -rf $COMMIT_OUT_PATH
+
+mkdir -p ${RESDIR}
+
+# create tuplex_config.json
+python3 create_conf.py --opt-pushdown --opt-filter --opt-llvm --executor-count 63 --executor-memory "6G" > tuplex_config.json
+
+echo "running out of order experiments"
+for ((r = 1; r <= NUM_RUNS; r++)); do
+  echo "trial ($r/$NUM_RUNS)"
+
+  echo "running plain (0/10)"
+  LOG="${RESDIR}/plain-out-of-order-e0-t$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --clear-cache --input-path "$DATA_PATH""0.csv" --output-path $PLAIN_OUT_PATH >$LOG 2>$LOG.stderr
+
+  echo "running incremental (0/10)"
+  LOG="${RESDIR}/incremental-out-of-order-e0-t$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --clear-cache --incremental-resolution --input-path "$DATA_PATH""0.csv" --output-path $INCREMENTAL_OUT_PATH >$LOG 2>$LOG.stderr
+
+  for ((s = 1; s <= 10; s++)) do
+    echo "running plain ($s/10)"
+    LOG="${RESDIR}/plain-out-of-order-e$s-t$r.txt"
+    timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --use-resolve-step --clear-cache --input-path "$DATA_PATH$s.csv" --output-path $PLAIN_OUT_PATH >$LOG 2>$LOG.stderr
+
+    echo "running incremental ($s/10)"
+    LOG="${RESDIR}/incremental-out-of-order-e$s-t$r.txt"
+    timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --use-resolve-step --clear-cache --incremental-resolution --input-path "$DATA_PATH$s.csv" --output-path $INCREMENTAL_OUT_PATH >$LOG 2>$LOG.stderr
+  done
+done
+
+echo "running in order experiments"
+for ((r = 1; r <= NUM_RUNS; r++)); do
+  echo "trial ($r/$NUM_RUNS)"
+
+  echo "running plain (0/10)"
+  LOG="${RESDIR}/plain-in-order-e0-t$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --clear-cache --resolve-in-order --input-path "$DATA_PATH""0.csv" --output-path $PLAIN_OUT_PATH >$LOG 2>$LOG.stderr
+
+  echo "running incremental (0/10)"
+  LOG="${RESDIR}/incremental-in-order-e0-t$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --clear-cache --resolve-in-order --incremental-resolution --input-path "$DATA_PATH""0.csv" --output-path $INCREMENTAL_OUT_PATH >$LOG 2>$LOG.stderr
+
+  echo "running commit (0/10)"
+  LOG="${RESDIR}/commit-in-order-e0-t$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --clear-cache --resolve-in-order --incremental-resolution --commit --input-path "$DATA_PATH""0.csv" --output-path $COMMIT_OUT_PATH >$LOG 2>$LOG.stderr
+
+  for ((s = 1; s <= 10; s++)) do
+    echo "running plain ($s/10)"
+    LOG="${RESDIR}/plain-in-order-e$s-t$r.txt"
+    timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --use-resolve-step --clear-cache --resolve-in-order --input-path "$DATA_PATH$s.csv" --output-path $PLAIN_OUT_PATH >$LOG 2>$LOG.stderr
+
+    echo "running incremental ($s/10)"
+    LOG="${RESDIR}/incremental-in-order-e$s-t$r.txt"
+    timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --use-resolve-step --clear-cache --resolve-in-order --incremental-resolution --input-path "$DATA_PATH$s.csv" --output-path $INCREMENTAL_OUT_PATH >$LOG 2>$LOG.stderr
+
+    echo "running commit ($s/10)"
+    LOG="${RESDIR}/commit-in-order-e$s-t$r.txt"
+    timeout $TIMEOUT ${HWLOC} python3 runsynthetic.py --use-resolve-step --clear-cache --resolve-in-order --incremental-resolution --commit --input-path "$DATA_PATH$s.csv" --output-path $COMMIT_OUT_PATH >$LOG 2>$LOG.stderr
+  done
+done
+
+
+rm -rf $INCREMENTAL_OUT_PATH
+rm -rf $PLAIN_OUT_PATH
+rm -rf $COMMIT_OUT_PATH
diff --git a/benchmarks/incremental/benchmark.sh b/benchmarks/incremental/benchmark.sh
@@ -0,0 +1,90 @@
+#!/usr/bin/env bash
+
+# Parse HWLOC settings
+HWLOC=""
+if [ $# -ne 0 ] && [ $# -ne 1 ]; then # check nmber of inputs
+  echo "usage: ./benchmark.sh [-hwloc]"
+  exit 1
+fi
+
+if [ $# -eq 1 ]; then # check if hwloc
+  if [ "$1" != "-hwloc" ]; then # check flag
+    echo -e "invalid flag: $1\nusage: ./benchmark.sh [-hwloc]"
+    exit 1
+  fi
+  HWLOC="hwloc-bind --cpubind node:1 --membind node:1 --cpubind node:2 --membind node:2"
+fi
+
+# use 10 runs (3 for very long jobs) and a timeout after 180min/3h
+NUM_RUNS=3
+TIMEOUT=14400
+
+RESDIR='results_dirty_zillow@100G'
+DATA_PATH_SSD='/hot/scratch/bgivertz/data/zillow_dirty@100G.csv'
+INCREMENTAL_OUT_PATH_SSD='/hot/scratch/bgivertz/output/incremental'
+INCREMENTAL_COMMIT_OUT_PATH_SSD='/hot/scratch/bgivertz/output/commit'
+PLAIN_OUT_PATH_SSD='/hot/scratch/bgivertz/output/plain'
+
+rm -rf $RESDIR
+rm -rf $INCREMENTAL_OUT_PATH_SSD
+rm -rf $PLAIN_OUT_PATH_SSD
+rm -rf $INCREMENTAL_COMMIT_OUT_PATH_SSD
+
+# does file exist?
+if [[ ! -f "$DATA_PATH_SSD" ]]; then
+	echo "file $DATA_PATH_SSD not found, abort."
+	exit 1
+fi
+
+mkdir -p ${RESDIR}
+
+# create tuplex_config.json
+python3 create_conf.py --opt-pushdown --opt-filter --opt-llvm --executor-count 63 --executor-memory "6G" > tuplex_config.json
+
+echo "running out-of-order ssd experiments"
+for ((r = 1; r <= NUM_RUNS; r++)); do
+  echo "trial ($r/$NUM_RUNS)"
+
+  echo "running plain"
+  LOG="${RESDIR}/tuplex-plain-out-of-order-ssd-$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runtuplex.py --clear-cache --path $DATA_PATH_SSD --output-path $PLAIN_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+
+  echo "running incremental"
+  LOG="${RESDIR}/tuplex-incremental-out-of-order-ssd-$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runtuplex.py --clear-cache --incremental-resolution --path $DATA_PATH_SSD --output-path $INCREMENTAL_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+
+#  echo "validating results"
+#  LOG="${RESDIR}/tuplex-compare-out-of-order-ssd-$r.txt"
+#  timeout $TIMEOUT ${HWLOC} python3 compare_folders.py $PLAIN_OUT_PATH_SSD $INCREMENTAL_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+done
+
+echo "running in-order ssd experiments"
+for ((r = 1; r <= NUM_RUNS; r++)); do
+  echo "trial ($r/$NUM_RUNS)"
+
+  echo "running plain"
+  LOG="${RESDIR}/tuplex-plain-in-order-ssd-$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runtuplex.py --clear-cache --resolve-in-order --path $DATA_PATH_SSD --output-path $PLAIN_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+
+  echo "running incremental"
+  LOG="${RESDIR}/tuplex-incremental-in-order-ssd-$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runtuplex.py --clear-cache --resolve-in-order --incremental-resolution --path $DATA_PATH_SSD --output-path $INCREMENTAL_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+
+  echo "running commit"
+  LOG="${RESDIR}/tuplex-incremental-in-order-commit-ssd-$r.txt"
+  timeout $TIMEOUT ${HWLOC} python3 runtuplex.py --clear-cache --resolve-in-order --incremental-resolution --commit --path $DATA_PATH_SSD --output-path $INCREMENTAL_COMMIT_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+
+#  echo "validating results"
+#  LOG="${RESDIR}/tuplex-compare-in-order-ssd-$r.txt"
+#  timeout $TIMEOUT ${HWLOC} python3 compare_folders.py --in-order $PLAIN_OUT_PATH_SSD $INCREMENTAL_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+#
+#  LOG="${RESDIR}/tuplex-compare-in-order-commit-ssd-$r.txt"
+#  timeout $TIMEOUT ${HWLOC} python3 compare_folders.py --in-order $INCREMENTAL_COMMIT_OUT_PATH_SSD $INCREMENTAL_OUT_PATH_SSD >$LOG 2>$LOG.stderr
+done
+
+echo "graphing results"
+python3 graph.py --results-path $RESDIR --num-trials $NUM_RUNS --num-steps 7
+
+rm -rf $INCREMENTAL_OUT_PATH_SSD
+rm -rf $PLAIN_OUT_PATH_SSD
+rm -rf $INCREMENTAL_COMMIT_OUT_PATH_SSD
diff --git a/benchmarks/incremental/compare_folders.py b/benchmarks/incremental/compare_folders.py
@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+# (c) L.Spiegelberg 2021
+# compare the csv output contents of two folders (ignoring order)
+
+import os
+import sys
+import argparse
+import glob
+
+
+def wc_files(path):
+    files = sorted(glob.glob(os.path.join(path, '*.csv')))
+
+    all_lines = []
+    num_rows = 0
+    header = None
+    matching_headers = 0
+    for f in files:
+        with open(f, 'r') as fp:
+            lines = fp.readlines()
+            if header is None and len(lines) > 0:
+                header = lines[0]
+            num_rows += len(lines)
+            if len(lines) > 0:
+                if header == lines[0]:
+                    matching_headers += 1
+                    all_lines += lines[1:]
+                else:
+                    all_lines += lines
+
+    if matching_headers == len(files):
+        num_rows -= matching_headers
+
+    print('-- counted {} rows in {} files in folder {}'.format(num_rows, len(files), path))
+    return num_rows, len(files), all_lines
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("folderA")
+    parser.add_argument("folderB")
+    parser.add_argument("--in-order", help='whether to compare in order', action='store_true')
+    args = parser.parse_args()
+
+    print('== Dirty Zillow experiment validation ==')
+
+    # count lines in each folder
+    print('-- loading folder contents...')
+    rowCountA, filesA, rowsA  = wc_files(args.folderA)
+    rowCountB, filesB, rowsB = wc_files(args.folderB)
+
+    if rowCountA != rowCountB:
+        print('>>> number of rows does not match')
+        sys.exit(1)
+
+    # sort lines and compare them
+    if not args.in_order:
+        print('-- sorting rows from {}'.format(args.folderA))
+        rowsA = sorted(rowsA)
+        print('-- sorting rows from {}'.format(args.folderB))
+        rowsB = sorted(rowsB)
+
+    print('-- computing comparison of rows...')
+    non_matching_indices = [ind for ind, (i, j) in enumerate(zip(rowsA, rowsB)) if i != j]
+
+    if len(non_matching_indices) > 0:
+        print('>>> rows do not match up, details:')
+
+        for idx in non_matching_indices:
+            print('{:5d}: {} != {}'.format(idx, rowsA[idx], rowsB[idx]))
+        sys.exit(1)
+
+    print('>>> contents of folders match.')
+
+    sys.exit(0)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/benchmarks/incremental/create-synthetic.sh b/benchmarks/incremental/create-synthetic.sh
@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+
+set -x
+
+python3 synthesize-data.py --dataset-size $1 --output-path /hot/scratch/bgivertz/data/synthetic/synth0.csv --exceptions 0
+
+for ((i = 1; i <= 9; i++)) do
+  python3 synthesize-data.py --dataset-size $1 --output-path /hot/scratch/bgivertz/data/synthetic/synth$i.csv --exceptions 0.$i
+done
+
+python3 synthesize-data.py --dataset-size $1 --output-path /hot/scratch/bgivertz/data/synthetic/synth10.csv --exceptions 1
diff --git a/benchmarks/incremental/create_conf.py b/benchmarks/incremental/create_conf.py
@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+# (c) 2020 L.Spiegelberg
+# this script creates Tuplex json configuration files for benchmarks
+
+import json
+import argparse
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--executor-memory', default='10G', help='how much memory each thread gets')
+    parser.add_argument('--executor-count', default=15, help='how many worker threads')
+    parser.add_argument('--partition-size', default='32MB', help='task size')
+    parser.add_argument('--runtime-memory', default='64MB', help='how much maximum runtime memory to use')
+    parser.add_argument('--input-split-size', default='64MB', help='chunk size of input files')
+    parser.add_argument('--opt-null', help='enable null value optimization', action='store_true')
+    parser.add_argument('--opt-pushdown', help='enable projection pushdown', action='store_true')
+    parser.add_argument('--opt-filter', help='enable filter pushdown', action='store_true')
+    parser.add_argument('--opt-parser', help='generate CSV parser', action='store_true')
+    parser.add_argument('--opt-llvm', help='run llvm optimizers', action='store_true')
+
+    args = parser.parse_args()
+
+    conf = {'webui.enable' : False,
+            'executorMemory' : args.executor_memory,
+            'executorCount' : args.executor_count,
+            'driverMemory' : args.executor_memory,
+            'partitionSize' : args.partition_size,
+            'runTimeMemory' : args.runtime_memory,
+            'inputSplitSize' : args.input_split_size,
+            'useLLVMOptimizer' : args.opt_llvm,
+            'optimizer.nullValueOptimization' : args.opt_null,
+            'csv.selectionPushdown' : args.opt_pushdown,
+            'optimizer.generateParser' : args.opt_parser,
+            'optimizer.filterPushdown' : args.opt_filter}
+
+    print(json.dumps(conf))