From b8b89f283e62a4a13b64a5e77287eab50d7de3bb Mon Sep 17 00:00:00 2001 From: Kevin German Date: Mon, 24 Aug 2020 13:58:41 -0700 Subject: [PATCH 01/51] common configuration options To enable automation command line, environment and config file options should flow through the same validation and all options should be avaailable to each interface. --- tpcx_bb/setup.py | 26 +++- tpcx_bb/xbb_tools/config.py | 215 ++++++++++++++++++++++++++ tpcx_bb/xbb_tools/default.config.json | 57 +++++++ tpcx_bb/xbb_tools/utils.py | 11 +- 4 files changed, 298 insertions(+), 11 deletions(-) create mode 100644 tpcx_bb/xbb_tools/config.py create mode 100644 tpcx_bb/xbb_tools/default.config.json diff --git a/tpcx_bb/setup.py b/tpcx_bb/setup.py index 4f665e25..a99db996 100644 --- a/tpcx_bb/setup.py +++ b/tpcx_bb/setup.py @@ -1,13 +1,33 @@ # 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"]}, + entry_points={ + "console_scripts": [ + "daskcluster=xbb_tools.daskcluster:cli" + ] + }, include_package_data=True, + install_requires=requirements ) diff --git a/tpcx_bb/xbb_tools/config.py b/tpcx_bb/xbb_tools/config.py new file mode 100644 index 00000000..51b572e9 --- /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) + d.pop('type') + parser.add_argument( name, **d) + + parser.add_argument( + '-c', '--config', default="./{}.config.json".format(self._prefix.lower()), + 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,''):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/default.config.json b/tpcx_bb/xbb_tools/default.config.json new file mode 100644 index 00000000..7e3cdac9 --- /dev/null +++ b/tpcx_bb/xbb_tools/default.config.json @@ -0,0 +1,57 @@ +[ + {"name":"data_dir", + "default":"/datasets/tpcx-bb/sf1/parquet_1gb/", + "type":"str", + "help":"Data Dir" + }, + {"name":"output_dir", + "default":"./", + "type":"str", + "help":"Query Output Directry. Defaults to the directory of the query script." + }, + {"name":"dask_dir", + "default":"./", + "type":"str", + "help":"Dask Dir" + }, + {"name":"dask_profile", + "default":"False", + "type":"bool", + "help":"Include Dask Performance Report" + }, + {"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_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":"dask_n_workers", + "default":1, + "type":"int", + "help":"Number of worker processes to start" + }, + {"name":"logdir", + "default":"./", + "type":"str", + "help":"directory to create log files`" + }, + {"name":"timeout", + "type":"int", + "default":"300", + "help":"max duration for each query" + }, + {"name":"dask_scheduler-file", + "type":"str", + "default":"sched.json", + "help":"File to write connection information." + } +] \ No newline at end of file diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index a7b46b24..982aea75 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 . import config + ################################# # Benchmark Timing ################################# @@ -378,14 +380,7 @@ def tpcxbb_argparser(): 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 = config.get_config().build_argparser( description="Run TPCx-BB query") args = parser.parse_args() args = vars(args) From 03b2b9a095716cac1a3f76ecbcc51bfdbf4a8377 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Wed, 26 Aug 2020 10:19:25 -0700 Subject: [PATCH 02/51] Dockerfile sets up tpcx-bb suite --- Dockerfile | 40 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) create mode 100644 Dockerfile diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 00000000..18b3ff35 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,40 @@ +# 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 + + +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 + +RUN groupadd --gid ${GROUP_ID} conda && \ + 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 conda /conda && \ + chmod -R g+rwx /conda + +COPY conda/rapids-tpcx-bb.yml /home/${USER_GROUP_NAME}/environment.yml +COPY tpcx_bb /home/${USER_GROUP_NAME}/tpcx_bb + +RUN /bin/bash -c "source /conda/etc/profile.d/conda.sh && \ + conda env create -f /home/${USER_GROUP_NAME}/environment.yml -n tpcxbb && \ + conda activate tpcxbb && \ + python -m pip install /home/${USER_GROUP_NAME}/tpcx_bb/." + +USER ${USER_GROUP_NAME} +WORKDIR /home/${USER_GROUP_NAME}/ From b27170f3ac427b8ba9d364d37691595f497859ff Mon Sep 17 00:00:00 2001 From: Kevin German Date: Wed, 26 Aug 2020 12:43:57 -0700 Subject: [PATCH 03/51] Patch for Blazing support Patched from richard gelhausen - https://github.com/rapidsai/tpcx-bb/commit/1a06cc6d929598eee4e4c83b664ec3ceeecfc310 --- tpcx_bb/xbb_tools/cluster_startup.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/tpcx_bb/xbb_tools/cluster_startup.py b/tpcx_bb/xbb_tools/cluster_startup.py index 637e6fb3..e4107090 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, @@ -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 From 96445659348611bc477c53b584e84e50fdc0160c Mon Sep 17 00:00:00 2001 From: Kevin German Date: Mon, 31 Aug 2020 22:30:04 -0700 Subject: [PATCH 04/51] cluster_config allow env override --- .../bsql-cluster-startup.sh | 30 +++++++-------- .../cluster_configuration/cluster-startup.sh | 37 +++++++++---------- 2 files changed, 33 insertions(+), 34 deletions(-) diff --git a/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh b/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh index ae228ea2..8809e537 100644 --- a/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh +++ b/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh @@ -1,25 +1,25 @@ #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 +35,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 diff --git a/tpcx_bb/cluster_configuration/cluster-startup.sh b/tpcx_bb/cluster_configuration/cluster-startup.sh index 72519c80..579bdcd8 100644 --- a/tpcx_bb/cluster_configuration/cluster-startup.sh +++ b/tpcx_bb/cluster_configuration/cluster-startup.sh @@ -1,25 +1,25 @@ #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 +35,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 +53,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 & From 564af41d020ccbc3b862b10f3672e7b764c0b12d Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 1 Sep 2020 18:04:32 -0700 Subject: [PATCH 05/51] make executable cluster_config --- Dockerfile | 7 ++++--- tpcx_bb/cluster_configuration/bsql-cluster-startup.sh | 10 ++++++++-- tpcx_bb/cluster_configuration/cluster-startup.sh | 4 ++++ 3 files changed, 16 insertions(+), 5 deletions(-) mode change 100644 => 100755 tpcx_bb/cluster_configuration/bsql-cluster-startup.sh mode change 100644 => 100755 tpcx_bb/cluster_configuration/cluster-startup.sh diff --git a/Dockerfile b/Dockerfile index 18b3ff35..aa333c9f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -19,19 +19,20 @@ ENV hostname=${hostname}_cuda${CUDA_VERSION}-${DISTRO} ENV DEBIAN_FRONTEND=noninteractive -RUN groupadd --gid ${GROUP_ID} conda && \ +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 conda /conda && \ + chgrp -R ${USER_GROUP_NAME} /conda && \ chmod -R g+rwx /conda COPY conda/rapids-tpcx-bb.yml /home/${USER_GROUP_NAME}/environment.yml COPY tpcx_bb /home/${USER_GROUP_NAME}/tpcx_bb -RUN /bin/bash -c "source /conda/etc/profile.d/conda.sh && \ +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 env create -f /home/${USER_GROUP_NAME}/environment.yml -n tpcxbb && \ conda activate tpcxbb && \ python -m pip install /home/${USER_GROUP_NAME}/tpcx_bb/." 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 8809e537..43e335bd --- a/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh +++ b/tpcx_bb/cluster_configuration/bsql-cluster-startup.sh @@ -1,3 +1,7 @@ +#!/usr/bin/env bash + +set -x + #IB, NVLINK, or TCP CLUSTER_MODE=${CLUSTER_MODE:="IB"} USERNAME=$(whoami) @@ -30,8 +34,10 @@ 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 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 579bdcd8..f7c83005 --- a/tpcx_bb/cluster_configuration/cluster-startup.sh +++ b/tpcx_bb/cluster_configuration/cluster-startup.sh @@ -1,3 +1,7 @@ +#!/usr/bin/env bash + +set -x + #IB, NVLINK, or TCP CLUSTER_MODE=${CLUSTER_MODE:="IB"} USERNAME=$(whoami) From 8311f4b1a3e154677f15e08f93bc362e2388cd5f Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 1 Sep 2020 21:42:54 -0700 Subject: [PATCH 06/51] python entrypoint for dask cluster mgmt --- tpcx_bb/xbb_tools/daskcluster.py | 149 +++++++++++++++++++++++++++++++ tpcx_bb/xbb_tools/device.py | 43 +++++++++ 2 files changed, 192 insertions(+) create mode 100644 tpcx_bb/xbb_tools/daskcluster.py create mode 100644 tpcx_bb/xbb_tools/device.py diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py new file mode 100644 index 00000000..39c70206 --- /dev/null +++ b/tpcx_bb/xbb_tools/daskcluster.py @@ -0,0 +1,149 @@ +# +# 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 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 as part of ASV workload. Script shoulb be callable + from the asv.conf#install_command. Configuration is loaded from xbb_tools.config. +""" + +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. More details: help-commands' + ) + + parser.usage="Start and stop, and get data from dask cluster described in config file." + args = vars(parser.parse_args( commandline )) + conf=config.get_config( args, fname=args.get('configfile') ) + + for cmd in conf.commands: + cmdf = cmd.upper().strip().replace('-','_') + if cmdf == 'START_WORKERS': + start_workers( conf ) + elif cmdf == 'START_SCHEDULER': + start_scheduler( conf ) + elif cmdf == 'STOP_WORKERS': + stop_workers( conf ) + elif cmdf == 'STOP_SCHEDULER': + stop_scheduler( conf ) + elif cmdf == 'DUMP_TASK_STREAM': + dump_task_stream( conf ) + else: + print( 'Unknown comand "{}"'.format( cmdf ) ) + time.sleep(2) + +def start_scheduler( conf ): + clusterkeys=( 'dask_dashboard_address', 'dask_diagnostics_port', 'dask_host', + 'dask_interface', 'dask_port', 'dask_preload', + 'dask_protocol', 'dask_scheduler-file' ) + + args=[str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [k.replace('dask_','--'),conf.get(k)]] + return subprocess.Popen([sys.executable, '-m', 'distributed.cli.dask_scheduler'] + 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={'tpcxbb_benchmark_sweep_run':'True', + 'DASK_RMM__POOL_SIZE':'1GB', + 'DASK_UCX__CUDA_COPY':'True', + 'DASK_UCX__TCP':'True'}, + restore_signals=False, start_new_session=True) + +def start_workers( conf ): + nworkers=int(conf.dask_n_workers) + gpu_max_mem=float(device_memory_limit()) + device_mem_limit="{}MB".format(int(float(conf.get('dask_device-memory-limit',gpu_max_mem))/(1024*1024)*.8)) + sys_max_mem="{}MB".format( int(memory_limit()/(int(nworkers)*1024*1024))) + env=os.environ.copy() + env.update({'DEVICE_MEMORY_LIMIT':device_mem_limit, + 'POOL_SIZE':'{}MB'.format( int(gpu_max_mem/(1024*1024))), + 'LOGDIR':conf.get('log_dir', './'), + 'WORKER_DIR':conf.get('dask_worker_dir','./'), + 'MAX_SYSTEM_MEMORY':sys_max_mem }) + + args=[sys.executable, '-m', 'dask_cuda.cli.dask_cuda_worker', + '--device-memory-limit', device_mem_limit, '--no-reconnect', + '--{}-tcp-over-ucx'.format(conf.get('dask_tcp-over-ucx','disable')), + '--memory-limit', conf.get('dask_memory-limit',sys_max_mem), + '--rmm-pool-size', conf.get('dask_rmm-pool-size',device_mem_limit), + '--{}-infiniband'.format( conf.get('dask_infiniband','disable')), + '--{}-nvlink'.format( conf.get( 'dask_nvlink', 'disable')), + '--{}-rdmacm'.format( conf.get( 'dask_rdmacm', 'disable'))] + + clusterkeys=( 'dask_diagnostics_port', 'dask_host', 'dask_local-directory', + 'dask_interface', 'dask_port', 'dask_preload', + 'dask_scheduler-file', 'dask_net-devices', 'dask_nthreads', + 'dask_tls-ca-file', 'dask_tls-cert', 'dask_tls-key' ) + + args+=[str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [k.replace('dask_','--'),conf.get(k)]] + #print ("EXECUTE: {}".format( ' '.join( map(str,args)))) + + 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'] + #print ("EXECUTE 'worker-{}: {} on GPUs: {}".format( i,' '.join( map(str,args)),env['CUDA_VISIBLE_DEVICES'])) + + 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) + #print( "*** WORKER {} PID: {}".format( i, pid.pid )) + +def stop_scheduler( conf ): + try: + client = attach_to_cluster( conf ) + client.shutdown() + except: + log.exception( "Failed to stop scheduler", sys.exc_info()) + + +def stop_workers( conf ): + try: + client = attach_to_cluster( conf ) + client.cancel( client.futures()) + except: + log.exception( "Failed to stop workers", sys.exc_info()) + + +def dump_task_stream( conf ): + 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/device.py b/tpcx_bb/xbb_tools/device.py new file mode 100644 index 00000000..8c2ac8d7 --- /dev/null +++ b/tpcx_bb/xbb_tools/device.py @@ -0,0 +1,43 @@ +# +# 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(): + envvar = os.getenv( 'NVIDIA_VISIBLE_DEVICES', + os.getenv('CUDA_VISIBLE_DEVICES', 'all' )) + if envvar.upper() == 'ALL': + return tuple(range(pynvml.nvmlDeviceGetCount())) + return tuple(map( int, envvar.split(','))) + + +def device_memory_limit( devices=None ): + indices=[] + if not devices: + indices=visible_devices() + else: + indices=map(int,re.split( '[, ]+', devices)) + + return min( [pynvml.nvmlDeviceGetMemoryInfo( + pynvml.nvmlDeviceGetHandleByIndex(d)).total for d in indices] ) + + +def memory_limit(): + return psutil.virtual_memory().available From 40c4d8373438c07ddceee994051083dd77499319 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 1 Sep 2020 23:57:14 -0700 Subject: [PATCH 07/51] migrate yaml options into common config --- tpcx_bb/xbb_tools/default.config.json | 53 +++++++++++++++++++++++++++ tpcx_bb/xbb_tools/utils.py | 2 - 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index 7e3cdac9..5aae6059 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -1,4 +1,57 @@ [ + {"name":"get_read_time", + "default":"True", + "type":"bool", + "help":"whether or not to track read time" + }, + {"name":"split_row_groups", + "default":"False", + "type":"bool", + "help":"not really sure" + }, + { "name":"file_format", + "default":"parquet", + "type":"str", + "help":"help string" + }, + { "name":"output_filetype", + "default":"parquet", + "type":"str", + "help":"help string" + }, + { "name":"repartition_small_table", + "default":"True", + "type":"str", + "help":"help string" + }, + { "name":"benchmark_runner_include_bsql", + "default":"False", + "type":"str", + "help":"help string" + }, + { "name":"32GB_workers", + "help":"help string" + }, + { "name":"verify_results", + "default":"False", + "type":"str", + "help":"help string" + }, + { "name":"verify_dir", + "default":"", + "type":"str", + "help":"help string" + }, + { "name":"sheet", + "default":"TPCx-BB", + "type":"str", + "help":"help string" + }, + { "name":"tab", + "default":"SF1000 Benchmarking Matrix", + "type":"str", + "help":"help string" + }, {"name":"data_dir", "default":"/datasets/tpcx-bb/sf1/parquet_1gb/", "type":"str", diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index 982aea75..8d712f63 100644 --- a/tpcx_bb/xbb_tools/utils.py +++ b/tpcx_bb/xbb_tools/utils.py @@ -372,8 +372,6 @@ def add_empty_config(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 From 92646a2f85e436550863e6855abed124ff6864b7 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Wed, 2 Sep 2020 15:56:02 -0700 Subject: [PATCH 08/51] use path relative to module file rather than cwd --- tpcx_bb/benchmark_runner.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tpcx_bb/benchmark_runner.py b/tpcx_bb/benchmark_runner.py index fa3371b2..599fc18f 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): @@ -47,8 +48,6 @@ def load_query(qnum, fn): # 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") From 098b8f30954524dc8fcf9977dcdf4046b3e0b8cc Mon Sep 17 00:00:00 2001 From: Kevin German Date: Thu, 3 Sep 2020 10:50:02 -0700 Subject: [PATCH 09/51] integrate command lines used in cluster_config scripts --- tpcx_bb/xbb_tools/daskcluster.py | 61 +++++++++++++++++++++----------- 1 file changed, 40 insertions(+), 21 deletions(-) diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py index 39c70206..547882ec 100644 --- a/tpcx_bb/xbb_tools/daskcluster.py +++ b/tpcx_bb/xbb_tools/daskcluster.py @@ -24,8 +24,7 @@ """ Thin wrapper around a thin wrapper to use args from xbb_tools.config to call - Start stop dask cluster as part of ASV workload. Script shoulb be callable - from the asv.conf#install_command. Configuration is loaded from xbb_tools.config. + Start stop dask cluster for automated benchmarks. """ def cli(commandline=None): @@ -39,24 +38,47 @@ def cli(commandline=None): parser.usage="Start and stop, and get data from dask cluster described in config file." args = vars(parser.parse_args( commandline )) conf=config.get_config( args, fname=args.get('configfile') ) + env=os.environ.copy() + env.update({ + 'tpcxbb_benchmark_sweep_run':conf.get('tpcxbb_benchmark_sweep_run', + os.getenv( 'tpcxbb_benchmark_sweep_run','True')), + 'DASK_RMM__POOL_SIZE':conf.get('DASK_RMM__POOL_SIZE', + os.getenv('DASK_RMM__POOL_SIZE','1GB')), + 'DASK_UCX__CUDA_COPY':conf.get('DASK_UCX__CUDA_COPY', + os.getenv('DASK_UCX__CUDA_COEPY','True')), + 'DASK_UCX__TCP':conf.get('DASK_UCX__TCP', + os.getenv('DASK_UCX__TCP','True')), + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT':conf.get( + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT', + os.getenv('DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT',"100s")), + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP':conf.get( + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP', os.getenv( + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP',"600s")), + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN':conf.get( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN', os.getenv( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN',"1s")), + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX':conf.get( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX', os.getenv( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX',"60s")) + }) for cmd in conf.commands: cmdf = cmd.upper().strip().replace('-','_') if cmdf == 'START_WORKERS': - start_workers( conf ) + start_workers( conf, env ) elif cmdf == 'START_SCHEDULER': - start_scheduler( conf ) + start_scheduler( conf, env ) elif cmdf == 'STOP_WORKERS': - stop_workers( conf ) + stop_workers( conf, env ) elif cmdf == 'STOP_SCHEDULER': - stop_scheduler( conf ) + stop_scheduler( conf, env ) elif cmdf == 'DUMP_TASK_STREAM': - dump_task_stream( conf ) + dump_task_stream( conf, env ) else: print( 'Unknown comand "{}"'.format( cmdf ) ) time.sleep(2) -def start_scheduler( conf ): +def start_scheduler( conf, env ): clusterkeys=( 'dask_dashboard_address', 'dask_diagnostics_port', 'dask_host', 'dask_interface', 'dask_port', 'dask_preload', 'dask_protocol', 'dask_scheduler-file' ) @@ -65,23 +87,19 @@ def start_scheduler( conf ): return subprocess.Popen([sys.executable, '-m', 'distributed.cli.dask_scheduler'] + 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={'tpcxbb_benchmark_sweep_run':'True', - 'DASK_RMM__POOL_SIZE':'1GB', - 'DASK_UCX__CUDA_COPY':'True', - 'DASK_UCX__TCP':'True'}, + close_fds=True, env, restore_signals=False, start_new_session=True) -def start_workers( conf ): +def start_workers( conf, env ): nworkers=int(conf.dask_n_workers) gpu_max_mem=float(device_memory_limit()) - device_mem_limit="{}MB".format(int(float(conf.get('dask_device-memory-limit',gpu_max_mem))/(1024*1024)*.8)) - sys_max_mem="{}MB".format( int(memory_limit()/(int(nworkers)*1024*1024))) - env=os.environ.copy() + device_mem_limit="{}MB".format(int(float(conf.get('dask_device-memory-limit',gpu_max_mem))/(1024**2)*.8)) + sys_max_mem="{}MB".format( int(memory_limit()/(int(nworkers)*1024**2))) env.update({'DEVICE_MEMORY_LIMIT':device_mem_limit, - 'POOL_SIZE':'{}MB'.format( int(gpu_max_mem/(1024*1024))), + 'POOL_SIZE':'{}MB'.format( int(gpu_max_mem/(1024**2))), 'LOGDIR':conf.get('log_dir', './'), 'WORKER_DIR':conf.get('dask_worker_dir','./'), - 'MAX_SYSTEM_MEMORY':sys_max_mem }) + 'MAX_SYSTEM_MEMORY':sys_max_mem}) args=[sys.executable, '-m', 'dask_cuda.cli.dask_cuda_worker', '--device-memory-limit', device_mem_limit, '--no-reconnect', @@ -119,7 +137,8 @@ def start_workers( conf ): close_fds=True, env=env, restore_signals=False, start_new_session=True) #print( "*** WORKER {} PID: {}".format( i, pid.pid )) -def stop_scheduler( conf ): + +def stop_scheduler( conf, env ): try: client = attach_to_cluster( conf ) client.shutdown() @@ -127,7 +146,7 @@ def stop_scheduler( conf ): log.exception( "Failed to stop scheduler", sys.exc_info()) -def stop_workers( conf ): +def stop_workers( conf, env ): try: client = attach_to_cluster( conf ) client.cancel( client.futures()) @@ -135,7 +154,7 @@ def stop_workers( conf ): log.exception( "Failed to stop workers", sys.exc_info()) -def dump_task_stream( conf ): +def dump_task_stream( conf, env ): try: client = attach_to_cluster( conf ) client.get_task_stream() From 5f30a83fd04c62bf764684e469d0ee2b0e66b118 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 4 Sep 2020 12:39:49 -0700 Subject: [PATCH 10/51] sort keys --- tpcx_bb/xbb_tools/default.config.json | 242 ++++++++++++++------------ 1 file changed, 133 insertions(+), 109 deletions(-) diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index 5aae6059..2c2cb27c 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -1,110 +1,134 @@ [ - {"name":"get_read_time", - "default":"True", - "type":"bool", - "help":"whether or not to track read time" - }, - {"name":"split_row_groups", - "default":"False", - "type":"bool", - "help":"not really sure" - }, - { "name":"file_format", - "default":"parquet", - "type":"str", - "help":"help string" - }, - { "name":"output_filetype", - "default":"parquet", - "type":"str", - "help":"help string" - }, - { "name":"repartition_small_table", - "default":"True", - "type":"str", - "help":"help string" - }, - { "name":"benchmark_runner_include_bsql", - "default":"False", - "type":"str", - "help":"help string" - }, - { "name":"32GB_workers", - "help":"help string" - }, - { "name":"verify_results", - "default":"False", - "type":"str", - "help":"help string" - }, - { "name":"verify_dir", - "default":"", - "type":"str", - "help":"help string" - }, - { "name":"sheet", - "default":"TPCx-BB", - "type":"str", - "help":"help string" - }, - { "name":"tab", - "default":"SF1000 Benchmarking Matrix", - "type":"str", - "help":"help string" - }, - {"name":"data_dir", - "default":"/datasets/tpcx-bb/sf1/parquet_1gb/", - "type":"str", - "help":"Data Dir" - }, - {"name":"output_dir", - "default":"./", - "type":"str", - "help":"Query Output Directry. Defaults to the directory of the query script." - }, - {"name":"dask_dir", - "default":"./", - "type":"str", - "help":"Dask Dir" - }, - {"name":"dask_profile", - "default":"False", - "type":"bool", - "help":"Include Dask Performance Report" - }, - {"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_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":"dask_n_workers", - "default":1, - "type":"int", - "help":"Number of worker processes to start" - }, - {"name":"logdir", - "default":"./", - "type":"str", - "help":"directory to create log files`" - }, - {"name":"timeout", - "type":"int", - "default":"300", - "help":"max duration for each query" - }, - {"name":"dask_scheduler-file", - "type":"str", - "default":"sched.json", - "help":"File to write connection information." - } -] \ No newline at end of file + { + "name": "benchmark_runner_include_bsql", + "default": "False", + "type": "str", + "help": "help string" + }, + { + "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": "dask_dir", + "default": "./", + "type": "str", + "help": "Dask Dir" + }, + { + "name": "dask_n_workers", + "default": 1, + "type": "int", + "help": "Number of worker processes to start" + }, + { + "name": "dask_profile", + "default": "False", + "type": "bool", + "help": "Include Dask Performance Report" + }, + { + "name": "dask_scheduler-file", + "type": "str", + "default": "sched.json", + "help": "File to write connection information." + }, + { + "name": "data_dir", + "default": "/datasets/tpcx-bb/sf1/parquet_1gb/", + "type": "str", + "help": "Data Dir" + }, + { + "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": "logdir", + "default": "./", + "type": "str", + "help": "directory to create log files`" + }, + { + "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": "sheet", + "default": "TPCx-BB", + "type": "str", + "help": "help string" + }, + { + "name": "split_row_groups", + "default": "False", + "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", + "default": "False", + "type": "str", + "help": "help string" + } +] From 2ad3868bc3d0b373db845210a75c638f8f9f0d2c Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 4 Sep 2020 13:21:21 -0700 Subject: [PATCH 11/51] config look in cwd for default file --- tpcx_bb/xbb_tools/config.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tpcx_bb/xbb_tools/config.py b/tpcx_bb/xbb_tools/config.py index 51b572e9..66d13c41 100644 --- a/tpcx_bb/xbb_tools/config.py +++ b/tpcx_bb/xbb_tools/config.py @@ -68,7 +68,8 @@ def build_argparser( self, description="Argparser generated from config module" parser.add_argument( name, **d) parser.add_argument( - '-c', '--config', default="./{}.config.json".format(self._prefix.lower()), + '-c', '--config', + default=os.path.join( os.getcwd(),'.'.join([self._prefix.lower()]+["config.json"])), action='store', dest='configfile', help=self.__doc__ ) From 3f4d96b2b8e479b341ccae52e8e93d3f41b3a493 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 4 Sep 2020 13:23:17 -0700 Subject: [PATCH 12/51] blazing args in common config --- tpcx_bb/xbb_tools/daskcluster.py | 55 +++++++++++++++++---------- tpcx_bb/xbb_tools/default.config.json | 19 ++++++++- 2 files changed, 53 insertions(+), 21 deletions(-) diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py index 547882ec..16f70acf 100644 --- a/tpcx_bb/xbb_tools/daskcluster.py +++ b/tpcx_bb/xbb_tools/daskcluster.py @@ -41,27 +41,42 @@ def cli(commandline=None): env=os.environ.copy() env.update({ - 'tpcxbb_benchmark_sweep_run':conf.get('tpcxbb_benchmark_sweep_run', - os.getenv( 'tpcxbb_benchmark_sweep_run','True')), - 'DASK_RMM__POOL_SIZE':conf.get('DASK_RMM__POOL_SIZE', - os.getenv('DASK_RMM__POOL_SIZE','1GB')), - 'DASK_UCX__CUDA_COPY':conf.get('DASK_UCX__CUDA_COPY', - os.getenv('DASK_UCX__CUDA_COEPY','True')), - 'DASK_UCX__TCP':conf.get('DASK_UCX__TCP', - os.getenv('DASK_UCX__TCP','True')), - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT':conf.get( - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT', - os.getenv('DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT',"100s")), - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP':conf.get( - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP', os.getenv( - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP',"600s")), - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN':conf.get( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN', os.getenv( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN',"1s")), - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX':conf.get( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX', os.getenv( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX',"60s")) + 'CUDA_VISIBLE_DEVICES':conf.get('CUDA_VISIBLE_DEVICES',os.getenv( + 'CUDA_VISIBLE_DEVICES':visible_devices())) + }) + if conf.cluster_mode == "TCP": + env.update({ + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT':conf.get( + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT', os.getenv( + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT',"100s")), + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP':conf.get( + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP', os.getenv( + 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP',"600s")), + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN':conf.get( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN', os.getenv( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN',"1s")), + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX':conf.get( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX', os.getenv( + 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX',"60s")) }) + if conf.cluster_mode=='NVLINK': + env.update({ + 'tpcxbb_benchmark_sweep_run':bool(conf.get('tpcxbb_benchmark_sweep_run', + os.getenv( 'tpcxbb_benchmark_sweep_run',True))), + 'DASK_RMM__POOL_SIZE':conf.get('DASK_RMM__POOL_SIZE', + os.getenv('DASK_RMM__POOL_SIZE','1GB')), + 'DASK_UCX__CUDA_COPY':bool(conf.get('DASK_UCX__CUDA_COPY', + os.getenv('DASK_UCX__CUDA_COEPY',True))), + 'DASK_UCX__TCP':bool(conf.get('DASK_UCX__TCP', + os.getenv('DASK_UCX__TCP',True))), + 'DASK_UCX__NVLINK':bool(conf.get('DASK_UCX__NVLINK', + os.getenv('DASK_UCX__NVLINK', True))), + 'DASK_UCX__INFINIBAND':bool(conf.get('DASK_UCX__INFINIBAND', + os.getenv('DASK_UCX__INFINIBAND',False))), + 'DASK_UCX__RDMACM':bool(conf.get('DASK_UCX__RDMACM', + os.getenv('DASK_UCX__RDMACM',False))), + }) + for cmd in conf.commands: cmdf = cmd.upper().strip().replace('-','_') if cmdf == 'START_WORKERS': diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index 2c2cb27c..ae520f55 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -5,6 +5,12 @@ "type": "str", "help": "help string" }, + { + "name":"blazing_allocator", + "type":"bool", + "default":"False", + "help":"If true RMM_POOL_SIZE is set to 1/2 the difference between available GPU mem and DEVICE_MEMORY_LIMIT" + }, { "name": "cluster_host", "default": "localhost", @@ -17,6 +23,11 @@ "type": "str", "help": "transport used by dask_cudf. Should be coordinated with interface. Options: IB, TCP, NVLINK" }, + { + "name":"interface", + "type":"str", + "help":"network interface used by dask_cu*" + }, { "name": "cluster_port", "default": 8786, @@ -39,7 +50,7 @@ "name": "dask_n_workers", "default": 1, "type": "int", - "help": "Number of worker processes to start" + "help": "Number of worker processes to start with each call to daskcluster start_workers" }, { "name": "dask_profile", @@ -59,6 +70,12 @@ "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": "file_format", "default": "parquet", From 0ddb70a1efc48d400eb033efbe86d0447a426336 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Sat, 5 Sep 2020 21:58:29 -0700 Subject: [PATCH 13/51] config imports all env with empty prefix --- tpcx_bb/xbb_tools/config.py | 5 ++--- tpcx_bb/xbb_tools/utils.py | 2 +- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/tpcx_bb/xbb_tools/config.py b/tpcx_bb/xbb_tools/config.py index 66d13c41..31ad5c6c 100644 --- a/tpcx_bb/xbb_tools/config.py +++ b/tpcx_bb/xbb_tools/config.py @@ -29,8 +29,7 @@ class Config(dict): _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 + __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. @@ -124,7 +123,7 @@ def get_default_config( clz ): def get_config( conf={}, fname=None, envprefix=Config._prefix): - env={k.replace(envprefix,''):os.getenv(k) for k in + 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 ) diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index 8d712f63..d991f61f 100644 --- a/tpcx_bb/xbb_tools/utils.py +++ b/tpcx_bb/xbb_tools/utils.py @@ -378,7 +378,7 @@ def tpcxbb_argparser(): def get_tpcxbb_argparser_commandline_args(): - parser = config.get_config().build_argparser( description="Run TPCx-BB query") + parser = config.get_config(envprefix='').build_argparser( description="Run TPCx-BB query") args = parser.parse_args() args = vars(args) From 6404ba74ba60d3409910d1ac6345182311630c1d Mon Sep 17 00:00:00 2001 From: Kevin German Date: Mon, 7 Sep 2020 08:52:56 -0700 Subject: [PATCH 14/51] benchmark_runner entry point --- tpcx_bb/benchmark_runner.py | 7 +++- tpcx_bb/setup.py | 5 ++- tpcx_bb/xbb_tools/daskcluster.py | 60 ++++++++++----------------- tpcx_bb/xbb_tools/default.config.json | 10 +---- 4 files changed, 32 insertions(+), 50 deletions(-) diff --git a/tpcx_bb/benchmark_runner.py b/tpcx_bb/benchmark_runner.py index 599fc18f..83b10d59 100644 --- a/tpcx_bb/benchmark_runner.py +++ b/tpcx_bb/benchmark_runner.py @@ -25,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 @@ -42,7 +42,7 @@ 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 @@ -92,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/setup.py b/tpcx_bb/setup.py index a99db996..e5689765 100644 --- a/tpcx_bb/setup.py +++ b/tpcx_bb/setup.py @@ -22,10 +22,11 @@ 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" + "daskcluster=xbb_tools.daskcluster:cli", + "benchmark_runner=benchmark_runner:main" ] }, include_package_data=True, diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py index 16f70acf..6b264109 100644 --- a/tpcx_bb/xbb_tools/daskcluster.py +++ b/tpcx_bb/xbb_tools/daskcluster.py @@ -34,47 +34,31 @@ def cli(commandline=None): 'commands', nargs='*', help='one or more of.. start_workers, stop_workers, start_scheduler, stop_scheduler. More details: help-commands' ) - parser.usage="Start and stop, and get data from dask cluster described in config file." args = vars(parser.parse_args( commandline )) - conf=config.get_config( args, fname=args.get('configfile') ) - env=os.environ.copy() + conf=config.get_config( args, fname=args.get('configfile'), envprefix='DASK_') + + env={'CUDA_VISIBLE_DEVICES':conf.get('CUDA_VISIBLE_DEVICES',os.getenv( + 'CUDA_VISIBLE_DEVICES',visible_devices()))} - env.update({ - 'CUDA_VISIBLE_DEVICES':conf.get('CUDA_VISIBLE_DEVICES',os.getenv( - 'CUDA_VISIBLE_DEVICES':visible_devices())) - }) if conf.cluster_mode == "TCP": - env.update({ - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT':conf.get( - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT', os.getenv( - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT',"100s")), - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP':conf.get( - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP', os.getenv( - 'DASK_DISTRIBUTED__COMM__TIMEOUTS__TCP',"600s")), - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN':conf.get( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN', os.getenv( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MIN',"1s")), - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX':conf.get( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX', os.getenv( - 'DASK_DISTRIBUTED__COMM__RETRY__DELAY__MAX',"60s")) - }) - if conf.cluster_mode=='NVLINK': - env.update({ - 'tpcxbb_benchmark_sweep_run':bool(conf.get('tpcxbb_benchmark_sweep_run', - os.getenv( 'tpcxbb_benchmark_sweep_run',True))), - 'DASK_RMM__POOL_SIZE':conf.get('DASK_RMM__POOL_SIZE', - os.getenv('DASK_RMM__POOL_SIZE','1GB')), - 'DASK_UCX__CUDA_COPY':bool(conf.get('DASK_UCX__CUDA_COPY', - os.getenv('DASK_UCX__CUDA_COEPY',True))), - 'DASK_UCX__TCP':bool(conf.get('DASK_UCX__TCP', - os.getenv('DASK_UCX__TCP',True))), - 'DASK_UCX__NVLINK':bool(conf.get('DASK_UCX__NVLINK', - os.getenv('DASK_UCX__NVLINK', True))), - 'DASK_UCX__INFINIBAND':bool(conf.get('DASK_UCX__INFINIBAND', - os.getenv('DASK_UCX__INFINIBAND',False))), - 'DASK_UCX__RDMACM':bool(conf.get('DASK_UCX__RDMACM', - os.getenv('DASK_UCX__RDMACM',False))), + env.update({'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") + }) + elif conf.cluster_mode=='NVLINK': + env.update({'tpcxbb_benchmark_sweep_run':bool(conf.get('tpcxbb_benchmark_sweep_run',True)), + 'DASK_RMM__POOL_SIZE':conf.get('rmm__pool_size','1GB'), + 'DASK_UCX__CUDA_COPY':conf.get('ucx__cuda_copy',True), + 'DASK_UCX__TCP':conf.get('ucx__tcp',True), + 'DASK_UCX__NVLINK':conf.get('ucx__nvlink',True), + 'DASK_UCX__INFINIBAND':conf.get('ucx_infiniband',False), + 'DASK_UCX__RDMACM':conf.get('ucx__rdmacm',False), }) for cmd in conf.commands: @@ -94,7 +78,7 @@ def cli(commandline=None): time.sleep(2) def start_scheduler( conf, env ): - clusterkeys=( 'dask_dashboard_address', 'dask_diagnostics_port', 'dask_host', + clusterkeys=( 'dashboard_address', 'diagnostics_port', 'dask_host', 'dask_interface', 'dask_port', 'dask_preload', 'dask_protocol', 'dask_scheduler-file' ) diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index ae520f55..9235e95b 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -1,15 +1,9 @@ [ { - "name": "benchmark_runner_include_bsql", - "default": "False", - "type": "str", - "help": "help string" - }, - { - "name":"blazing_allocator", + "name":"with_blazing", "type":"bool", "default":"False", - "help":"If true RMM_POOL_SIZE is set to 1/2 the difference between available GPU mem and DEVICE_MEMORY_LIMIT" + "help":"If True cluster and runners setup with suppoort for BSQL" }, { "name": "cluster_host", From b0706a7e0ac74a52d8da09d483bfd9bf15838429 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 8 Sep 2020 09:52:12 -0700 Subject: [PATCH 15/51] rename and sort default config --- tpcx_bb/xbb_tools/default.config.json | 72 ++++++++++++--------------- 1 file changed, 33 insertions(+), 39 deletions(-) diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index 9235e95b..f145ba22 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -1,10 +1,4 @@ [ - { - "name":"with_blazing", - "type":"bool", - "default":"False", - "help":"If True cluster and runners setup with suppoort for BSQL" - }, { "name": "cluster_host", "default": "localhost", @@ -17,11 +11,6 @@ "type": "str", "help": "transport used by dask_cudf. Should be coordinated with interface. Options: IB, TCP, NVLINK" }, - { - "name":"interface", - "type":"str", - "help":"network interface used by dask_cu*" - }, { "name": "cluster_port", "default": 8786, @@ -34,30 +23,6 @@ "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": "dask_dir", - "default": "./", - "type": "str", - "help": "Dask Dir" - }, - { - "name": "dask_n_workers", - "default": 1, - "type": "int", - "help": "Number of worker processes to start with each call to daskcluster start_workers" - }, - { - "name": "dask_profile", - "default": "False", - "type": "bool", - "help": "Include Dask Performance Report" - }, - { - "name": "dask_scheduler-file", - "type": "str", - "default": "sched.json", - "help": "File to write connection information." - }, { "name": "data_dir", "default": "/datasets/tpcx-bb/sf1/parquet_1gb/", @@ -65,10 +30,10 @@ "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": "device_memory_limit", + "type": "float", + "default": "0.9", + "help": "GPU memory limit is set ot a percent of available GPU memory" }, { "name": "file_format", @@ -82,12 +47,23 @@ "type": "bool", "help": "whether or not to track read time" }, + { + "name": "interface", + "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": "./", @@ -100,12 +76,24 @@ "type": "str", "help": "help string" }, + { + "name": "profile", + "default": "False", + "type": "bool", + "help": "Include Dask Performance Report" + }, { "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", @@ -141,5 +129,11 @@ "default": "False", "type": "str", "help": "help string" + }, + { + "name": "with_blazing", + "type": "bool", + "default": "False", + "help": "If True cluster and runners setup with suppoort for BSQL" } ] From 649bf7e8e0260354c7ebbb03e6285bbce83a2ccb Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 8 Sep 2020 11:27:45 -0700 Subject: [PATCH 16/51] daskcluster args and env Using _DASK prefix for env, but conf variables do not use the prefix mem limits in MB handle cluster mode variants as default handle with_blazing case --- tpcx_bb/xbb_tools/daskcluster.py | 111 ++++++++++++++------------ tpcx_bb/xbb_tools/default.config.json | 1 + 2 files changed, 59 insertions(+), 53 deletions(-) diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py index 6b264109..818f7395 100644 --- a/tpcx_bb/xbb_tools/daskcluster.py +++ b/tpcx_bb/xbb_tools/daskcluster.py @@ -22,9 +22,11 @@ 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): @@ -32,36 +34,24 @@ def cli(commandline=None): parser.add_argument( 'commands', nargs='*', - help='one or more of.. start_workers, stop_workers, start_scheduler, stop_scheduler. More details: help-commands' + help='one or more of.. start_workers, stop_workers, start_scheduler, stop_scheduler.' ) parser.usage="Start and stop, and get data from dask cluster described in config file." args = vars(parser.parse_args( commandline )) conf=config.get_config( args, fname=args.get('configfile'), envprefix='DASK_') - env={'CUDA_VISIBLE_DEVICES':conf.get('CUDA_VISIBLE_DEVICES',os.getenv( - 'CUDA_VISIBLE_DEVICES',visible_devices()))} - - if conf.cluster_mode == "TCP": - env.update({'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") - }) - elif conf.cluster_mode=='NVLINK': - env.update({'tpcxbb_benchmark_sweep_run':bool(conf.get('tpcxbb_benchmark_sweep_run',True)), - 'DASK_RMM__POOL_SIZE':conf.get('rmm__pool_size','1GB'), - 'DASK_UCX__CUDA_COPY':conf.get('ucx__cuda_copy',True), - 'DASK_UCX__TCP':conf.get('ucx__tcp',True), - 'DASK_UCX__NVLINK':conf.get('ucx__nvlink',True), - 'DASK_UCX__INFINIBAND':conf.get('ucx_infiniband',False), - 'DASK_UCX__RDMACM':conf.get('ucx__rdmacm',False), - }) - - for cmd in conf.commands: + env={'CUDA_VISIBLE_DEVICES':conf.get( + 'CUDA_VISIBLE_DEVICES',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 ) @@ -78,43 +68,58 @@ def cli(commandline=None): time.sleep(2) def start_scheduler( conf, env ): - clusterkeys=( 'dashboard_address', 'diagnostics_port', 'dask_host', - 'dask_interface', 'dask_port', 'dask_preload', - 'dask_protocol', 'dask_scheduler-file' ) + clusterkeys=( 'dashboard_address', 'diagnostics_port', 'host', + 'interface', 'port', 'preload', + 'protocol', 'scheduler-file' ) - args=[str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [k.replace('dask_','--'),conf.get(k)]] + if conf.cluster_mode=='NVLINK': + env.update({'tpcxbb_benchmark_sweep_run':conf.get('tpcxbb_benchmark_sweep_run',True), + 'DASK_RMM__POOL_SIZE':conf.get('rmm__pool_size','1GB'), + 'DASK_UCX__CUDA_COPY':conf.get('ucx__cuda_copy',True), + 'DASK_UCX__TCP':conf.get('ucx__tcp',True), + 'DASK_UCX__NVLINK':conf.get('ucx__nvlink',True), + 'DASK_UCX__INFINIBAND':conf.get('ucx_infiniband',False), + 'DASK_UCX__RDMACM':conf.get('ucx__rdmacm',False), + }) + + args=[str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [f"--{k}",conf.get(k)]] return subprocess.Popen([sys.executable, '-m', 'distributed.cli.dask_scheduler'] + 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, - restore_signals=False, start_new_session=True) + close_fds=True, env=env, restore_signals=False, start_new_session=True) def start_workers( conf, env ): - nworkers=int(conf.dask_n_workers) - gpu_max_mem=float(device_memory_limit()) - device_mem_limit="{}MB".format(int(float(conf.get('dask_device-memory-limit',gpu_max_mem))/(1024**2)*.8)) - sys_max_mem="{}MB".format( int(memory_limit()/(int(nworkers)*1024**2))) - env.update({'DEVICE_MEMORY_LIMIT':device_mem_limit, - 'POOL_SIZE':'{}MB'.format( int(gpu_max_mem/(1024**2))), + 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('dask_worker_dir','./'), - 'MAX_SYSTEM_MEMORY':sys_max_mem}) + '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', device_mem_limit, '--no-reconnect', - '--{}-tcp-over-ucx'.format(conf.get('dask_tcp-over-ucx','disable')), - '--memory-limit', conf.get('dask_memory-limit',sys_max_mem), - '--rmm-pool-size', conf.get('dask_rmm-pool-size',device_mem_limit), - '--{}-infiniband'.format( conf.get('dask_infiniband','disable')), - '--{}-nvlink'.format( conf.get( 'dask_nvlink', 'disable')), - '--{}-rdmacm'.format( conf.get( 'dask_rdmacm', 'disable'))] - - clusterkeys=( 'dask_diagnostics_port', 'dask_host', 'dask_local-directory', - 'dask_interface', 'dask_port', 'dask_preload', - 'dask_scheduler-file', 'dask_net-devices', 'dask_nthreads', - 'dask_tls-ca-file', 'dask_tls-cert', 'dask_tls-key' ) - - args+=[str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [k.replace('dask_','--'),conf.get(k)]] + '--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_mem_max_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)]] #print ("EXECUTE: {}".format( ' '.join( map(str,args)))) visible_gpus=visible_devices() diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index f145ba22..4b1cf980 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -49,6 +49,7 @@ }, { "name": "interface", + "default":"eth0", "type": "str", "help": "network interface used by dask_cu*" }, From caf2b2c3b3a7f4b0c3b02a851c0438b6e9223bdc Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 15 Sep 2020 07:10:49 -0700 Subject: [PATCH 17/51] load_test uses common config --- .../queries/load_test/tpcx_bb_load_test.py | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) 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..bd753bc2 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,11 @@ from xbb_tools.utils import benchmark, tpcxbb_argparser, run_query from xbb_tools.readers import build_reader import os, subprocess, math, time +from xbb_tools import config - -config["data_dir"] = "/".join(config["data_dir"].rstrip("/").split("/")[:-1]) - -spark_schema_dir = f"{os.getcwd()}/../../spark_table_schemas/" +conf=config.get_config() +print( conf._prefix) +spark_schema_dir = conf.get('spark_schema_dir', f"{os.getcwd()}/../../spark_table_schemas/") # these tables have extra data produced by bigbench dataGen refresh_tables = [ @@ -23,7 +23,7 @@ ] 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] +scale = [x for x in conf["data_dir"].split("/") if "sf" in x][0] part_size = 3 chunksize = "128 MiB" @@ -50,7 +50,7 @@ def read_csv_table(table, chunksize="256 MiB"): names, types = get_schema(table) dtype = {names[i]: types[i] for i in range(0, len(names))} - data_dir = config["data_dir"] + data_dir = conf["data_dir"] base_path = f"{data_dir}/data/{table}" files = os.listdir(base_path) # item_marketprices has "audit" files that should be excluded @@ -100,7 +100,7 @@ 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"] + data_dir = conf["data_dir"] path = data_dir + "/data/" + table size = subprocess.check_output(["du", "-sh", path]).split()[0].decode("utf-8") unit = size[-1] @@ -144,9 +144,9 @@ def repartition(table, outdir, npartitions=None, chunksize=None, compression="sn ).to_parquet(outdir + table, compression=compression) -def main(client, config): +def main(client, conf): # location you want to write Parquet versions of the table data - data_dir = "/".join(config["data_dir"].split("/")[:-1]) + data_dir = "/".join(conf["data_dir"].split("/")[:-1]) outdir = f"{data_dir}/parquet_{part_size}gb/" total = 0 @@ -168,6 +168,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) From 4f467cfabca77f8302d839f4ab73ed136090ca28 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 15 Sep 2020 07:28:00 -0700 Subject: [PATCH 18/51] daskcluster will wait on and then kill subprocesses --- tpcx_bb/xbb_tools/daskcluster.py | 63 ++++++++++++++++++++++++-------- 1 file changed, 48 insertions(+), 15 deletions(-) diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py index 818f7395..9757e18f 100644 --- a/tpcx_bb/xbb_tools/daskcluster.py +++ b/tpcx_bb/xbb_tools/daskcluster.py @@ -18,6 +18,7 @@ from .device import device_memory_limit, memory_limit, visible_devices import subprocess +import signal import logging, os, sys, math, time log=logging.getLogger() @@ -36,12 +37,14 @@ def cli(commandline=None): 'commands', nargs='*', help='one or more of.. start_workers, stop_workers, start_scheduler, stop_scheduler.' ) - parser.usage="Start and stop, and get data from dask cluster described in config file." + 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_') env={'CUDA_VISIBLE_DEVICES':conf.get( - 'CUDA_VISIBLE_DEVICES',visible_devices()), + '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( @@ -61,6 +64,11 @@ def cli(commandline=None): 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: @@ -73,17 +81,22 @@ def start_scheduler( conf, env ): 'protocol', 'scheduler-file' ) if conf.cluster_mode=='NVLINK': - env.update({'tpcxbb_benchmark_sweep_run':conf.get('tpcxbb_benchmark_sweep_run',True), - 'DASK_RMM__POOL_SIZE':conf.get('rmm__pool_size','1GB'), - 'DASK_UCX__CUDA_COPY':conf.get('ucx__cuda_copy',True), - 'DASK_UCX__TCP':conf.get('ucx__tcp',True), - 'DASK_UCX__NVLINK':conf.get('ucx__nvlink',True), - 'DASK_UCX__INFINIBAND':conf.get('ucx_infiniband',False), - 'DASK_UCX__RDMACM':conf.get('ucx__rdmacm',False), + 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=[str(a) for k in filter( lambda x: x in clusterkeys, conf.keys()) for a in [f"--{k}",conf.get(k)]] - return subprocess.Popen([sys.executable, '-m', 'distributed.cli.dask_scheduler'] + args, + 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) @@ -112,7 +125,7 @@ def start_workers( conf, env ): 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_mem_max_mb*(float(device_mem_limit_mb)/(2*gpu_max_mem_mb)))))] + int(gpu_max_mem_mb*(float(device_mem_limit_mb)/(2*gpu_max_mem_mb)))))] clusterkeys=( 'diagnostics_port', 'host', 'local-directory', 'interface', 'port', 'preload', @@ -120,7 +133,7 @@ def start_workers( conf, env ): '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)]] - #print ("EXECUTE: {}".format( ' '.join( map(str,args)))) + visible_gpus=visible_devices() scale=math.ceil(len(visible_gpus)/nworkers) @@ -131,7 +144,10 @@ def start_workers( conf, env ): nworkers] env['CUDA_VISIBLE_DEVICES']=','.join(list(map(str,mygpus))) env['NVIDIA_VISIBLE_DEVICES']=env['CUDA_VISIBLE_DEVICES'] - #print ("EXECUTE 'worker-{}: {} on GPUs: {}".format( i,' '.join( map(str,args)),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()), @@ -139,7 +155,6 @@ def start_workers( conf, env ): 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) - #print( "*** WORKER {} PID: {}".format( i, pid.pid )) def stop_scheduler( conf, env ): @@ -158,6 +173,24 @@ def stop_workers( conf, env ): 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 ) From 4635a26899534962e399563f306a39e460de2792 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 15 Sep 2020 07:29:43 -0700 Subject: [PATCH 19/51] only need get_config method from config module util uses shared config --- tpcx_bb/xbb_tools/utils.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index d991f61f..0ef4a983 100644 --- a/tpcx_bb/xbb_tools/utils.py +++ b/tpcx_bb/xbb_tools/utils.py @@ -45,7 +45,7 @@ import gspread from oauth2client.service_account import ServiceAccountCredentials -from . import config +from .config import get_config ################################# # Benchmark Timing @@ -279,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( @@ -378,11 +378,9 @@ def tpcxbb_argparser(): def get_tpcxbb_argparser_commandline_args(): - parser = config.get_config(envprefix='').build_argparser( description="Run TPCx-BB query") + 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): From 9a3c3b07f353881e0424443d782caf7677db3768 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Wed, 16 Sep 2020 09:51:28 -0700 Subject: [PATCH 20/51] parameterize conda env --- Dockerfile | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index aa333c9f..eb5355d3 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,7 +5,7 @@ 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 @@ -28,7 +28,7 @@ RUN groupadd --gid ${GROUP_ID} ${USER_GROUP_NAME} && \ chgrp -R ${USER_GROUP_NAME} /conda && \ chmod -R g+rwx /conda -COPY conda/rapids-tpcx-bb.yml /home/${USER_GROUP_NAME}/environment.yml +COPY ${CONDA_ENV_FILE} /home/${USER_GROUP_NAME}/environment.yml COPY tpcx_bb /home/${USER_GROUP_NAME}/tpcx_bb RUN chown -R ${USER_GROUP_NAME}:${USER_GROUP_NAME} /home/${USER_GROUP_NAME}/tpcx_bb && \ From 1d2f25e3d118fc6e32936965708b91594a9eaa48 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Wed, 16 Sep 2020 09:52:17 -0700 Subject: [PATCH 21/51] env for open mpi cuda support --- Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/Dockerfile b/Dockerfile index eb5355d3..530730e9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -18,6 +18,7 @@ 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 RUN groupadd --gid ${GROUP_ID} ${USER_GROUP_NAME} && \ useradd -g ${GROUP_ID} -u ${USER_ID} -ms /bin/bash ${USER_GROUP_NAME} && \ From 61d0293bcd5709f0bda0933839293d1e24c52217 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 18 Sep 2020 09:48:04 -0700 Subject: [PATCH 22/51] ngc uses string identifiers --- tpcx_bb/xbb_tools/device.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tpcx_bb/xbb_tools/device.py b/tpcx_bb/xbb_tools/device.py index 8c2ac8d7..f9eb97ec 100644 --- a/tpcx_bb/xbb_tools/device.py +++ b/tpcx_bb/xbb_tools/device.py @@ -25,7 +25,7 @@ def visible_devices(): os.getenv('CUDA_VISIBLE_DEVICES', 'all' )) if envvar.upper() == 'ALL': return tuple(range(pynvml.nvmlDeviceGetCount())) - return tuple(map( int, envvar.split(','))) + return tuple(envvar.split(',')) def device_memory_limit( devices=None ): From 74fcec8e494b9b7a3d385b847c00fd1dd5053b01 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 18 Sep 2020 09:57:01 -0700 Subject: [PATCH 23/51] code in seperate layer from conda env --- Dockerfile | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index 530730e9..75d42d7d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -30,11 +30,13 @@ RUN groupadd --gid ${GROUP_ID} ${USER_GROUP_NAME} && \ chmod -R g+rwx /conda COPY ${CONDA_ENV_FILE} /home/${USER_GROUP_NAME}/environment.yml -COPY tpcx_bb /home/${USER_GROUP_NAME}/tpcx_bb +RUN /bin/bash -c "source /conda/etc/profile.d/conda.sh && \ + conda env create -f /home/${USER_GROUP_NAME}/environment.yml -n 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 && \ + chown -R ${USER_GROUP_NAME}:${USER_GROUP_NAME} /conda/envs/tpcxbb && \ /bin/bash -c "source /conda/etc/profile.d/conda.sh && \ - conda env create -f /home/${USER_GROUP_NAME}/environment.yml -n tpcxbb && \ conda activate tpcxbb && \ python -m pip install /home/${USER_GROUP_NAME}/tpcx_bb/." From aea55aa8c33178b68cce1665f28733fc4c1c9514 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Thu, 24 Sep 2020 21:49:05 -0700 Subject: [PATCH 24/51] load test parameterized with shared config --- .../queries/load_test/tpcx_bb_load_test.py | 72 +++++++++---------- 1 file changed, 32 insertions(+), 40 deletions(-) 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 bd753bc2..32903614 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 -from xbb_tools import config - -conf=config.get_config() -print( conf._prefix) -spark_schema_dir = conf.get('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 conf["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 = conf["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 = conf["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, conf): +def main(client, config): # location you want to write Parquet versions of the table data - data_dir = "/".join(conf["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"{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() From 79d891a621e96d885a58c1bf84f26849357db941 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Thu, 24 Sep 2020 22:51:57 -0700 Subject: [PATCH 25/51] Workaround for nvbugs/2903712 --- Dockerfile | 3 +++ 1 file changed, 3 insertions(+) diff --git a/Dockerfile b/Dockerfile index 75d42d7d..8a68cd49 100644 --- a/Dockerfile +++ b/Dockerfile @@ -20,6 +20,9 @@ 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 && \ From 1822b4885c1d26190b42608fab8cccdb9133718b Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 25 Sep 2020 10:57:00 -0700 Subject: [PATCH 26/51] always pop type to allow None default --- tpcx_bb/xbb_tools/config.py | 2 +- tpcx_bb/xbb_tools/default.config.json | 5 +---- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/tpcx_bb/xbb_tools/config.py b/tpcx_bb/xbb_tools/config.py index 31ad5c6c..d1afa427 100644 --- a/tpcx_bb/xbb_tools/config.py +++ b/tpcx_bb/xbb_tools/config.py @@ -63,9 +63,9 @@ def build_argparser( self, description="Argparser generated from config module" 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"])), diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index 4b1cf980..b8963a1b 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -103,7 +103,6 @@ }, { "name": "split_row_groups", - "default": "False", "type": "bool", "help": "not really sure" }, @@ -127,14 +126,12 @@ }, { "name": "verify_results", - "default": "False", - "type": "str", + "type": "bool", "help": "help string" }, { "name": "with_blazing", "type": "bool", - "default": "False", "help": "If True cluster and runners setup with suppoort for BSQL" } ] From 336ff19d5a48e270ee903d6eaaea47c33695c9f3 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 25 Sep 2020 11:02:44 -0700 Subject: [PATCH 27/51] dask_profile is not a dask param --- tpcx_bb/xbb_tools/default.config.json | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/tpcx_bb/xbb_tools/default.config.json b/tpcx_bb/xbb_tools/default.config.json index b8963a1b..4ccfbdcd 100644 --- a/tpcx_bb/xbb_tools/default.config.json +++ b/tpcx_bb/xbb_tools/default.config.json @@ -35,6 +35,11 @@ "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", @@ -77,12 +82,6 @@ "type": "str", "help": "help string" }, - { - "name": "profile", - "default": "False", - "type": "bool", - "help": "Include Dask Performance Report" - }, { "name": "repartition_small_table", "default": "True", From 1efdf2635af35a4fe50c50bfb16c3f0a30dc6854 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Fri, 25 Sep 2020 11:21:00 -0700 Subject: [PATCH 28/51] add_empty_args is covered by the default config spec --- tpcx_bb/xbb_tools/utils.py | 27 +-------------------------- 1 file changed, 1 insertion(+), 26 deletions(-) diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index 0ef4a983..0f4bc899 100644 --- a/tpcx_bb/xbb_tools/utils.py +++ b/tpcx_bb/xbb_tools/utils.py @@ -348,33 +348,8 @@ 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() - args = add_empty_config(args) - - return args + return get_tpcxbb_argparser_commandline_args() def get_tpcxbb_argparser_commandline_args(): From 2f86a76c3d2fbb3b1ff49f14e60588c8c2104905 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Sat, 26 Sep 2020 18:39:44 -0700 Subject: [PATCH 29/51] log to file --- tpcx_bb/xbb_tools/daskcluster.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tpcx_bb/xbb_tools/daskcluster.py b/tpcx_bb/xbb_tools/daskcluster.py index 9757e18f..a3f44ec3 100644 --- a/tpcx_bb/xbb_tools/daskcluster.py +++ b/tpcx_bb/xbb_tools/daskcluster.py @@ -43,6 +43,8 @@ def cli(commandline=None): 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( From 0660e0436ea1049947c7199f1106a348aa7ad6ec Mon Sep 17 00:00:00 2001 From: christian Date: Fri, 18 Sep 2020 19:10:41 -0500 Subject: [PATCH 30/51] Remove ORDER BY tstamp --- tpcx_bb/queries/q02/tpcx_bb_query_02_sql.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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) From e60e52138ad383bea6b56d34d6975c81ced6e38e Mon Sep 17 00:00:00 2001 From: esoha-nvidia <69258779+esoha-nvidia@users.noreply.github.com> Date: Tue, 29 Sep 2020 11:33:56 -0600 Subject: [PATCH 31/51] Use os.path.join() to concatenate directory parts `os.path.join` is always better than `+` for making a path. --- tpcx_bb/queries/q03/tpcx_bb_query_03.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tpcx_bb/queries/q03/tpcx_bb_query_03.py b/tpcx_bb/queries/q03/tpcx_bb_query_03.py index 9ea7931d..d3593f8f 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 From b0b14a865f1e9d809117209dd173c0be90dffc87 Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Tue, 6 Oct 2020 08:28:20 -0700 Subject: [PATCH 32/51] use DataFrame onstrucotr instead of from_gpu_matrix to convert a cupy array --- tpcx_bb/queries/q18/tpcx_bb_query_18.py | 2 +- tpcx_bb/queries/q18/tpcx_bb_query_18_sql.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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)) From 3eb2c630f4a71a9e608b63598ef7ed228b8542ba Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Wed, 7 Oct 2020 07:05:57 -0700 Subject: [PATCH 33/51] update accesing of gpu memory from the scheduler to fit the new data structure in distributed 2.30 --- tpcx_bb/xbb_tools/cluster_startup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tpcx_bb/xbb_tools/cluster_startup.py b/tpcx_bb/xbb_tools/cluster_startup.py index e4107090..24db6387 100644 --- a/tpcx_bb/xbb_tools/cluster_startup.py +++ b/tpcx_bb/xbb_tools/cluster_startup.py @@ -142,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 From 66bb380ef1a413194f8c224970d632d66133d50c Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Wed, 7 Oct 2020 13:54:08 -0700 Subject: [PATCH 34/51] shuffle tested for q02 --- tpcx_bb/queries/q02/tpcx_bb_query_02.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 0dd94d75d762ae9bb20e028744ead41942dbfc21 Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Wed, 7 Oct 2020 14:05:04 -0700 Subject: [PATCH 35/51] more shuffle updates 3,4,8 --- tpcx_bb/queries/q03/tpcx_bb_query_03.py | 2 +- tpcx_bb/queries/q04/tpcx_bb_query_04.py | 2 +- tpcx_bb/queries/q08/tpcx_bb_query_08.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/tpcx_bb/queries/q03/tpcx_bb_query_03.py b/tpcx_bb/queries/q03/tpcx_bb_query_03.py index d3593f8f..3ad5233e 100644 --- a/tpcx_bb/queries/q03/tpcx_bb_query_03.py +++ b/tpcx_bb/queries/q03/tpcx_bb_query_03.py @@ -252,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, From ac3c9780e1dae311b123c7ad2a4f7b29dbf3fe5a Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 08:41:35 -0700 Subject: [PATCH 36/51] ignore index in correctness chceck for clustering queries --- tpcx_bb/xbb_tools/utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index 0f4bc899..5ab80c8a 100644 --- a/tpcx_bb/xbb_tools/utils.py +++ b/tpcx_bb/xbb_tools/utils.py @@ -421,6 +421,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( From 104cc24fd4b22a8c55705aef9206e90036a02a29 Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 08:43:12 -0700 Subject: [PATCH 37/51] shuffle in q20 --- tpcx_bb/queries/q20/tpcx_bb_query_20.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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() From 52a5c0bc738c3158739d82b5c48e752eaa99f55c Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 08:59:51 -0700 Subject: [PATCH 38/51] updated q25 for shuffle and no longer doing an implicit sort upstream --- tpcx_bb/queries/q25/tpcx_bb_query_25.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 From 2ae613fc56f0e175239607280bddad9322ed887d Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 09:20:45 -0700 Subject: [PATCH 39/51] q29 and q30 --- tpcx_bb/queries/q29/tpcx_bb_query_29.py | 2 +- tpcx_bb/queries/q30/tpcx_bb_query_30.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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. From 39780ca72e298ba1dd636637e726b9ff4efb4ec9 Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 09:21:23 -0700 Subject: [PATCH 40/51] q08 sql --- tpcx_bb/queries/q08/tpcx_bb_query_08_sql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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( From d21c317ef70e1cbfc8b27bbedfe31118c52d275d Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 14:19:17 -0700 Subject: [PATCH 41/51] groupby sort needs to be explicit now --- tpcx_bb/queries/q16/tpcx_bb_query_16.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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) From 5e6a08b1341c6ae15e93351b19594fec544f5391 Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 14:19:43 -0700 Subject: [PATCH 42/51] shuffle in hash merge --- tpcx_bb/xbb_tools/merge_util.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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, From 383edef841e5bc60c0e47e0d70f2295df51ca892 Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 15:21:59 -0700 Subject: [PATCH 43/51] q26 sort --- tpcx_bb/queries/q26/tpcx_bb_query_26.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tpcx_bb/queries/q26/tpcx_bb_query_26.py b/tpcx_bb/queries/q26/tpcx_bb_query_26.py index 2ecabbce..d3fb4b80 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 q25-result formatting + output = output.sort_values(["ss_customer_sk"]) + # Based on CDH6.1 q26-result formatting results_dict["cid_labels"] = output return results_dict From f48986e027c6e2ff7e473031cb3d591bad8f43de Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 15:23:46 -0700 Subject: [PATCH 44/51] q22 implicit sort to explicit groupby sort --- tpcx_bb/queries/q22/tpcx_bb_query_22.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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() ) From 8c7d7988df343e22b559c1bda944b7be48683a86 Mon Sep 17 00:00:00 2001 From: Nick Becker Date: Thu, 8 Oct 2020 15:27:25 -0700 Subject: [PATCH 45/51] typo in q26 comment --- tpcx_bb/queries/q26/tpcx_bb_query_26.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tpcx_bb/queries/q26/tpcx_bb_query_26.py b/tpcx_bb/queries/q26/tpcx_bb_query_26.py index d3fb4b80..28b202a0 100644 --- a/tpcx_bb/queries/q26/tpcx_bb_query_26.py +++ b/tpcx_bb/queries/q26/tpcx_bb_query_26.py @@ -81,7 +81,7 @@ def get_clusters(client, kmeans_input_df): ) output["label"] = labels_final.reset_index()[0] - # Sort based on CDH6.1 q25-result formatting + # Sort based on CDH6.1 q26-result formatting output = output.sort_values(["ss_customer_sk"]) # Based on CDH6.1 q26-result formatting From e87be7724d665e0f24425506d95202465f3b62cc Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 29 Sep 2020 11:06:21 -0700 Subject: [PATCH 46/51] suppport VISIBLE_DEVICES as UUIDs --- tpcx_bb/xbb_tools/device.py | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/tpcx_bb/xbb_tools/device.py b/tpcx_bb/xbb_tools/device.py index f9eb97ec..fbaed818 100644 --- a/tpcx_bb/xbb_tools/device.py +++ b/tpcx_bb/xbb_tools/device.py @@ -20,23 +20,20 @@ pynvml.nvmlInit() -def visible_devices(): +def visible_devices(devices='all'): + """returns a tuple of integers representing device indices""" envvar = os.getenv( 'NVIDIA_VISIBLE_DEVICES', - os.getenv('CUDA_VISIBLE_DEVICES', 'all' )) + os.getenv('CUDA_VISIBLE_DEVICES', devices )) if envvar.upper() == 'ALL': - return tuple(range(pynvml.nvmlDeviceGetCount())) - return tuple(envvar.split(',')) + 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=None ): - indices=[] - if not devices: - indices=visible_devices() - else: - indices=map(int,re.split( '[, ]+', devices)) - +def device_memory_limit( devices=[] ): return min( [pynvml.nvmlDeviceGetMemoryInfo( - pynvml.nvmlDeviceGetHandleByIndex(d)).total for d in indices] ) + pynvml.nvmlDeviceGetHandleByIndex(d)).total for d in visible_devices(','.join(devices) if devices else None)] ) def memory_limit(): From db8d05ed525546177bcfabac631912786eb50c4e Mon Sep 17 00:00:00 2001 From: Kevin German Date: Wed, 30 Sep 2020 10:58:55 -0700 Subject: [PATCH 47/51] enable output dir overrride via config --- tpcx_bb/queries/load_test/tpcx_bb_load_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 32903614..dca605ee 100644 --- a/tpcx_bb/queries/load_test/tpcx_bb_load_test.py +++ b/tpcx_bb/queries/load_test/tpcx_bb_load_test.py @@ -138,7 +138,7 @@ def repartition(table, data_dir, schema_dir, outdir, npartitions=None, chunksize def main(client, config): # location you want to write Parquet versions of the table data data_dir = config.get("data_dir",'.') - outdir = f"{data_dir}/parquet_{config.get('partitions',3)}gb/" + 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() From 9328a29c0247d129b6c3e754925f3e9d02a1e6d3 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Mon, 5 Oct 2020 12:31:41 -0700 Subject: [PATCH 48/51] FU Dockerfile layers --- Dockerfile | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/Dockerfile b/Dockerfile index 8a68cd49..988fca01 100644 --- a/Dockerfile +++ b/Dockerfile @@ -33,12 +33,13 @@ RUN groupadd --gid ${GROUP_ID} ${USER_GROUP_NAME} && \ 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" +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 && \ - chown -R ${USER_GROUP_NAME}:${USER_GROUP_NAME} /conda/envs/tpcxbb && \ /bin/bash -c "source /conda/etc/profile.d/conda.sh && \ conda activate tpcxbb && \ python -m pip install /home/${USER_GROUP_NAME}/tpcx_bb/." From 31f69f286670681efb940eaacf4988f81365227a Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 6 Oct 2020 08:22:39 -0700 Subject: [PATCH 49/51] Use os.path.join to construct paths external path elements may or may not have trailing fs delimiters --- tpcx_bb/xbb_tools/utils.py | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index 5ab80c8a..dd872129 100644 --- a/tpcx_bb/xbb_tools/utils.py +++ b/tpcx_bb/xbb_tools/utils.py @@ -126,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) @@ -136,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 @@ -150,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 ) @@ -180,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( @@ -190,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 ) @@ -207,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") @@ -223,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 @@ -589,7 +589,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" @@ -634,7 +635,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" From 5e333b8317aec3d8bc8d28ea65c513bbb57f3e2a Mon Sep 17 00:00:00 2001 From: Kevin German Date: Tue, 6 Oct 2020 11:20:55 -0700 Subject: [PATCH 50/51] Assume CUDA version from environment --- conda/rapids-tpcx-bb.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 From 0d798014c6199e4aed1bc1d4523005538946e362 Mon Sep 17 00:00:00 2001 From: Kevin German Date: Wed, 7 Oct 2020 11:01:44 -0700 Subject: [PATCH 51/51] more strategies to find query_num --- tpcx_bb/xbb_tools/utils.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tpcx_bb/xbb_tools/utils.py b/tpcx_bb/xbb_tools/utils.py index dd872129..f95d4522 100644 --- a/tpcx_bb/xbb_tools/utils.py +++ b/tpcx_bb/xbb_tools/utils.py @@ -374,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') #################################