diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 00000000..988fca01
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,48 @@
+# syntax = docker/dockerfile:1.0-experimental
+ARG BASE_IMAGE=nvidia/cuda:11.0-runtime-ubuntu20.04
+FROM ${BASE_IMAGE}
+
+ARG GROUP_ID=10000
+ARG USER_ID=10000
+ARG USER_GROUP_NAME=rapids
+ARG CONDA_ENV_FILE=conda/rapids-tpcx-bb.yml
+
+ENV PATH /conda/bin:$PATH
+ADD https://repo.continuum.io/miniconda/Miniconda3-latest-Linux-x86_64.sh miniconda.sh
+RUN bash miniconda.sh -f -b -p /conda && \
+    conda init bash && \
+    apt-get update && apt-get install -y git vim
+
+ENV CUDA_VERSION=${CUDA_VERSION}
+ENV NVIDIA_VISIBLE_DEVICES=${NVIDIA_VISIBLE_DEVICES}
+ENV hostname=${hostname}_cuda${CUDA_VERSION}-${DISTRO}
+
+ENV DEBIAN_FRONTEND=noninteractive
+ENV OMPI_MCA_opal_cuda_support=true
+
+#Workaround for http://nvbugs/2903712
+ENV IBV_DRIVERS=/usr/lib/libibverbs/libmlx5
+
+RUN groupadd --gid ${GROUP_ID} ${USER_GROUP_NAME} && \
+    useradd -g ${GROUP_ID} -u ${USER_ID} -ms /bin/bash ${USER_GROUP_NAME} && \
+    cat /conda/etc/profile.d/conda.sh >> /etc/profile && \
+    echo "conda activate tpcxbb" >> /etc/profile && \
+    cat /conda/etc/profile.d/conda.sh >> /home/${USER_GROUP_NAME}/.bashrc && \
+    echo "conda activate tpcxbb" >> /home/${USER_GROUP_NAME}/.bashrc && \
+    chgrp -R ${USER_GROUP_NAME} /conda && \
+    chmod -R g+rwx /conda
+
+COPY ${CONDA_ENV_FILE} /home/${USER_GROUP_NAME}/environment.yml
+RUN bin/bash -c "source /conda/etc/profile.d/conda.sh && \
+                 conda env create -f /home/${USER_GROUP_NAME}/environment.yml -n tpcxbb" && \
+    chmod -R g+w /conda && \
+    chown -R ${USER_GROUP_NAME}:${USER_GROUP_NAME} /conda/envs/tpcxbb
+
+COPY tpcx_bb /home/${USER_GROUP_NAME}/tpcx_bb
+RUN chown -R ${USER_GROUP_NAME}:${USER_GROUP_NAME} /home/${USER_GROUP_NAME}/tpcx_bb && \
+    /bin/bash -c "source /conda/etc/profile.d/conda.sh && \
+                  conda activate tpcxbb && \
+                  python -m pip install /home/${USER_GROUP_NAME}/tpcx_bb/."
+
+USER ${USER_GROUP_NAME}
+WORKDIR /home/${USER_GROUP_NAME}/
diff --git a/conda/rapids-tpcx-bb.yml b/conda/rapids-tpcx-bb.yml
index 2381f582..b5e9318d 100644
--- a/conda/rapids-tpcx-bb.yml
+++ b/conda/rapids-tpcx-bb.yml
@@ -1,5 +1,5 @@
 channels:
-  - blazingsql-nightly/label/cuda10.2
+  - blazingsql-nightly
   - rapidsai-nightly
   - nvidia
   - pytorch-nightly
@@ -9,7 +9,7 @@ channels:
 
 dependencies:
   - python=3.7
-  - cudatoolkit=10.2
+  - cudatoolkit
   - cudf
   - rmm
   - dask-cuda
diff --git a/tpcx_bb/benchmark_runner.py b/tpcx_bb/benchmark_runner.py
index fa3371b2..83b10d59 100644
--- a/tpcx_bb/benchmark_runner.py
+++ b/tpcx_bb/benchmark_runner.py
@@ -11,10 +11,11 @@ def get_qnum_from_filename(name):
     m = re.search("[0-9]{2}", name).group()
     return m
 
+base_path = os.path.split(os.path.abspath(__file__))[0]
 
 dask_qnums = [str(i).zfill(2) for i in range(1, 31)]
 # Not all queries are implemented with BSQL
-bsql_query_files = sorted(glob.glob("./queries/q*/t*_sql.py"))
+bsql_query_files = sorted(glob.glob(os.path.join( base_path, "queries/q*/t*_sql.py")))
 bsql_qnums = [get_qnum_from_filename(x.split("/")[-1]) for x in bsql_query_files]
 
 def load_query(qnum, fn):
@@ -24,7 +25,7 @@ def load_query(qnum, fn):
     loader.exec_module(mod)
     return mod.main
 
-if __name__ == "__main__":
+def main():
     from xbb_tools.cluster_startup import attach_to_cluster, import_query_libs
     from xbb_tools.utils import run_query, tpcxbb_argparser
 
@@ -41,14 +42,12 @@ def load_query(qnum, fn):
 
     
     config = tpcxbb_argparser()
-    include_blazing = config.get("benchmark_runner_include_bsql")
+    include_blazing = config.get("with_blazing")
     client, bc = attach_to_cluster(config, create_blazing_context=include_blazing)
 
     # Preload required libraries for queries on all workers
     client.run(import_query_libs)
 
-    base_path = os.getcwd()
-
     # Run Pure Dask Queries
     if len(dask_qnums) > 0:
         print("Pure Dask Queries")
@@ -93,3 +92,6 @@ def load_query(qnum, fn):
                 client.run_on_scheduler(gc.collect)
                 gc.collect()
                 time.sleep(3)
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file
diff --git a/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh b/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh
old mode 100644
new mode 100755
index ae228ea2..43e335bd
--- a/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh
+++ b/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh
@@ -1,25 +1,29 @@
+#!/usr/bin/env bash
+
+set -x
+
 #IB, NVLINK, or TCP
-CLUSTER_MODE=$1
+CLUSTER_MODE=${CLUSTER_MODE:="IB"}
 USERNAME=$(whoami)
 
 MAX_SYSTEM_MEMORY=$(free -m | awk '/^Mem:/{print $2}')M
-DEVICE_MEMORY_LIMIT="25GB"
-POOL_SIZE="30GB"
+DEVICE_MEMORY_LIMIT=${DEVICE_MEMORY_LIMIT:="25GB"}
+POOL_SIZE=${POOL_SIZE:="30GB"}
 
 # Fill in your environment name and conda path on each node
-TPCX_BB_HOME="/home/$USERNAME/shared/tpcx-bb"
-CONDA_ENV_NAME="rapids-tpcx-bb"
-CONDA_ENV_PATH="/home/$USERNAME/conda/etc/profile.d/conda.sh"
+TPCX_BB_HOME=${TPCX_BB_HOME:="/home/$USERNAME/shared/tpcx-bb"}
+CONDA_ENV_NAME=${CONDA_ENV_NAME:="rapids-tpcx-bb"}
+CONDA_ENV_PATH=${CONDA_ENV_PATH:="/home/$USERNAME/conda/etc/profile.d/conda.sh"}
 
 # TODO: Unify interface/IP setting/getting for cluster startup
 # and scheduler file
-INTERFACE="ib0"
+INTERFACE=${INTERFACE:="ib0"}
 
 # TODO: Remove hard-coding of scheduler
-SCHEDULER=$(hostname)
-SCHEDULER_FILE=$TPCX_BB_HOME/tpcx_bb/cluster_configuration/example-cluster-scheduler.json
-LOGDIR="/tmp/tpcx-bb-dask-logs/"
-WORKER_DIR="/tmp/tpcx-bb-dask-workers/"
+SCHEDULER=${SCHEDULER:=$(hostname)}
+SCHEDULER_FILE=${SCHEDULER_FILE:="${TPCX_BB_HOME}/tpcx_bb/cluster_configuration/example-cluster-scheduler.json"}
+LOGDIR=${LOGDIR:="/tmp/tpcx-bb-dask-logs/"}
+WORKER_DIR=${WORKER_DIR:="/tmp/tpcx-bb-dask-workers/"}
 
 # Purge Dask worker and log directories
 rm -rf $LOGDIR/*
@@ -30,15 +34,17 @@ mkdir -p $WORKER_DIR
 # Purge Dask config directories
 rm -rf ~/.config/dask
 
-# Activate conda environment
-source $CONDA_ENV_PATH
+if [ "${CONDA_DEFAULT_ENV}" != "${CONDA_ENV_NAME}" ] ; then
+  # Activate conda environment
+  source $CONDA_ENV_PATH
+fi
 conda activate $CONDA_ENV_NAME
 
 # Dask/distributed configuration
-export DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT="100s"
-export DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP="600s"
-export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN="1s"
-export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX="60s"
+export DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT=${DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT:="100s"}
+export DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP=${DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP:="600s"}
+export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN=${DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN:="1s"}
+export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX=${DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX:="60s"}
 
 
 # Setup scheduler
diff --git a/tpcx_bb/cluster_configuration/cluster-startup.sh b/tpcx_bb/cluster_configuration/cluster-startup.sh
old mode 100644
new mode 100755
index 72519c80..f7c83005
--- a/tpcx_bb/cluster_configuration/cluster-startup.sh
+++ b/tpcx_bb/cluster_configuration/cluster-startup.sh
@@ -1,25 +1,29 @@
+#!/usr/bin/env bash
+
+set -x
+
 #IB, NVLINK, or TCP
-CLUSTER_MODE=$1
+CLUSTER_MODE=${CLUSTER_MODE:="IB"}
 USERNAME=$(whoami)
 
 MAX_SYSTEM_MEMORY=$(free -m | awk '/^Mem:/{print $2}')M
-DEVICE_MEMORY_LIMIT="25GB"
-POOL_SIZE="30GB"
+DEVICE_MEMORY_LIMIT=${DEVICE_MEMORY_LIMIT:="25GB"}
+POOL_SIZE=${POOL_SIZE:="30GB"}
 
 # Fill in your environment name and conda path on each node
-TPCX_BB_HOME="/home/$USERNAME/shared/tpcx-bb"
-CONDA_ENV_NAME="rapids-tpcx-bb"
-CONDA_ENV_PATH="/home/$USERNAME/conda/etc/profile.d/conda.sh"
+TPCX_BB_HOME=${TPCX_BB_HOME:="/home/$USERNAME/shared/tpcx-bb"}
+CONDA_ENV_NAME=${CONDA_ENV_NAME:="rapids-tpcx-bb"}
+CONDA_ENV_PATH=${CONDA_ENV_PATH:="/home/$USERNAME/conda/etc/profile.d/conda.sh"}
 
 # TODO: Unify interface/IP setting/getting for cluster startup
 # and scheduler file
-INTERFACE="ib0"
+INTERFACE=${INTERFACE:="ib0"}
 
 # TODO: Remove hard-coding of scheduler
-SCHEDULER=$(hostname)
-SCHEDULER_FILE=$TPCX_BB_HOME/tpcx_bb/cluster_configuration/example-cluster-scheduler.json
-LOGDIR="/tmp/tpcx-bb-dask-logs/"
-WORKER_DIR="/tmp/tpcx-bb-dask-workers/"
+SCHEDULER=${SCHEDULER:=$(hostname)}
+SCHEDULER_FILE=${SCHEDULER_FILE:="${TPCX_BB_HOME}/tpcx_bb/cluster_configuration/example-cluster-scheduler.json"}
+LOGDIR=${LOGDIR:="/tmp/tpcx-bb-dask-logs/"}
+WORKER_DIR=${WORKER_DIR:="/tmp/tpcx-bb-dask-workers/"}
 
 # Purge Dask worker and log directories
 rm -rf $LOGDIR/*
@@ -35,10 +39,10 @@ source $CONDA_ENV_PATH
 conda activate $CONDA_ENV_NAME
 
 # Dask/distributed configuration
-export DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT="100s"
-export DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP="600s"
-export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN="1s"
-export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX="60s"
+export DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT=${DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT:="100s"}
+export DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP=${DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP:="600s"}
+export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN=${DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN:="1s"}
+export DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX=${DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX:="60s"}
 
 
 # Setup scheduler
@@ -53,11 +57,10 @@ if [ "$HOSTNAME" = $SCHEDULER ]; then
 fi
 
 
+unset TRANSPORT_ARGS
 # Setup workers
 if [ "$CLUSTER_MODE" = "NVLINK" ]; then
-        dask-cuda-worker --device-memory-limit $DEVICE_MEMORY_LIMIT --local-directory $WORKER_DIR  --rmm-pool-size=$POOL_SIZE --memory-limit=$MAX_SYSTEM_MEMORY --enable-tcp-over-ucx --enable-nvlink  --disable-infiniband --scheduler-file $SCHEDULER_FILE >> $LOGDIR/worker.log 2>&1 &
+    TRANSPORT_ARGS="--enable-tcp-over-ucx --enable-nvlink  --disable-infiniband"
 fi
 
-if [ "$CLUSTER_MODE" = "TCP" ]; then
-    dask-cuda-worker --device-memory-limit $DEVICE_MEMORY_LIMIT --local-directory $WORKER_DIR  --rmm-pool-size=$POOL_SIZE --memory-limit=$MAX_SYSTEM_MEMORY --scheduler-file $SCHEDULER_FILE >> $LOGDIR/worker.log 2>&1 &
-fi
+dask-cuda-worker --device-memory-limit $DEVICE_MEMORY_LIMIT --local-directory $WORKER_DIR  --rmm-pool-size=$POOL_SIZE --memory-limit=$MAX_SYSTEM_MEMORY ${TRANSPORT_ARGS} --scheduler-file $SCHEDULER_FILE >> $LOGDIR/worker.log 2>&1 &
diff --git a/tpcx_bb/queries/load_test/tpcx_bb_load_test.py b/tpcx_bb/queries/load_test/tpcx_bb_load_test.py
index 4651d6f0..dca605ee 100644
--- a/tpcx_bb/queries/load_test/tpcx_bb_load_test.py
+++ b/tpcx_bb/queries/load_test/tpcx_bb_load_test.py
@@ -1,11 +1,7 @@
 from xbb_tools.utils import benchmark, tpcxbb_argparser, run_query
 from xbb_tools.readers import build_reader
 import os, subprocess, math, time
-
-
-config["data_dir"] = "/".join(config["data_dir"].rstrip("/").split("/")[:-1])
-
-spark_schema_dir = f"{os.getcwd()}/../../spark_table_schemas/"
+from xbb_tools.config import get_config
 
 # these tables have extra data produced by bigbench dataGen
 refresh_tables = [
@@ -21,15 +17,13 @@
     "web_returns",
     "web_sales",
 ]
-tables = [table.split(".")[0] for table in os.listdir(spark_schema_dir)]
 
-scale = [x for x in config["data_dir"].split("/") if "sf" in x][0]
-part_size = 3
-chunksize = "128 MiB"
+def get_tables( spark_schema_dir ):
+    return [table.split(".")[0] for table in os.listdir(spark_schema_dir)]
 
 # Spark uses different names for column types, and RAPIDS doesn't yet support Decimal types.
-def get_schema(table):
-    with open(f"{spark_schema_dir}{table}.schema") as fp:
+def get_schema(table, schema_dir):
+    with open(os.path.join(schema_dir,f"{table}.schema")) as fp:
         schema = fp.read()
         names = [line.replace(",", "").split()[0] for line in schema.split("\n")]
         types = [
@@ -45,12 +39,9 @@ def get_schema(table):
         return names, types
 
 
-def read_csv_table(table, chunksize="256 MiB"):
-    # build dict of dtypes to use when reading CSV
-    names, types = get_schema(table)
-    dtype = {names[i]: types[i] for i in range(0, len(names))}
-
-    data_dir = config["data_dir"]
+def read_csv_table(table, data_dir, schema_dir, chunksize="256 MiB"):
+    names, types = get_schema(table, schema_dir)
+    dtype=dict(zip(names, types))
     base_path = f"{data_dir}/data/{table}"
     files = os.listdir(base_path)
     # item_marketprices has "audit" files that should be excluded
@@ -99,26 +90,26 @@ def multiplier(unit):
 
 
 # we use size of the CSV data on disk to determine number of Parquet partitions
-def get_size_gb(table):
-    data_dir = config["data_dir"]
-    path = data_dir + "/data/" + table
-    size = subprocess.check_output(["du", "-sh", path]).split()[0].decode("utf-8")
+def get_size_gb(table, data_dir):
+    table_path=os.path.join( data_dir, 'data', table)
+    print( f"Getting size of {table_path}")
+    size = subprocess.check_output(["du", "-sh", table_path]).split()[0].decode("utf-8")
     unit = size[-1]
 
     size = math.ceil(float(size[:-1])) * multiplier(unit)
 
     if table in refresh_tables:
-        path = data_dir + "/data_refresh/" + table
+        table_path = os.path.join(data_dir,'data_refresh',table)
         refresh_size = (
-            subprocess.check_output(["du", "-sh", path]).split()[0].decode("utf-8")
+            subprocess.check_output(["du", "-sh", table_path]).split()[0].decode("utf-8")
         )
         size = size + math.ceil(float(refresh_size[:-1])) * multiplier(refresh_size[-1])
 
     return size
 
 
-def repartition(table, outdir, npartitions=None, chunksize=None, compression="snappy"):
-    size = get_size_gb(table)
+def repartition(table, data_dir, schema_dir, outdir, npartitions=None, chunksize=None, compression="snappy"):
+    size = get_size_gb(table, data_dir)
     if npartitions is None:
         npartitions = max(1, size)
 
@@ -128,38 +119,39 @@ def repartition(table, outdir, npartitions=None, chunksize=None, compression="sn
     # web_clickstreams is particularly memory intensive
     # we sacrifice a bit of speed for stability, converting half at a time
     if table in ["web_clickstreams"]:
-        df = read_csv_table(table, chunksize)
-        half = int(df.npartitions / 2)
-        df.partitions[0:half].repartition(npartitions=int(npartitions / 2)).to_parquet(
+        df = read_csv_table(table, data_dir, schema_dir, chunksize)
+        half = max(1,int(df.npartitions / 2))
+        df.partitions[0:half].repartition(npartitions=max(1,int(npartitions / 2))).to_parquet(
             outdir + table, compression=compression
         )
         print("Completed first half of web_clickstreams..")
-        df.partitions[half:].repartition(npartitions=int(npartitions / 2)).to_parquet(
+        df.partitions[half:].repartition(npartitions=max(1,int(npartitions / 2))).to_parquet(
             outdir + table, compression=compression
         )
 
     else:
-        read_csv_table(table, chunksize).repartition(
+        read_csv_table(table, data_dir, schema_dir, chunksize).repartition(
             npartitions=npartitions
         ).to_parquet(outdir + table, compression=compression)
 
 
 def main(client, config):
     # location you want to write Parquet versions of the table data
-    data_dir = "/".join(config["data_dir"].split("/")[:-1])
-    outdir = f"{data_dir}/parquet_{part_size}gb/"
-
-    total = 0
-    for table in tables:
-        size_gb = get_size_gb(table)
+    data_dir = config.get("data_dir",'.')
+    outdir = f"{config.get('output_dir',data_dir)}/parquet_{config.get('partitions',3)}gb/"
+    schema_dir= config.get('spark_schema_dir',
+                           os.path.join(os.getcwd(),'..','..','spark_table_schemas'))
+    began = time.time()
+    for table in get_tables(schema_dir):
+        size_gb = get_size_gb(table, data_dir)
         # product_reviews has lengthy strings which exceed cudf's max number of characters per column
         # we use smaller partitions to avoid overflowing this character limit
         if table == "product_reviews":
             npartitions = max(1, int(size_gb / 1))
         else:
-            npartitions = max(1, int(size_gb / part_size))
-        repartition(table, outdir, npartitions, chunksize, compression="snappy")
-    print(f"{chunksize} took {total}s")
+            npartitions = max(1, int(size_gb / config.get('partitions',3)))
+        repartition(table, data_dir, schema_dir, outdir, npartitions, config.get('chunk_size',"128 MiB"), compression="snappy")
+    print(f"{config.get('chunk_size','128 MiB')} took {time.time() - began}s")
     return cudf.DataFrame()
 
 
@@ -168,6 +160,6 @@ def main(client, config):
     import cudf
     import dask_cudf
 
-    config = tpcxbb_argparser()
-    client, bc = attach_to_cluster(config)
-    run_query(config=config, client=client, query_func=main)
+    conf = tpcxbb_argparser()
+    client, bc = attach_to_cluster(conf)
+    run_query(config=conf, client=client, query_func=main)
diff --git a/tpcx_bb/queries/q02/tpcx_bb_query_02.py b/tpcx_bb/queries/q02/tpcx_bb_query_02.py
index 4ce81c35..8c33d100 100644
--- a/tpcx_bb/queries/q02/tpcx_bb_query_02.py
+++ b/tpcx_bb/queries/q02/tpcx_bb_query_02.py
@@ -115,7 +115,7 @@ def main(client, config):
     # AND   wcs_user_sk IS NOT NULL
 
     f_wcs_df = wcs_df.map_partitions(pre_repartition_task)
-    f_wcs_df = f_wcs_df.repartition(columns=["wcs_user_sk"])
+    f_wcs_df = f_wcs_df.shuffle(on=["wcs_user_sk"])
 
     ### Main Query
     # SELECT
diff --git a/tpcx_bb/queries/q02/tpcx_bb_query_02_sql.py b/tpcx_bb/queries/q02/tpcx_bb_query_02_sql.py
index 234baf0b..895a22d2 100644
--- a/tpcx_bb/queries/q02/tpcx_bb_query_02_sql.py
+++ b/tpcx_bb/queries/q02/tpcx_bb_query_02_sql.py
@@ -44,13 +44,13 @@ def main(data_dir, client, bc, config):
 
     query_1 = """
         SELECT
-            CAST( wcs_user_sk AS INTEGER) AS wcs_user_sk,
-            CAST( wcs_item_sk AS INTEGER) AS wcs_item_sk,
+            CAST(wcs_user_sk AS INTEGER) AS wcs_user_sk,
+            CAST(wcs_item_sk AS INTEGER) AS wcs_item_sk,
             (wcs_click_date_sk * 86400 + wcs_click_time_sk) AS tstamp_inSec
         FROM web_clickstreams
         WHERE wcs_item_sk IS NOT NULL
         AND   wcs_user_sk IS NOT NULL
-        ORDER BY wcs_user_sk, tstamp_inSec
+        ORDER BY wcs_user_sk
     """
     wcs_result = bc.sql(query_1)
 
diff --git a/tpcx_bb/queries/q03/tpcx_bb_query_03.py b/tpcx_bb/queries/q03/tpcx_bb_query_03.py
index 9ea7931d..3ad5233e 100644
--- a/tpcx_bb/queries/q03/tpcx_bb_query_03.py
+++ b/tpcx_bb/queries/q03/tpcx_bb_query_03.py
@@ -15,6 +15,7 @@
 #
 
 import sys
+import os
 
 
 from xbb_tools.utils import (
@@ -43,7 +44,7 @@ def get_wcs_minima(config):
     import dask_cudf
 
     wcs_df = dask_cudf.read_parquet(
-        config["data_dir"] + "web_clickstreams/*.parquet",
+        os.path.join(config["data_dir"], "web_clickstreams/*.parquet"),
         columns=["wcs_click_date_sk", "wcs_click_time_sk"],
     )
 
@@ -234,7 +235,7 @@ def main(client, config):
     ### Below Pr has the dashboard snapshot which makes the problem clear
     ### https://github.com/rapidsai/tpcx-bb-internal/pull/496#issue-399946141
 
-    web_clickstream_flist = glob.glob(config["data_dir"] + "web_clickstreams/*.parquet")
+    web_clickstream_flist = glob.glob(os.path.join(config["data_dir"], "web_clickstreams/*.parquet"))
     task_ls = [
         delayed(pre_repartition_task)(fn, item_df.to_delayed()[0], wcs_tstamp_min)
         for fn in web_clickstream_flist
@@ -251,7 +252,7 @@ def main(client, config):
 
     merged_df = dask_cudf.from_delayed(task_ls, meta=meta_df)
 
-    merged_df = merged_df.repartition(columns="wcs_user_sk")
+    merged_df = merged_df.shuffle(on="wcs_user_sk")
 
     meta_d = {
         "i_item_sk": np.ones(1, dtype=merged_df["wcs_item_sk"].dtype),
diff --git a/tpcx_bb/queries/q04/tpcx_bb_query_04.py b/tpcx_bb/queries/q04/tpcx_bb_query_04.py
index f90caaae..be39641a 100644
--- a/tpcx_bb/queries/q04/tpcx_bb_query_04.py
+++ b/tpcx_bb/queries/q04/tpcx_bb_query_04.py
@@ -131,7 +131,7 @@ def main(client, config):
     keep_cols = ["wcs_user_sk", "tstamp_inSec", "wcs_web_page_sk"]
     f_wcs_df = f_wcs_df[keep_cols]
 
-    f_wcs_df = f_wcs_df.repartition(columns=["wcs_user_sk"])
+    f_wcs_df = f_wcs_df.shuffle(on=["wcs_user_sk"])
 
     # Convert wp_type to categorical and get cat_id of review and dynamic type
     wp["wp_type"] = wp["wp_type"].map_partitions(lambda ser: ser.astype("category"))
diff --git a/tpcx_bb/queries/q08/tpcx_bb_query_08.py b/tpcx_bb/queries/q08/tpcx_bb_query_08.py
index fe7761dd..075f6a67 100644
--- a/tpcx_bb/queries/q08/tpcx_bb_query_08.py
+++ b/tpcx_bb/queries/q08/tpcx_bb_query_08.py
@@ -269,7 +269,7 @@ def main(client, config):
     meta_df = cudf.DataFrame(meta_d)
     merged_df = dask_cudf.from_delayed(task_ls, meta=meta_df)
 
-    merged_df = merged_df.repartition(columns=["wcs_user_sk"])
+    merged_df = merged_df.shuffle(on=["wcs_user_sk"])
     reviewed_sales = merged_df.map_partitions(
         reduction_function,
         REVIEW_CAT_CODE,
diff --git a/tpcx_bb/queries/q08/tpcx_bb_query_08_sql.py b/tpcx_bb/queries/q08/tpcx_bb_query_08_sql.py
index d1d166d4..894ccfb7 100644
--- a/tpcx_bb/queries/q08/tpcx_bb_query_08_sql.py
+++ b/tpcx_bb/queries/q08/tpcx_bb_query_08_sql.py
@@ -210,7 +210,7 @@ def main(data_dir, client, bc, config):
     bc.drop_table("web_page_2")
     del web_page_df
 
-    merged_df = merged_df.repartition(columns=["wcs_user_sk"])
+    merged_df = merged_df.shuffle(on=["wcs_user_sk"])
     merged_df["review_flag"] = merged_df.wp_type_codes == REVIEW_CAT_CODE
 
     prepped = merged_df.map_partitions(
diff --git a/tpcx_bb/queries/q16/tpcx_bb_query_16.py b/tpcx_bb/queries/q16/tpcx_bb_query_16.py
index 1243abba..ae3bc24c 100644
--- a/tpcx_bb/queries/q16/tpcx_bb_query_16.py
+++ b/tpcx_bb/queries/q16/tpcx_bb_query_16.py
@@ -236,7 +236,7 @@ def main(client, config):
     ## group by logic
     group_cols = ["w_state_code", "i_item_id_code"]
 
-    agg_df = sales_before_after_df.groupby(group_cols).agg(
+    agg_df = sales_before_after_df.groupby(group_cols, sort=True).agg(
         {"sales_before": "sum", "sales_after": "sum"}
     )
     agg_df = agg_df.reset_index(drop=False)
diff --git a/tpcx_bb/queries/q18/tpcx_bb_query_18.py b/tpcx_bb/queries/q18/tpcx_bb_query_18.py
index a47e1729..7ae35367 100644
--- a/tpcx_bb/queries/q18/tpcx_bb_query_18.py
+++ b/tpcx_bb/queries/q18/tpcx_bb_query_18.py
@@ -117,7 +117,7 @@ def find_targets_in_reviews_helper(ddf, targets, str_col_name="pr_review_content
     lowered = ddf[str_col_name].str.lower()
 
     ## TODO: Do the replace/any in cupy land before going to cuDF
-    resdf = cudf.DataFrame.from_gpu_matrix(
+    resdf = cudf.DataFrame(
         cp.asarray(
             find_multiple.find_multiple(lowered._column, targets._column)
         ).reshape(-1, len(targets))
diff --git a/tpcx_bb/queries/q18/tpcx_bb_query_18_sql.py b/tpcx_bb/queries/q18/tpcx_bb_query_18_sql.py
index fa796108..1e87693b 100644
--- a/tpcx_bb/queries/q18/tpcx_bb_query_18_sql.py
+++ b/tpcx_bb/queries/q18/tpcx_bb_query_18_sql.py
@@ -82,7 +82,7 @@ def find_targets_in_reviews_helper(ddf, targets, str_col_name="pr_review_content
     lowered = ddf[str_col_name].str.lower()
 
     ## TODO: Do the replace/any in cupy land before going to cuDF
-    resdf = cudf.DataFrame.from_gpu_matrix(
+    resdf = cudf.DataFrame(
         cp.asarray(
             find_multiple.find_multiple(lowered._column, targets._column)
         ).reshape(-1, len(targets))
diff --git a/tpcx_bb/queries/q20/tpcx_bb_query_20.py b/tpcx_bb/queries/q20/tpcx_bb_query_20.py
index cc1c1488..6c31184a 100644
--- a/tpcx_bb/queries/q20/tpcx_bb_query_20.py
+++ b/tpcx_bb/queries/q20/tpcx_bb_query_20.py
@@ -121,7 +121,7 @@ def main(client, config):
     unique_sales = store_sales_df[
         ["ss_ticket_number", "ss_customer_sk"]
     ].map_partitions(lambda df: df.drop_duplicates())
-    unique_sales = unique_sales.repartition(columns=["ss_customer_sk"])
+    unique_sales = unique_sales.shuffle(on=["ss_customer_sk"])
     unique_sales = unique_sales.map_partitions(lambda df: df.drop_duplicates())
 
     unique_sales = unique_sales.persist()
diff --git a/tpcx_bb/queries/q22/tpcx_bb_query_22.py b/tpcx_bb/queries/q22/tpcx_bb_query_22.py
index 45274786..82686ad7 100644
--- a/tpcx_bb/queries/q22/tpcx_bb_query_22.py
+++ b/tpcx_bb/queries/q22/tpcx_bb_query_22.py
@@ -133,7 +133,7 @@ def main(client, config):
     output_table = output_table[keep_columns]
 
     output_table = (
-        output_table.groupby(by=["w_warehouse_name", "i_item_id"])
+        output_table.groupby(by=["w_warehouse_name", "i_item_id"], sort=True)
         .agg({"inv_before": "sum", "inv_after": "sum"})
         .reset_index()
     )
diff --git a/tpcx_bb/queries/q25/tpcx_bb_query_25.py b/tpcx_bb/queries/q25/tpcx_bb_query_25.py
index 942a32b6..6c238955 100644
--- a/tpcx_bb/queries/q25/tpcx_bb_query_25.py
+++ b/tpcx_bb/queries/q25/tpcx_bb_query_25.py
@@ -71,7 +71,7 @@ def agg_count_distinct(df, group_key, counted_key, client):
     unique_df = df[[group_key, counted_key]].map_partitions(
         lambda df: df.drop_duplicates()
     )
-    unique_df = unique_df.repartition(columns=[group_key])
+    unique_df = unique_df.shuffle(on=[group_key])
     unique_df = unique_df.map_partitions(lambda df: df.drop_duplicates())
 
     return unique_df.groupby(group_key)[counted_key].count(split_every=2)
@@ -93,7 +93,9 @@ def get_clusters(client, ml_input_df):
     )
     output["label"] = labels_final.reset_index()[0]
 
-    # Based on CDH6.1 q25-result formatting
+    # Sort based on CDH6.1 q25-result formatting
+    output = output.sort_values(["cid"])
+
     results_dict["cid_labels"] = output
     return results_dict
 
diff --git a/tpcx_bb/queries/q26/tpcx_bb_query_26.py b/tpcx_bb/queries/q26/tpcx_bb_query_26.py
index 2ecabbce..28b202a0 100644
--- a/tpcx_bb/queries/q26/tpcx_bb_query_26.py
+++ b/tpcx_bb/queries/q26/tpcx_bb_query_26.py
@@ -81,6 +81,9 @@ def get_clusters(client, kmeans_input_df):
     )
     output["label"] = labels_final.reset_index()[0]
 
+    # Sort based on CDH6.1 q26-result formatting
+    output = output.sort_values(["ss_customer_sk"])
+
     # Based on CDH6.1 q26-result formatting
     results_dict["cid_labels"] = output
     return results_dict
diff --git a/tpcx_bb/queries/q29/tpcx_bb_query_29.py b/tpcx_bb/queries/q29/tpcx_bb_query_29.py
index 755d3808..2b6fc742 100644
--- a/tpcx_bb/queries/q29/tpcx_bb_query_29.py
+++ b/tpcx_bb/queries/q29/tpcx_bb_query_29.py
@@ -101,7 +101,7 @@ def main(client, config):
         dask_profile=config["dask_profile"],
     )
     ### setting index on ws_order_number
-    ws_df = ws_df.repartition(columns=["ws_order_number"])
+    ws_df = ws_df.shuffle(on=["ws_order_number"])
     ### at sf-100k we will have max of 17M rows and 17 M rows with 2 columns, 1 part is very reasonable
     item_df = item_df.repartition(npartitions=1)
 
diff --git a/tpcx_bb/queries/q30/tpcx_bb_query_30.py b/tpcx_bb/queries/q30/tpcx_bb_query_30.py
index af33ee0f..f000cd38 100644
--- a/tpcx_bb/queries/q30/tpcx_bb_query_30.py
+++ b/tpcx_bb/queries/q30/tpcx_bb_query_30.py
@@ -128,7 +128,7 @@ def main(client, config):
     merged_df = dask_cudf.from_delayed(task_ls, meta=meta_df)
 
     ### that the click for each user ends up at the same partition
-    merged_df = merged_df.repartition(columns=["wcs_user_sk"])
+    merged_df = merged_df.shuffle(on=["wcs_user_sk"])
 
     ### Main Query
     ### sessionize logic.
diff --git a/tpcx_bb/setup.py b/tpcx_bb/setup.py
index 4f665e25..e5689765 100644
--- a/tpcx_bb/setup.py
+++ b/tpcx_bb/setup.py
@@ -1,13 +1,34 @@
 # Copyright (c) 2020, NVIDIA CORPORATION.
 from setuptools import find_packages, setup
+import os
+
+requirements = [
+    "dask", "cudf", "dask_cudf", "cupy", "pandas", "requests", "rmm", "pynvml"
+]
 
 qnums = [str(i).zfill(2) for i in range(1, 31)]
 
+package_data={'':['*.json','xbb_tools/*.json'],
+              "benchmark_runner": ["benchmark_config.yaml"] }
+
+packages=['xbb_tools']
+
+for root, dir, files in os.walk( 'xbb_tools' ):
+    packages.append(root.replace(os.path.sep,'.'))
+
+
 setup(
-    name="xbb_tools",
-    version="0.2",
-    author="RAPIDS",
+    name='xbb_tools',
+    version='0.2',
+    author='RAPIDS',
     packages=["benchmark_runner", "xbb_tools"],
-    package_data={"benchmark_runner": ["benchmark_config.yaml"]},
+    package_data=package_data,
+    entry_points={
+        "console_scripts": [
+            "daskcluster=xbb_tools.daskcluster:cli",
+            "benchmark_runner=benchmark_runner:main"
+        ]
+    },
     include_package_data=True,
+    install_requires=requirements
 )
diff --git a/tpcx_bb/xbb_tools/cluster_startup.py b/tpcx_bb/xbb_tools/cluster_startup.py
index 637e6fb3..24db6387 100644
--- a/tpcx_bb/xbb_tools/cluster_startup.py
+++ b/tpcx_bb/xbb_tools/cluster_startup.py
@@ -22,7 +22,7 @@
 import dask
 from dask.distributed import Client
 from dask.utils import parse_bytes
-from blazingsql import BlazingContext
+#from blazingsql import BlazingContext
 
 
 def get_config_options():
@@ -60,10 +60,20 @@ def attach_to_cluster(config, create_blazing_context=False):
 
     Optionally, this will also create a BlazingContext.
     """
+    scheduler_file = config.get("scheduler_file")
     host = config.get("cluster_host")
     port = config.get("cluster_port", "8786")
+    
+    if scheduler_file is not None:
+        try:
+            with open(scheduler_file) as fp:
+                print(fp.read())
+            client = Client(scheduler_file=scheduler_file)
+            print('Connected!')
+        except OSError as e:
+            sys.exit(f"Unable to create a Dask Client connection: {e}")
 
-    if host is not None:
+    elif host is not None:
         try:
             content = requests.get(
                 "http://" + host + ":8787/info/main/workers.html"
@@ -80,7 +90,7 @@ def attach_to_cluster(config, create_blazing_context=False):
             sys.exit(f"Unable to create a Dask Client connection: {e}")
 
     else:
-        raise ValueError("Must pass a cluster address to the host argument.")
+        raise ValueError("Must pass a scheduler file or cluster address to the host argument.")
 
     def maybe_create_worker_directories(dask_worker):
         worker_dir = dask_worker.local_directory
@@ -109,6 +119,7 @@ def maybe_create_worker_directories(dask_worker):
     config["40GB_workers"] = worker_counts["40GB"]
 
     bc = None
+    create_blazing_context = False
     if create_blazing_context:
         bc = BlazingContext(
             dask_client=client,
@@ -131,7 +142,7 @@ def worker_count_info(client, gpu_sizes=["16GB", "32GB", "40GB"], tol="2.1GB"):
     worker_info = client.scheduler_info()["workers"]
     for worker, info in worker_info.items():
         # Assumption is that a node is homogeneous (on a specific node all gpus have the same size)
-        worker_device_memory = info["gpu"]["memory-total"][0]
+        worker_device_memory = info["gpu"]["memory-total"]
         for gpu_size in gpu_sizes:
             if abs(parse_bytes(gpu_size) - worker_device_memory) < parse_bytes(tol):
                 counts_by_gpu_size[gpu_size] += 1
@@ -169,4 +180,4 @@ def import_query_libs():
         "blazingsql",
     ]
     for lib in library_list:
-        importlib.import_module(lib)
+        importlib.import_module(lib)
\ No newline at end of file
diff --git a/tpcx_bb/xbb_tools/config.py b/tpcx_bb/xbb_tools/config.py
new file mode 100644
index 00000000..d1afa427
--- /dev/null
+++ b/tpcx_bb/xbb_tools/config.py
@@ -0,0 +1,215 @@
+#
+# Copyright (c) 2020, NVIDIA CORPORATION.
+#
+# Licensed 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.
+
+import os
+import copy
+import json
+import pkgutil
+import logging
+
+from argparse import ArgumentParser
+
+
+class Config(dict):
+
+    _default_config=None
+    _packagename=__module__.split('.')[0].replace('_','')
+    _prefix='{}_'.format(_packagename.upper())
+    _default_file_name='{}.config.json'.format(_packagename)
+
+    __doc__="""Config module takes parameters from command line, json files, environment, and the default config json in this pacakge and
+presents a config object with dict interface and direct accessors.  Default config is overidden by config file is overidden
+by environment is overidden by command line and c'tor args.  Environment will pick up any environment variables prefixed with
+"{}" Prefix can be overidden in module get_config method.
+    """.format( _prefix )
+
+    def __init__( self, fname=None, **kwargs ):
+        if not fname:
+            fname = os.path.join( os.getcwd(), self._default_file_name )
+        self.update( self.make_config( self.get_default_config() ))
+
+        if os.path.exists( fname ):
+            self.update(self.parse_config_file( fname))
+        else:
+            logging.getLogger().warning( "Unable to find {}".format(fname))
+        self.update(kwargs)
+
+    def parse_config_file( self, fname ):
+        assert os.path.exists(fname)
+        if fname.lower().endswith('yaml'):
+            return {}
+        elif fname.lower().endswith( 'json'):
+            return json.load(open( fname, 'r'))
+        else:
+            raise Exception( "unsupported configfile format, {}".format( fname))
+
+    def build_argparser( self, description="Argparser generated from config module" ):
+        parser = ArgumentParser(description=description,
+                                conflict_handler='resolve')
+        #TODO: eval should filter for poison inputs                                                                                                                                                             
+        for d in Config.get_default_config():
+            name="--{}".format(d.pop("name"))
+            if 'default' in d:
+                d['default']=self.eval_default_value(d)
+            if 'type' in d:
+                d.pop('type')
+            parser.add_argument( name, **d)
+        parser.add_argument(
+            '-c', '--config',
+            default=os.path.join( os.getcwd(),'.'.join([self._prefix.lower()]+["config.json"])),
+            action='store', dest='configfile',
+            help=self.__doc__    )
+
+        return parser
+
+    def override( self, kvdict, skipdefaults=False ):
+        defaults={}
+        ff=lambda x: str(defaults.get(x,'__'+str(kvdict[x]))) != str(kvdict[x])
+        if skipdefaults:
+            defaults=self.make_config( self.get_default_config() )
+        if kvdict:
+            self.update( {k:kvdict[k] for k in filter( ff, kvdict )} )
+        return self
+
+    def as_dict( self ):
+        return copy.deepcopy( self )
+
+    def as_environment_dict(self):
+        return { self._prefix + k:str(v) for k,v in self.items() }
+
+    def __getattr__(self, name):
+        if name in self:
+            if isinstance( self.get(name), dict):
+                c=Config()
+                c._data=self.get(name)
+                return c
+            else:
+                return self.get(name)
+        else:
+            raise AttributeError( "Attribute {} does not exist in {}".format( name, self.__class__ ))
+
+    @classmethod
+    def eval_default_value( clz, elem ):
+        defaulttype=elem.get('type','str').strip()
+        if 'type' == eval( "type({}).__name__".format( defaulttype)):
+            return eval(defaulttype)( elem.get('default'))
+        else:
+            return elem.get('default',None)
+
+    @classmethod
+    def make_config( clz, spec):
+        return { d['name']:clz.eval_default_value(d) for d in spec }
+
+    @classmethod
+    def get_default_config( clz ):
+        if not clz._default_config:
+            pkg=clz.__module__.split('.')[0]
+            data=pkgutil.get_data( pkg, 'default.config.json')
+            clz._default_config=json.loads(data.decode('UTF-8'))
+
+        return copy.deepcopy( clz._default_config )
+
+
+def get_config( conf={}, fname=None, envprefix=Config._prefix):
+    env={k.replace(envprefix,'').lower():os.getenv(k) for k in
+         filter( lambda x: x.startswith( envprefix), os.environ )}
+    fname=fname if fname and os.path.exists(fname) else conf.get( 'configfile', env.get('configfile'))
+    return Config( fname ).override( env ).override( conf, skipdefaults=True )
+
+
+class TestConfig:
+    def test_get_config_is_config_instance(self):
+        assert type(get_config()) == type(Config())
+
+    def test_config_field_as_property(self):
+        it=get_config( conf={"fish":"trout"} )
+        assert it.fish == "trout"
+
+    def test_environment_export(self):
+        it=get_config( conf={"fish":"trout"} )
+        for k,v in it.as_environment_dict().items():
+            os.environ[k]=str(v)
+        assert( get_config().fish == 'trout' )
+
+    def test_overrides_configfile_with_environment( self, tmpdir ):
+        fname="test_overrides_environment_options_with_configfile.json"
+        os.environ["{}fish".format(Config._prefix)]="trout"
+        ffix = tmpdir.join(fname).write('{"fish":"perch"}')
+        it=get_config( fname=os.path.join( tmpdir, fname))
+        assert it.fish == "trout"
+
+    def test_overrides_default_with_file(self, tmpdir):
+        fname="test_overrides_environment_options_with_configfile.json"
+        ffix = tmpdir.join(fname).write('{"cluster_port":"44"}')
+        it=Config( fname=os.path.join( tmpdir, fname) )
+        assert it.cluster_port == "44"
+
+    def test_overrides_default_with_kvpairs(self, tmpdir):
+        it=get_config( {'cluster_port':'55'})
+        assert it.cluster_port == "55"
+
+    def test_overrides_default_with_environment(self, tmpdir):
+        os.environ["{}cluster_port".format(Config._prefix)]="trout"
+        it=get_config()
+        os.environ.pop("{}cluster_port".format(Config._prefix))
+        assert it.cluster_port == "trout"
+
+    def test_overrides_environment_with_nvpairs(self):
+        fname='test_overrides_configfile_with_nvpairs.json'
+        os.environ["{}fish".format(Config._prefix)]="carp"
+        it=get_config( {"fish":"bass"} )
+        assert it.fish == "bass"
+
+    def test_skipsdefault_values_from_file(self, tmpdir):
+        defaults=Config.make_config(Config.get_default_config())
+        fname="test_skipsdefault_values_from_nvpairs.json"
+        ffix = tmpdir.join(fname).write('{"cluster_port":"99"}')
+        it=get_config( fname=os.path.join( tmpdir,fname) )
+
+        assert( int(it.cluster_port) == 99)
+
+        it.override(defaults, skipdefaults=True)
+        assert( int(it.cluster_port) == 99)
+
+        it.override(defaults)
+        assert( it.cluster_port == defaults.get('cluster_port'))
+
+    def test_skipsdefault_values_from_kvpair(self, tmpdir):
+        defaults=Config.make_config(Config.get_default_config())
+        it=Config( cluster_port=33 )
+
+        assert( int(it.cluster_port) == 33)
+
+        it.override(defaults, skipdefaults=True)
+        assert( int(it.cluster_port) == 33)
+
+        it.override(defaults)
+        assert( it.cluster_port == defaults.get('cluster_port'))
+
+    #TODO: additional tests confirm all variants of this command line are supported: 
+    # add_argument(name or flags...[, action][, nargs][, const][, default][, type][, choices][, required][, help][, metavar][, dest])
+
+    def test_yamlformat(self, tmpdir):
+        fname="test_yamlformat.yaml"
+        ffix = tmpdir.join(fname).write('cluster_port: 44')
+        it=Config( fname=os.path.join( tmpdir, fname) )
+        assert it.cluster_port == "44"
+
+
+if __name__ == '__main__':
+    #run as `python -m xbb_tools.config`
+    import pytest
+    pytest.main([__file__])
+
diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py
new file mode 100644
index 00000000..a3f44ec3
--- /dev/null
+++ b/tpcx_bb/xbb_tools/daskcluster.py
@@ -0,0 +1,207 @@
+#
+# Copyright (c) 2020, NVIDIA CORPORATION.
+#
+# Licensed 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.
+
+from . import config
+from .cluster_startup import attach_to_cluster
+from .device import device_memory_limit, memory_limit, visible_devices
+
+import subprocess
+import signal
+
+import logging, os, sys, math, time
+log=logging.getLogger()
+
+
+"""
+   Thin wrapper around a thin wrapper to use args from xbb_tools.config to call
+   Start stop dask cluster for automated benchmarks.
+   * config uses environment prefix 'DASK_'
+"""
+
+def cli(commandline=None):
+    parser = config.get_config().build_argparser()
+
+    parser.add_argument(
+        'commands', nargs='*',
+        help='one or more of.. start_workers, stop_workers, start_scheduler, stop_scheduler.'
+    )
+    parser.add_argument( '--duration', action='store', default=0,
+                         help='duration to wait before killing processes started by this command and exitting' )
+    parser.usage="Start and stop, and get data from dask cluster described in config file. Commands executed in sequence.  Always put WAIT last"
+    args = vars(parser.parse_args( commandline ))
+    conf=config.get_config( args, fname=args.get('configfile'), envprefix='DASK_')
+
+    logging.basicConfig( filename=os.path.join( conf.get('logdir', os.getcwd()), f"daskcluster_{os.getpid()}.log"))
+
+    env={'CUDA_VISIBLE_DEVICES':conf.get(
+            'CUDA_VISIBLE_DEVICES',','.join([str(x) for x in visible_devices()])),
+         'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT':conf.get(
+            'distributed__comm__timeouts__connect', "100s"),
+         'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP':conf.get(
+            'distrubuted__comm__timeouts__tcp', "600s"),
+         'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN':conf.get(
+            'distributed__comm__retry__delay__min', "1s"),
+         'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX':conf.get(
+            'distributed__comm__retry__delay__max', "60s")}
+
+    for i, cmd in enumerate(conf.commands):
+        cmdf = cmd.upper().strip().replace('-','_')
+        if cmdf == 'START_WORKERS':
+            start_workers( conf, env )
+        elif cmdf == 'START_SCHEDULER':
+            start_scheduler( conf, env )
+        elif cmdf == 'STOP_WORKERS':
+            stop_workers( conf, env )
+        elif cmdf == 'STOP_SCHEDULER':
+            stop_scheduler( conf, env )
+        elif cmdf == 'WAIT':
+            if i == len(conf.commands):
+                wait_on_pids(conf, env)
+            else:
+                conf.commands.append(cmdf)
+        elif cmdf == 'DUMP_TASK_STREAM':
+            dump_task_stream( conf, env )
+        else:
+            print( 'Unknown comand "{}"'.format( cmdf ) )
+        time.sleep(2)
+
+def start_scheduler( conf, env ):
+    clusterkeys=( 'dashboard_address', 'diagnostics_port', 'host',
+                  'interface', 'port', 'preload',
+                  'protocol', 'scheduler-file' )
+
+    if conf.cluster_mode=='NVLINK':
+        env.update({'tpcxbb_benchmark_sweep_run':str(conf.get('tpcxbb_benchmark_sweep_run',True)),
+                    'DASK_RMM__POOL_SIZE':str(conf.get('rmm__pool_size','1GB')),
+                    'DASK_UCX__CUDA_COPY':str(conf.get('ucx__cuda_copy',True)),
+                    'DASK_UCX__TCP':str(conf.get('ucx__tcp',True)),
+                    'DASK_UCX__NVLINK':str(conf.get('ucx__nvlink',True)),
+                    'DASK_UCX__INFINIBAND':str(conf.get('ucx_infiniband',False)),
+                    'DASK_UCX__RDMACM':str(conf.get('ucx__rdmacm',False)),
+                    })
+
+    args=[sys.executable, '-m', 'distributed.cli.dask_scheduler'] + [str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [f"--{k}",conf.get(k)]]
+    log.info( "Starting Scheduler with command \"{}\" with environment {}".format( ' '.join(args),
+                                                                                   ', '.join([':'.join(map(str,i)) for i in env.items()])))
+    args.append('--pid-file')
+    args.append( os.path.join( conf.get('logdir', os.getcwd()), 'scheduler_{}.pid'.format(os.getpid())))
+
+    return subprocess.Popen( args,
+                            stdout=open( os.path.join( conf.get('logdir', os.getcwd()), 'scheduler_{}.stdout.log'.format(os.getpid())), 'w'),
+                            stderr=open( os.path.join( conf.get('logdir', os.getcwd()), 'scheduler_{}.stderr.log'.format(os.getpid())), 'w'),
+                            close_fds=True, env=env, restore_signals=False, start_new_session=True)
+
+def start_workers( conf, env ):
+    nworkers=int(conf.n_workers)
+    gpu_max_mem_mb=float(device_memory_limit())/(1024**2)
+
+    device_mem_limit_mb=int(conf.get('device-memory-limit',0.8)*gpu_max_mem_mb)
+    sys_max_mem_mb=int(memory_limit()/(int(nworkers)*1024**2))
+
+    env.update({'DEVICE_MEMORY_LIMIT':f"{device_mem_limit_mb}MB",
+                'POOL_SIZE':F'{gpu_max_mem_mb}MB',
+                'LOGDIR':conf.get('log_dir', './'),
+                'WORKER_DIR':conf.get('worker_dir','./'),
+                'MAX_SYSTEM_MEMORY':f"{sys_max_mem_mb}MB"})
+
+    args=[sys.executable, '-m', 'dask_cuda.cli.dask_cuda_worker',
+          '--device-memory-limit', f"{device_mem_limit_mb}MB", '--no-reconnect',
+          '--memory-limit', conf.get('memory-limit',f"{sys_max_mem_mb}MB"),
+          '--{}-tcp-over-ucx'.format(conf.get('tcp-over-ucx','enable' if conf.cluster_mode.lower() == "NVLINK" else "disable")),
+          '--{}-infiniband'.format( conf.get('infiniband','disable' if conf.cluster_mode.lower() == "NVLINK" else "enable")),
+          '--{}-nvlink'.format( conf.get( 'nvlink', 'enable' if conf.cluster_mode.lower() == "NVLINK" else "disable")),
+          '--{}-rdmacm'.format( conf.get( 'rdmacm', 'disable'))]
+
+    if conf.with_blazing:
+        ##with blazing: RMM_POOL_SIZE is reduced from total GPU mem by half the amount that DEVICE_MEMORY_LIMIT
+        args+=['--rmm-pool-size',conf.get('rmm-pool-size', "{}MB".format(
+                                          int(gpu_max_mem_mb*(float(device_mem_limit_mb)/(2*gpu_max_mem_mb)))))]
+
+    clusterkeys=( 'diagnostics_port', 'host', 'local-directory',
+                  'interface', 'port', 'preload',
+                  'scheduler-file', 'net-devices', 'nthreads',
+                  'tls-ca-file', 'tls-cert', 'tls-key' )
+
+    args+=[str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [f"--{k}",conf.get(k)]]
+ 
+
+    visible_gpus=visible_devices()
+    scale=math.ceil(len(visible_gpus)/nworkers)
+    
+    for i in range( nworkers):
+        mygpus=list(visible_gpus*scale)[i%len(visible_gpus):
+                                        i+(nworkers*scale):
+                                        nworkers]
+        env['CUDA_VISIBLE_DEVICES']=','.join(list(map(str,mygpus)))
+        env['NVIDIA_VISIBLE_DEVICES']=env['CUDA_VISIBLE_DEVICES']
+        log.info("EXECUTE 'worker-{}: {} with env {}".format( i,' '.join( map(str,args)),
+                                                              ', '.join([':'.join(map(str,i)) for i in env.items()])))
+        args.append('--pid-file')
+        args.append( os.path.join( conf.get('logdir', os.getcwd()), 'worker_n{}_{}.pid'.format(i,os.getpid())))
+
+        pid= subprocess.Popen(args + ['--name', 'worker-{}'.format(i)],
+                             stdout=open( os.path.join( conf.get('logdir', os.getcwd()), 
+                                                       'worker{}_{}.stdout.log'.format(i,os.getpid())), 'w'),
+                             stderr=open( os.path.join( conf.get('logdir', os.getcwd()), 
+                                                       'worker{}_{}.stderr.log'.format(i,os.getpid())), 'w'),
+                             close_fds=True, env=env, restore_signals=False, start_new_session=True)
+
+
+def stop_scheduler( conf, env ):
+    try:
+        client = attach_to_cluster( conf )
+        client.shutdown()
+    except:
+        log.exception( "Failed to stop scheduler", sys.exc_info())
+
+
+def stop_workers( conf, env ):
+    try:
+        client = attach_to_cluster( conf )
+        client.cancel( client.futures())
+    except:
+        log.exception( "Failed to stop workers", sys.exc_info())
+
+
+def wait_on_pids( conf, env ):
+    starttime=time.time()
+    waitflag=True
+
+    while waitflag:
+        client = attach_to_cluster( conf )
+        time.sleep( (time.time()-starttime)/conf.get('duration',1) )
+        for pidfile in filter( lambda s: s.endswith( '{}.pid'.format(os.getpid())), 
+                               os.listdir(conf.get('logdir', os.getcwd()))):
+            pid = int(open(pidfile,'r').read())
+            if conf.get('duration',0) and starttime + conf.get('duration') > time.time():
+                os.kill(pid, signal.SIGTERM)
+                waitflag=False
+        #is pid alive?
+        #is scheduler alive?
+        #if duration, has it passed?
+
+
+def dump_task_stream( conf, env ):
+    try:
+        client = attach_to_cluster( conf )
+        client.get_task_stream()
+        raise Exception("Not yet implemented!!!")
+    except:
+        log.exception( "Failed to dump task_stream", sys.exc_info())
+
+
+if __name__ == '__main__':
+    import sys
+    cli(sys.argv[1:])
diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json
new file mode 100644
index 00000000..4ccfbdcd
--- /dev/null
+++ b/tpcx_bb/xbb_tools/default.config.json
@@ -0,0 +1,136 @@
+[
+  {
+    "name": "cluster_host",
+    "default": "localhost",
+    "type": "str",
+    "help": "Hostname to use for the cluster scheduler. If you are trying to spin up a fresh SSHCluster, please ignore this and use --hosts instead."
+  },
+  {
+    "name": "cluster_mode",
+    "default": "TCP",
+    "type": "str",
+    "help": "transport used by dask_cudf. Should be coordinated with interface. Options: IB, TCP, NVLINK"
+  },
+  {
+    "name": "cluster_port",
+    "default": 8786,
+    "type": "int",
+    "help": "Which port to use for the cluster scheduler. If you are trying to spin up a fresh SSHCluster, please ignore this and use --hosts instead."
+  },
+  {
+    "name": "cluster_type",
+    "default": "local",
+    "type": "str",
+    "help": "The type of cluster to spin up (local CUDACluster or an SSH CUDACluster). You do not need to specify the cluster_type when attaching to an already running cluster."
+  },
+  {
+    "name": "data_dir",
+    "default": "/datasets/tpcx-bb/sf1/parquet_1gb/",
+    "type": "str",
+    "help": "Data Dir"
+  },
+  {
+    "name": "device_memory_limit",
+    "type": "float",
+    "default": "0.9",
+    "help": "GPU memory limit is set ot a percent of available GPU memory"
+  },
+  {
+    "name": "dask_profile",
+    "type": "bool",
+    "help": "Include Dask Performance Report"
+  },
+  {
+    "name": "file_format",
+    "default": "parquet",
+    "type": "str",
+    "help": "help string"
+  },
+  {
+    "name": "get_read_time",
+    "default": "True",
+    "type": "bool",
+    "help": "whether or not to track read time"
+  },
+  {
+    "name": "interface",
+    "default":"eth0",
+    "type": "str",
+    "help": "network interface used by dask_cu*"
+  },
+  {
+    "name": "logdir",
+    "default": "./",
+    "type": "str",
+    "help": "directory to create log files`"
+  },
+  {
+    "name": "n_workers",
+    "default": 1,
+    "type": "int",
+    "help": "Number of worker processes to start with each call to daskcluster start_workers"
+  },
+  {
+    "name": "output_dir",
+    "default": "./",
+    "type": "str",
+    "help": "Query Output Directry. Defaults to the directory of the query script."
+  },
+  {
+    "name": "output_filetype",
+    "default": "parquet",
+    "type": "str",
+    "help": "help string"
+  },
+  {
+    "name": "repartition_small_table",
+    "default": "True",
+    "type": "str",
+    "help": "help string"
+  },
+  {
+    "name": "scheduler-file",
+    "type": "str",
+    "default": "sched.json",
+    "help": "File to write connection information."
+  },
+  {
+    "name": "sheet",
+    "default": "TPCx-BB",
+    "type": "str",
+    "help": "help string"
+  },
+  {
+    "name": "split_row_groups",
+    "type": "bool",
+    "help": "not really sure"
+  },
+  {
+    "name": "tab",
+    "default": "SF1000 Benchmarking Matrix",
+    "type": "str",
+    "help": "help string"
+  },
+  {
+    "name": "timeout",
+    "type": "int",
+    "default": "300",
+    "help": "max duration for each query"
+  },
+  {
+    "name": "verify_dir",
+    "default": "",
+    "type": "str",
+    "help": "help string"
+  },
+  {
+    "name": "verify_results",
+    "type": "bool",
+    "help": "help string"
+  },
+  {
+    "name": "with_blazing",
+    "type": "bool",
+    "help": "If True cluster and runners setup with suppoort for BSQL"
+  }
+]
diff --git a/tpcx_bb/xbb_tools/device.py b/tpcx_bb/xbb_tools/device.py
new file mode 100644
index 00000000..fbaed818
--- /dev/null
+++ b/tpcx_bb/xbb_tools/device.py
@@ -0,0 +1,40 @@
+#
+# Copyright (c) 2020, NVIDIA CORPORATION.
+#
+# Licensed 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.
+
+import pynvml
+import psutil
+import re, os
+
+
+pynvml.nvmlInit()
+
+def visible_devices(devices='all'):
+    """returns a tuple of integers representing device indices"""
+    envvar = os.getenv( 'NVIDIA_VISIBLE_DEVICES',
+                        os.getenv('CUDA_VISIBLE_DEVICES', devices ))
+    if envvar.upper() == 'ALL':
+        return tuple(ndx for ndx in range(pynvml.nvmlDeviceGetCount()))
+    return tuple( int(ndx) if re.match( '^\d+$', ndx.strip())
+                           else pynvml.nvmlDeviceGetIndex( pynvml.nvmlDeviceGetHandleByUUID(ndx.encode('UTF-8')))
+                           for ndx in re.split('[, ]+',envvar))
+
+
+def device_memory_limit( devices=[] ):
+    return min( [pynvml.nvmlDeviceGetMemoryInfo(
+                    pynvml.nvmlDeviceGetHandleByIndex(d)).total for d in visible_devices(','.join(devices) if devices else None)] )
+
+
+def memory_limit():
+    return psutil.virtual_memory().available
diff --git a/tpcx_bb/xbb_tools/merge_util.py b/tpcx_bb/xbb_tools/merge_util.py
index e408fe6f..934600f6 100644
--- a/tpcx_bb/xbb_tools/merge_util.py
+++ b/tpcx_bb/xbb_tools/merge_util.py
@@ -36,8 +36,8 @@ def hash_merge(
     if npartitions is None:
         npartitions = max(lhs.npartitions, rhs.npartitions)
 
-    lhs2 = lhs.repartition(columns=left_on, npartitions=npartitions)
-    rhs2 = rhs.repartition(columns=right_on, npartitions=npartitions)
+    lhs2 = lhs.shuffle(on=left_on, npartitions=npartitions)
+    rhs2 = rhs.shuffle(on=right_on, npartitions=npartitions)
 
     kwargs = dict(
         how=how,
diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py
index a7b46b24..f95d4522 100644
--- a/tpcx_bb/xbb_tools/utils.py
+++ b/tpcx_bb/xbb_tools/utils.py
@@ -45,6 +45,8 @@
 import gspread
 from oauth2client.service_account import ServiceAccountCredentials
 
+from .config import get_config
+
 #################################
 # Benchmark Timing
 #################################
@@ -124,7 +126,7 @@ def write_etl_result(df, filetype="parquet", output_directory="./"):
 
     QUERY_NUM = get_query_number()
     if filetype == "csv":
-        output_path = f"{output_directory}q{QUERY_NUM}-results.csv"
+        output_path = os.path.join(output_directory,f"q{QUERY_NUM}-results.csv")
 
         if os.path.exists(output_path):
             shutil.rmtree(output_path)
@@ -134,7 +136,7 @@ def write_etl_result(df, filetype="parquet", output_directory="./"):
 
         df.to_csv(output_path, header=True, index=False)
     else:
-        output_path = f"{output_directory}q{QUERY_NUM}-results.parquet"
+        output_path = os.path.join(output_directory,f"q{QUERY_NUM}-results.parquet")
         if os.path.exists(output_path):
             if os.path.isdir(output_path):
                 ## to remove existing  directory
@@ -148,7 +150,7 @@ def write_etl_result(df, filetype="parquet", output_directory="./"):
 
         else:
             df.to_parquet(
-                f"{output_directory}q{QUERY_NUM}-results.parquet", index=False
+                os.path.join(output_directory,f"q{QUERY_NUM}-results.parquet"), index=False
             )
 
 
@@ -178,7 +180,7 @@ def write_supervised_learning_result(result_dict, output_directory, filetype="cs
         prec = result_dict["prec"]
         cmat = result_dict["cmat"]
 
-        with open(f"{output_directory}q{QUERY_NUM}-metrics-results.txt", "w") as out:
+        with open(os.path.join(output_directory,f"q{QUERY_NUM}-metrics-results.txt"), "w") as out:
             out.write("Precision: %s\n" % prec)
             out.write("Accuracy: %s\n" % acc)
             out.write(
@@ -188,11 +190,11 @@ def write_supervised_learning_result(result_dict, output_directory, filetype="cs
 
         if filetype == "csv":
             df.to_csv(
-                f"{output_directory}q{QUERY_NUM}-results.csv", header=False, index=None
+                os.path.join(output_directory,f"q{QUERY_NUM}-results.csv"), header=False, index=None
             )
         else:
             df.to_parquet(
-                f"{output_directory}q{QUERY_NUM}-results.parquet", write_index=False
+                os.path.join(output_directory,f"q{QUERY_NUM}-results.parquet"), write_index=False
             )
 
 
@@ -205,7 +207,7 @@ def write_clustering_result(result_dict, output_directory="./", filetype="csv"):
     QUERY_NUM = get_query_number()
     clustering_info_name = f"{QUERY_NUM}-results-cluster-info.txt"
 
-    with open(f"{output_directory}q{clustering_info_name}", "w") as fh:
+    with open(os.path.join(output_directory,clustering_info_name), "w") as fh:
         fh.write("Clusters:\n\n")
         fh.write(f"Number of Clusters: {result_dict.get('nclusters')}\n")
         fh.write(f"WSSSE: {result_dict.get('wssse')}\n")
@@ -221,11 +223,11 @@ def write_clustering_result(result_dict, output_directory="./", filetype="csv"):
     if filetype == "csv":
         clustering_result_name = f"q{QUERY_NUM}-results.csv"
         data.to_csv(
-            f"{output_directory}{clustering_result_name}", index=False, header=None
+            os.path.join(output_directory,clustering_result_name), index=False, header=None
         )
     else:
         clustering_result_name = f"q{QUERY_NUM}-results.parquet"
-        data.to_parquet(f"{output_directory}{clustering_result_name}", index=False)
+        data.to_parquet(os.path.join(output_directory,clustering_result_name), index=False)
 
     return 0
 
@@ -277,7 +279,7 @@ def run_dask_cudf_query(config, client, query_func, write_func=write_result):
             query_func,
             dask_profile=config.get("dask_profile"),
             client=client,
-            config=config,
+            config=config
         )
 
         benchmark(
@@ -346,50 +348,14 @@ def run_bsql_query(
     # google sheet benchmarking automation
     push_payload_to_googlesheet(config)
 
-
-def add_empty_config(args):
-    keys = [
-        "get_read_time",
-        "split_row_groups",
-        "dask_profile",
-        "verify_results",
-    ]
-
-    for key in keys:
-        if key not in args:
-            args[key] = None
-
-    if "file_format" not in args:
-        args["file_format"] = "parquet"
-
-    if "output_filetype" not in args:
-        args["output_filetype"] = "parquet"
-
-    return args
-
-
 def tpcxbb_argparser():
-    args = get_tpcxbb_argparser_commandline_args()
-    with open(args["config_file"]) as fp:
-        args = yaml.safe_load(fp.read())
-    args = add_empty_config(args)
-
-    return args
+    return get_tpcxbb_argparser_commandline_args()
 
 
 def get_tpcxbb_argparser_commandline_args():
-    parser = argparse.ArgumentParser(description="Run TPCx-BB query")
-    print("Using default arguments")
-    parser.add_argument(
-        "--config_file",
-        default="benchmark_runner/benchmark_config.yaml",
-        type=str,
-        help="Location of benchmark configuration yaml file",
-    )
+    parser = get_config().build_argparser( description="Run TPCx-BB query")
 
-    args = parser.parse_args()
-    args = vars(args)
-    return args
+    return get_config(vars(parser.parse_args()))
 
 
 def get_scale_factor(data_dir):
@@ -408,7 +374,12 @@ def get_query_number():
         ...
     """
     QUERY_NUM = os.getcwd().split("/")[-1].strip("q")
-    return QUERY_NUM
+    if re.match('\d+', QUERY_NUM):
+        return QUERY_NUM
+    for mg in filter( bool, map( lambda s: re.search('tpcx_bb_query_(\d+)',s), sys.argv )):
+        if re.match('\d+', mg.group(1)):
+           return mg.group(1)
+    return os.getenv( 'TPCXBB_QUERY_NUM')
 
 
 #################################
@@ -455,6 +426,7 @@ def calculate_label_overlap_percent(spark_labels, rapids_labels):
     rapids_labels.columns = ["cid", "label"]
 
     # assert that we clustered the same IDs
+    rapids_labels = rapids_labels.reset_index(drop=True)
     assert spark_labels.cid.equals(rapids_labels.cid)
 
     rapids_counts_normalized = rapids_labels.label.value_counts(
@@ -622,7 +594,8 @@ def verify_results(verify_dir):
     # Setup validation data
     if QUERY_NUM in SUPERVISED_LEARNING_QUERIES:
         verify_fname = os.path.join(
-            verify_dir, f"q{QUERY_NUM}-results/q{QUERY_NUM}-metrics-results.txt"
+            verify_dir, os.path.join(f"q{QUERY_NUM}-results",
+                                     f"q{QUERY_NUM}-metrics-results.txt")
         )
         result_fname = f"q{QUERY_NUM}-metrics-results.txt"
 
@@ -667,7 +640,8 @@ def verify_results(verify_dir):
         print("Clustering Query")
         try:
             cluster_info_validation_path = os.path.join(
-                verify_dir, f"q{QUERY_NUM}-results/clustering-results.txt"
+                verify_dir, os.pathh.join(f"q{QUERY_NUM}-results",
+                                          "clustering-results.txt")
             )
             cluster_info_rapids_path = f"q{QUERY_NUM}-results-cluster-info.txt"