code: def main():
      parser = argparse.ArgumentParser("rapidssample")
      parser.add_argument(
          "--data_dir", type=str, help="location of data"
      )
      parser.add_argument(
          "--num_gpu", type=int, help="Number of GPUs to use", default=1
      )
      parser.add_argument(
          "--part_count",
          type=int,
          help="Number of data files to train against",
          default=2,
      )
      parser.add_argument(
          "--end_year",
          type=int,
          help="Year to end the data load",
          default=2000,
      )
      parser.add_argument(
          "--cpu_predictor",
          type=str,
          help="Flag to use CPU for prediction",
          default="False",
      )
      parser.add_argument(
          "-f", type=str, default=""
      )  # added for notebook execution scenarios
      args = parser.parse_args()
      data_dir = args.data_dir
      num_gpu = args.num_gpu
      part_count = args.part_count
      end_year = args.end_year
      cpu_predictor = args.cpu_predictor.lower() in (
          "yes",
          "true",
          "t",
          "y",
          "1",
      )
  
      if cpu_predictor:
          print("Training with CPUs require num gpu = 1")
          num_gpu = 1
  
      print("data_dir = {0}".format(data_dir))
      print("num_gpu = {0}".format(num_gpu))
      print("part_count = {0}".format(part_count))
      print("end_year = {0}".format(end_year))
      print("cpu_predictor = {0}".format(cpu_predictor))
  
      import subprocess
  
      cmd = "hostname --all-ip-addresses"
      process = subprocess.Popen(cmd.split(), stdout=subprocess.PIPE)
      output, error = process.communicate()
      IPADDR = str(output.decode()).split()[0]
  
      cluster = LocalCUDACluster(ip=IPADDR, n_workers=num_gpu)
      client = Client(cluster)
      client
      print(client.ncores())
  
      # to download data for this notebook, visit https://rapidsai.github.io/demos/datasets/mortgage-data and update the following paths accordingly
      acq_data_path = "{0}/acq".format(
          data_dir
      )  # "/rapids/data/mortgage/acq"
      perf_data_path = "{0}/perf".format(
          data_dir
      )  # "/rapids/data/mortgage/perf"
      col_names_path = "{0}/names.csv".format(
          data_dir
      )  # "/rapids/data/mortgage/names.csv"
      start_year = 2000
  
      client
      print("--->>> Workers used: {0}".format(client.ncores()))
  
      # NOTE: The ETL calculates additional features which are then dropped before creating the XGBoost DMatrix.
      # This can be optimized to avoid calculating the dropped features.
      print("Reading ...")
      t1 = datetime.datetime.now()
      gpu_dfs = []
      gpu_time = 0
      quarter = 1
      year = start_year
      count = 0
      while year <= end_year:
          for file in glob(
              os.path.join(
                  perf_data_path
                  + "/Performance_"
                  + str(year)
                  + "Q"
                  + str(quarter)
                  + "*"
              )
          ):
              if count < part_count:
                  gpu_dfs.append(
                      process_quarter_gpu(
                          client,
                          col_names_path,
                          acq_data_path,
                          year=year,
                          quarter=quarter,
                          perf_file=file,
                      )
                  )
                  count += 1
                  print("file: {0}".format(file))
                  print("count: {0}".format(count))
          quarter += 1
          if quarter == 5:
              year += 1
              quarter = 1
  
      wait(gpu_dfs)
      t2 = datetime.datetime.now()
      print("Reading time: {0}".format(str(t2 - t1)))
      print("--->>> Number of data parts: {0}".format(len(gpu_dfs)))
  
      dxgb_gpu_params = {
          "nround": 100,
          "max_depth": 8,
          "max_leaves": 2**8,
          "alpha": 0.9,
          "eta": 0.1,
          "gamma": 0.1,
          "learning_rate": 0.1,
          "subsample": 1,
          "reg_lambda": 1,
          "scale_pos_weight": 2,
          "min_child_weight": 30,
          "tree_method": "gpu_hist",
          "n_gpus": 1,
          "distributed_dask": True,
          "loss": "ls",
          "objective": "reg:squarederror",
          "max_features": "auto",
          "criterion": "friedman_mse",
          "grow_policy": "lossguide",
          "verbose": True,
      }
  
      if cpu_predictor:
          print("\n---->>>> Training using CPUs <<<<----\n")
          dxgb_gpu_params["predictor"] = "cpu_predictor"
          dxgb_gpu_params["tree_method"] = "hist"
          dxgb_gpu_params["objective"] = "reg:linear"
  
      else:
          print("\n---->>>> Training using GPUs <<<<----\n")
  
      print("Training parameters are {0}".format(dxgb_gpu_params))
  
      gpu_dfs = [
          delayed(DataFrame.from_arrow)(gpu_df)
          for gpu_df in gpu_dfs[:part_count]
      ]
      gpu_dfs = [gpu_df for gpu_df in gpu_dfs]
      wait(gpu_dfs)
  
      tmp_map = [
          (gpu_df, list(client.who_has(gpu_df).values())[0])
          for gpu_df in gpu_dfs
      ]
      new_map = {}
      for key, value in tmp_map:
          if value not in new_map:
              new_map[value] = [key]
          else:
              new_map[value].append(key)
  
      del tmp_map
      gpu_dfs = []
      for list_delayed in new_map.values():
          gpu_dfs.append(delayed(cudf.concat)(list_delayed))
  
      del new_map
      gpu_dfs = [
          (
              gpu_df[["delinquency_12"]],
              gpu_df[
                  delayed(list)(
                      gpu_df.columns.difference(["delinquency_12"])
                  )
              ],
          )
          for gpu_df in gpu_dfs
      ]
      gpu_dfs = [
          (gpu_df[0].persist(), gpu_df[1].persist())
          for gpu_df in gpu_dfs
      ]
  
      gpu_dfs = [
          dask.delayed(xgb.DMatrix)(gpu_df[1], gpu_df[0])
          for gpu_df in gpu_dfs
      ]
      gpu_dfs = [gpu_df.persist() for gpu_df in gpu_dfs]
      gc.collect()
      wait(gpu_dfs)
  
      # TRAIN THE MODEL
      labels = None
      t1 = datetime.datetime.now()
      bst = dxgb_gpu.train(
          client,
          dxgb_gpu_params,
          gpu_dfs,
          labels,
          num_boost_round=dxgb_gpu_params["nround"],
      )
      t2 = datetime.datetime.now()
      print(
          "\n---->>>> Training time: {0} <<<<----\n".format(
              str(t2 - t1)
          )
      )
      print("Exiting script")
  
imports: import numpy
  import datetime
  import dask_xgboost
  import dask
  import dask_cudf
  from dask_cuda import LocalCUDACluster
  from dask.delayed import delayed
  from dask.distributed import Client, wait
  import xgboost
  import cudf
  from cudf.dataframe import DataFrame
  from collections import OrderedDict
  import gc
  from glob import glob
  import os
  import argparse
label_desc: Use image node_id 17 for calling a global function with example usage: main() without return types