code: def get_args(*args, **kwargs):
      parser = argparse.ArgumentParser()
      parser.add_argument(
          "--ray_address",
          default="auto",
          type=str,
          help="if set to None, will launch a local Ray cluster",
      )
      parser.add_argument(
          "--total_data_size",
          default=1 * 1000 * 1024 * 1024 * 1024,
          type=ByteCount,
          help="total data size in bytes",
      )
      parser.add_argument(
          "--num_mappers",
          default=256,
          type=int,
          help="number of map tasks",
      )
      parser.add_argument(
          "--num_mappers_per_round",
          default=16,
          type=int,
          help="number of map tasks per first-stage merge tasks",
      )
      parser.add_argument(
          "--num_reducers",
          default=16,
          type=int,
          help="number of second-stage reduce tasks",
      )
      parser.add_argument(
          "--num_concurrent_rounds",
          default=4,
          type=int,
          help="max number of rounds of map/merge tasks in flight",
      )
      parser.add_argument(
          "--reducer_input_chunk",
          default=100 * 1024 * 1024,
          type=ByteCount,
          help="bytes to read from each file in reduce tasks",
      )
      parser.add_argument(
          "--skip_sorting",
          default=False,
          action="store_true",
          help="if set, no sorting is actually performed",
      )
      parser.add_argument(
          "--skip_input",
          default=False,
          action="store_true",
          help="if set, mappers will not read data from disk",
      )
      parser.add_argument(
          "--skip_output",
          default=False,
          action="store_true",
          help="if set, reducers will not write out results to disk",
      )
      # Which tasks to run?
      tasks_group = parser.add_argument_group(
          "tasks to run", "if no task is specified, will run all tasks"
      )
      tasks = ["generate_input", "sort", "validate_output"]
      for task in tasks:
          tasks_group.add_argument(f"--{task}", action="store_true")
  
      args = parser.parse_args(*args, **kwargs)
      # Derive additional arguments.
      args.input_part_size = ByteCount(
          args.total_data_size / args.num_mappers
      )
      assert args.num_mappers % args.num_mappers_per_round == 0
      args.num_rounds = int(
          args.num_mappers / args.num_mappers_per_round
      )
      args.mount_points = _get_mount_points()
      # If no tasks are specified, run all tasks.
      args_dict = vars(args)
      if not any(args_dict[task] for task in tasks):
          for task in tasks:
              args_dict[task] = True
      return args
  
imports: import argparse
  import contextlib
  import csv
  import logging
  import os
  import random
  import subprocess
  import tempfile
  from typing import Callable, Dict, Iterable, List
  import numpy
  import ray
  from ray.experimental.raysort import constants, logging_utils, sortlib, tracing_utils
  from ray.experimental.raysort.types import BlockInfo, ByteCount, PartId, PartInfo, Path, RecordCount
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 1 for calling a global function with example usage: get_args() and returns: args