diff --git a/python/ray/parameter.py b/python/ray/parameter.py
new file mode 100644
index 000000000..350b118e0
--- /dev/null
+++ b/python/ray/parameter.py
@@ -0,0 +1,192 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import logging
+
+import ray.ray_constants as ray_constants
+
+
+class RayParams(object):
+    """A class used to store the parameters used by Ray.
+
+    Attributes:
+        address_info (dict): A dictionary with address information for
+            processes in a partially-started Ray cluster. If
+            start_ray_local=True, any processes not in this dictionary will be
+            started. If provided, an updated address_info dictionary will be
+            returned to include processes that are newly started.
+        start_ray_local (bool): If True then this will start any processes not
+            already in address_info, including Redis, a global scheduler, local
+            scheduler(s), object store(s), and worker(s). It will also kill
+            these processes when Python exits. If False, this will attach to an
+            existing Ray cluster.
+        redis_address (str): The address of the Redis server to connect to. If
+            this address is not provided, then this command will start Redis, a
+            global scheduler, a local scheduler, a plasma store, a plasma
+            manager, and some workers. It will also kill these processes when
+            Python exits.
+        redis_port (int): The port that the primary Redis shard should listen
+            to. If None, then a random port will be chosen. If the key
+            "redis_address" is in address_info, then this argument will be
+            ignored.
+        redis_shard_ports: A list of the ports to use for the non-primary Redis
+            shards.
+        num_cpus (int): Number of cpus the user wishes all local schedulers to
+            be configured with.
+        num_gpus (int): Number of gpus the user wishes all local schedulers to
+            be configured with.
+        num_local_schedulers (int): The number of local schedulers to start.
+            This is only provided if start_ray_local is True.
+        resources: A dictionary mapping the name of a resource to the quantity
+            of that resource available.
+        object_store_memory: The amount of memory (in bytes) to start the
+            object store with.
+        redis_max_memory: The max amount of memory (in bytes) to allow redis
+            to use, or None for no limit. Once the limit is exceeded, redis
+            will start LRU eviction of entries. This only applies to the
+            sharded redis tables (task and object tables).
+        object_manager_ports (list): A list of the ports to use for the object
+            managers. There should be one per object manager being started on
+            this node (typically just one).
+        node_manager_ports (list): A list of the ports to use for the node
+            managers. There should be one per node manager being started on
+            this node (typically just one).
+        collect_profiling_data: Whether to collect profiling data from workers.
+        node_ip_address (str): The IP address of the node that we are on.
+        object_id_seed (int): Used to seed the deterministic generation of
+            object IDs. The same value can be used across multiple runs of the
+            same job in order to generate the object IDs in a consistent
+            manner. However, the same ID should not be used for different jobs.
+        local_mode (bool): True if the code should be executed serially
+            without Ray. This is useful for debugging.
+        redirect_worker_output: True if the stdout and stderr of worker
+            processes should be redirected to files.
+        redirect_output (bool): True if stdout and stderr for non-worker
+            processes should be redirected to files and false otherwise.
+        num_redis_shards: The number of Redis shards to start in addition to
+            the primary Redis shard.
+        redis_max_clients: If provided, attempt to configure Redis with this
+            maxclients number.
+        redis_password (str): Prevents external clients without the password
+            from connecting to Redis if provided.
+        plasma_directory: A directory where the Plasma memory mapped files will
+            be created.
+        worker_path (str): The path of the source code that will be run by the
+            worker.
+        huge_pages: Boolean flag indicating whether to start the Object
+            Store with hugetlbfs support. Requires plasma_directory.
+        include_webui: Boolean flag indicating whether to start the web
+            UI, which is a Jupyter notebook.
+        plasma_store_socket_name (str): If provided, it will specify the socket
+            name used by the plasma store.
+        raylet_socket_name (str): If provided, it will specify the socket path
+            used by the raylet process.
+        temp_dir (str): If provided, it will specify the root temporary
+            directory for the Ray process.
+        include_log_monitor (bool): If True, then start a log monitor to
+            monitor the log files for all processes on this node and push their
+            contents to Redis.
+        autoscaling_config: path to autoscaling config file.
+        _internal_config (str): JSON configuration for overriding
+            RayConfig defaults. For testing purposes ONLY.
+    """
+
+    def __init__(self,
+                 address_info=None,
+                 start_ray_local=False,
+                 redis_address=None,
+                 num_cpus=None,
+                 num_gpus=None,
+                 num_local_schedulers=None,
+                 resources=None,
+                 object_store_memory=None,
+                 redis_max_memory=None,
+                 redis_port=None,
+                 redis_shard_ports=None,
+                 object_manager_ports=None,
+                 node_manager_ports=None,
+                 collect_profiling_data=True,
+                 node_ip_address=None,
+                 object_id_seed=None,
+                 num_workers=None,
+                 local_mode=False,
+                 driver_mode=None,
+                 redirect_worker_output=False,
+                 redirect_output=True,
+                 num_redis_shards=None,
+                 redis_max_clients=None,
+                 redis_password=None,
+                 plasma_directory=None,
+                 worker_path=None,
+                 huge_pages=False,
+                 include_webui=None,
+                 logging_level=logging.INFO,
+                 logging_format=ray_constants.LOGGER_FORMAT,
+                 plasma_store_socket_name=None,
+                 raylet_socket_name=None,
+                 temp_dir=None,
+                 include_log_monitor=None,
+                 autoscaling_config=None,
+                 _internal_config=None):
+        self.address_info = address_info
+        self.start_ray_local = start_ray_local
+        self.object_id_seed = object_id_seed
+        self.redis_address = redis_address
+        self.num_cpus = num_cpus
+        self.num_gpus = num_gpus
+        self.num_local_schedulers = num_local_schedulers
+        self.resources = resources
+        self.object_store_memory = object_store_memory
+        self.redis_max_memory = redis_max_memory
+        self.redis_port = redis_port
+        self.redis_shard_ports = redis_shard_ports
+        self.object_manager_ports = object_manager_ports
+        self.node_manager_ports = node_manager_ports
+        self.collect_profiling_data = collect_profiling_data
+        self.node_ip_address = node_ip_address
+        self.num_workers = num_workers
+        self.local_mode = local_mode
+        self.driver_mode = driver_mode
+        self.redirect_worker_output = redirect_worker_output
+        self.redirect_output = redirect_output
+        self.num_redis_shards = num_redis_shards
+        self.redis_max_clients = redis_max_clients
+        self.redis_password = redis_password
+        self.plasma_directory = plasma_directory
+        self.worker_path = worker_path
+        self.huge_pages = huge_pages
+        self.include_webui = include_webui
+        self.plasma_store_socket_name = plasma_store_socket_name
+        self.raylet_socket_name = raylet_socket_name
+        self.temp_dir = temp_dir
+        self.include_log_monitor = include_log_monitor
+        self.autoscaling_config = autoscaling_config
+        self._internal_config = _internal_config
+
+    def update(self, **kwargs):
+        """Update the settings according to the keyword arguments.
+
+        Args:
+            kwargs: The keyword arguments to set corresponding fields.
+        """
+        for arg in kwargs:
+            if (hasattr(self, arg)):
+                setattr(self, arg, kwargs[arg])
+            else:
+                raise ValueError("Invalid RayParams parameter in"
+                                 " update: %s" % arg)
+
+    def update_if_absent(self, **kwargs):
+        """Update the settings when the target fields are None.
+
+        Args:
+            kwargs: The keyword arguments to set corresponding fields.
+        """
+        for arg in kwargs:
+            if (hasattr(self, arg)):
+                if getattr(self, arg) is None:
+                    setattr(self, arg, kwargs[arg])
+            else:
+                raise ValueError("Invalid RayParams parameter in"
+                                 " update_if_absent: %s" % arg)
diff --git a/python/ray/scripts/scripts.py b/python/ray/scripts/scripts.py
index b84db6757..492eac51e 100644
--- a/python/ray/scripts/scripts.py
+++ b/python/ray/scripts/scripts.py
@@ -15,6 +15,8 @@ from ray.autoscaler.commands import (attach_cluster, exec_cluster,
 import ray.ray_constants as ray_constants
 import ray.utils
 
+from ray.parameter import RayParams
+
 logger = logging.getLogger(__name__)
 
 
@@ -243,6 +245,22 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
         resources["CPU"] = num_cpus
     if num_gpus is not None:
         resources["GPU"] = num_gpus
+    ray_params = RayParams(
+        node_ip_address=node_ip_address,
+        object_manager_ports=[object_manager_port],
+        node_manager_ports=[node_manager_port],
+        num_workers=num_workers,
+        object_store_memory=object_store_memory,
+        redis_password=redis_password,
+        redirect_worker_output=not no_redirect_worker_output,
+        redirect_output=not no_redirect_output,
+        resources=resources,
+        plasma_directory=plasma_directory,
+        huge_pages=huge_pages,
+        plasma_store_socket_name=plasma_store_socket_name,
+        raylet_socket_name=raylet_socket_name,
+        temp_dir=temp_dir,
+        _internal_config=internal_config)
 
     if head:
         # Start Ray on the head node.
@@ -266,36 +284,21 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
                             "provided.")
 
         # Get the node IP address if one is not provided.
-        if node_ip_address is None:
-            node_ip_address = services.get_node_ip_address()
+        ray_params.update_if_absent(
+            node_ip_address=services.get_node_ip_address())
         logger.info("Using IP address {} for this node."
-                    .format(node_ip_address))
-
-        address_info = services.start_ray_head(
-            object_manager_ports=[object_manager_port],
-            node_manager_ports=[node_manager_port],
-            node_ip_address=node_ip_address,
+                    .format(ray_params.node_ip_address))
+        ray_params.update_if_absent(
             redis_port=redis_port,
             redis_shard_ports=redis_shard_ports,
-            object_store_memory=object_store_memory,
             redis_max_memory=redis_max_memory,
             collect_profiling_data=collect_profiling_data,
-            num_workers=num_workers,
-            cleanup=False,
-            redirect_worker_output=not no_redirect_worker_output,
-            redirect_output=not no_redirect_output,
-            resources=resources,
             num_redis_shards=num_redis_shards,
             redis_max_clients=redis_max_clients,
-            redis_password=redis_password,
             include_webui=(not no_ui),
-            plasma_directory=plasma_directory,
-            huge_pages=huge_pages,
-            autoscaling_config=autoscaling_config,
-            plasma_store_socket_name=plasma_store_socket_name,
-            raylet_socket_name=raylet_socket_name,
-            temp_dir=temp_dir,
-            _internal_config=internal_config)
+            autoscaling_config=autoscaling_config)
+
+        address_info = services.start_ray_head(ray_params, cleanup=False)
         logger.info(address_info)
         logger.info(
             "\nStarted Ray on this node. You can add additional nodes to "
@@ -350,32 +353,17 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
         services.check_version_info(redis_client)
 
         # Get the node IP address if one is not provided.
-        if node_ip_address is None:
-            node_ip_address = services.get_node_ip_address(redis_address)
+        ray_params.update_if_absent(
+            node_ip_address=services.get_node_ip_address(redis_address))
         logger.info("Using IP address {} for this node."
-                    .format(node_ip_address))
+                    .format(ray_params.node_ip_address))
         # Check that there aren't already Redis clients with the same IP
         # address connected with this Redis instance. This raises an exception
         # if the Redis server already has clients on this node.
-        check_no_existing_redis_clients(node_ip_address, redis_client)
-        address_info = services.start_ray_node(
-            node_ip_address=node_ip_address,
-            redis_address=redis_address,
-            object_manager_ports=[object_manager_port],
-            node_manager_ports=[node_manager_port],
-            num_workers=num_workers,
-            object_store_memory=object_store_memory,
-            redis_password=redis_password,
-            cleanup=False,
-            redirect_worker_output=not no_redirect_worker_output,
-            redirect_output=not no_redirect_output,
-            resources=resources,
-            plasma_directory=plasma_directory,
-            huge_pages=huge_pages,
-            plasma_store_socket_name=plasma_store_socket_name,
-            raylet_socket_name=raylet_socket_name,
-            temp_dir=temp_dir,
-            _internal_config=internal_config)
+        check_no_existing_redis_clients(ray_params.node_ip_address,
+                                        redis_client)
+        ray_params.redis_address = redis_address
+        address_info = services.start_ray_node(ray_params, cleanup=False)
         logger.info(address_info)
         logger.info("\nStarted Ray on this node. If you wish to terminate the "
                     "processes that have been started, run\n\n"
diff --git a/python/ray/services.py b/python/ray/services.py
index 77138715d..083810156 100644
--- a/python/ray/services.py
+++ b/python/ray/services.py
@@ -867,41 +867,31 @@ def check_and_update_resources(resources):
     return resources
 
 
-def start_raylet(redis_address,
-                 node_ip_address,
+def start_raylet(ray_params,
+                 index,
                  raylet_name,
                  plasma_store_name,
-                 worker_path,
-                 resources=None,
-                 object_manager_port=None,
-                 node_manager_port=None,
                  num_workers=0,
                  use_valgrind=False,
                  use_profiler=False,
                  stdout_file=None,
                  stderr_file=None,
                  cleanup=True,
-                 config=None,
-                 redis_password=None,
-                 collect_profiling_data=True):
+                 config=None):
     """Start a raylet, which is a combined local scheduler and object manager.
 
     Args:
-        redis_address (str): The address of the Redis instance.
-        node_ip_address (str): The IP address of the node that this local
-            scheduler is running on.
-        plasma_store_name (str): The name of the plasma store socket to connect
-            to.
+        ray_params (ray.params.RayParams): The RayParams instance. The
+            following parameters could be checked: redis_address,
+            node_ip_address, worker_path, resources, object_manager_ports,
+            node_manager_ports, redis_password
+        index (int): Usually, this index is 0. When index > 0, it means
+            starting multiple raylet locally. The index will be used in
+            resources, object_manager_ports, node_manager_ports.
         raylet_name (str): The name of the raylet socket to create.
-        worker_path (str): The path of the script to use when the local
-            scheduler starts up new workers.
-        resources: The resources that this raylet has.
-        object_manager_port (int): The port to use for the object manager. If
-            this is not provided, we will use 0 and the object manager will
-            choose its own port.
-        node_manager_port (int): The port to use for the node manager. If
-            this is not provided, we will use 0 and the node manager will
-            choose its own port.
+        plasma_store_name (str): The name of the plasma store socket to connect
+             to.
+        num_workers (int): The number of workers to start.
         use_valgrind (bool): True if the raylet should be started inside
             of valgrind. If this is True, use_profiler must be False.
         use_profiler (bool): True if the raylet should be started inside
@@ -915,8 +905,6 @@ def start_raylet(redis_address,
             Python process that imported services exits.
         config (dict|None): Optional Raylet configuration that will
             override defaults in RayConfig.
-        redis_password (str): The password of the redis server.
-        collect_profiling_data: Whether to collect profiling data from workers.
 
     Returns:
         The raylet socket name.
@@ -927,7 +915,7 @@ def start_raylet(redis_address,
     if use_valgrind and use_profiler:
         raise Exception("Cannot use valgrind and profiler at the same time.")
 
-    static_resources = check_and_update_resources(resources)
+    static_resources = check_and_update_resources(ray_params.resources[index])
 
     # Limit the number of workers that can be started in parallel by the
     # raylet. However, make sure it is at least 1.
@@ -938,7 +926,7 @@ def start_raylet(redis_address,
     resource_argument = ",".join(
         ["{},{}".format(*kv) for kv in static_resources.items()])
 
-    gcs_ip_address, gcs_port = redis_address.split(":")
+    gcs_ip_address, gcs_port = ray_params.redis_address.split(":")
 
     # Create the command that the Raylet will use to start workers.
     start_worker_command = ("{} {} "
@@ -948,29 +936,31 @@ def start_raylet(redis_address,
                             "--redis-address={} "
                             "--collect-profiling-data={} "
                             "--temp-dir={}".format(
-                                sys.executable, worker_path, node_ip_address,
-                                plasma_store_name, raylet_name, redis_address,
-                                "1" if collect_profiling_data else "0",
+                                sys.executable, ray_params.worker_path,
+                                ray_params.node_ip_address, plasma_store_name,
+                                raylet_name, ray_params.redis_address, "1"
+                                if ray_params.collect_profiling_data else "0",
                                 get_temp_root()))
-    if redis_password:
-        start_worker_command += " --redis-password {}".format(redis_password)
+    if ray_params.redis_password:
+        start_worker_command += " --redis-password {}".format(
+            ray_params.redis_password)
 
     # If the object manager port is None, then use 0 to cause the object
     # manager to choose its own port.
-    if object_manager_port is None:
-        object_manager_port = 0
+    if ray_params.object_manager_ports[index] is None:
+        ray_params.object_manager_ports[index] = 0
     # If the node manager port is None, then use 0 to cause the node manager
     # to choose its own port.
-    if node_manager_port is None:
-        node_manager_port = 0
+    if ray_params.node_manager_ports[index] is None:
+        ray_params.node_manager_ports[index] = 0
 
     command = [
         RAYLET_EXECUTABLE,
         raylet_name,
         plasma_store_name,
-        str(object_manager_port),
-        str(node_manager_port),
-        node_ip_address,
+        str(ray_params.object_manager_ports[index]),
+        str(ray_params.node_manager_ports[index]),
+        ray_params.node_ip_address,
         gcs_ip_address,
         gcs_port,
         str(num_workers),
@@ -979,7 +969,7 @@ def start_raylet(redis_address,
         config_str,
         start_worker_command,
         "",  # Worker command for Java, not needed for Python.
-        redis_password or "",
+        ray_params.redis_password or "",
         get_temp_root(),
     ]
 
@@ -1009,9 +999,9 @@ def start_raylet(redis_address,
     if cleanup:
         all_processes[PROCESS_TYPE_RAYLET].append(pid)
     record_log_files_in_redis(
-        redis_address,
-        node_ip_address, [stdout_file, stderr_file],
-        password=redis_password)
+        ray_params.redis_address,
+        ray_params.node_ip_address, [stdout_file, stderr_file],
+        password=ray_params.redis_password)
 
     return raylet_name
 
@@ -1276,502 +1266,252 @@ def start_raylet_monitor(redis_address,
         all_processes[PROCESS_TYPE_MONITOR].append(p)
 
 
-def start_ray_processes(address_info=None,
-                        object_manager_ports=None,
-                        node_manager_ports=None,
-                        node_ip_address="127.0.0.1",
-                        redis_port=None,
-                        redis_shard_ports=None,
-                        num_workers=None,
-                        num_local_schedulers=1,
-                        object_store_memory=None,
-                        redis_max_memory=None,
-                        collect_profiling_data=True,
-                        num_redis_shards=1,
-                        redis_max_clients=None,
-                        redis_password=None,
-                        worker_path=None,
-                        cleanup=True,
-                        redirect_worker_output=False,
-                        redirect_output=False,
-                        include_log_monitor=False,
-                        include_webui=False,
-                        start_workers_from_local_scheduler=True,
-                        resources=None,
-                        plasma_directory=None,
-                        huge_pages=False,
-                        autoscaling_config=None,
-                        plasma_store_socket_name=None,
-                        raylet_socket_name=None,
-                        temp_dir=None,
-                        _internal_config=None):
+def start_ray_processes(ray_params, cleanup=True):
     """Helper method to start Ray processes.
 
     Args:
-        address_info (dict): A dictionary with address information for
-            processes that have already been started. If provided, address_info
-            will be modified to include processes that are newly started.
-        object_manager_ports (list): A list of the ports to use for the object
-            managers. There should be one per object manager being started on
-            this node (typically just one).
-        node_manager_ports (list): A list of the ports to use for the node
-            managers. There should be one per node manager being started on
-            this node (typically just one).
-        node_ip_address (str): The IP address of this node.
-        redis_port (int): The port that the primary Redis shard should listen
-            to. If None, then a random port will be chosen. If the key
-            "redis_address" is in address_info, then this argument will be
-            ignored.
-        redis_shard_ports: A list of the ports to use for the non-primary Redis
-            shards.
-        num_workers (int): The number of workers to start.
-        num_local_schedulers (int): The total number of local schedulers
-            required. This is also the total number of object stores required.
-            This method will start new instances of local schedulers and object
-            stores until there are num_local_schedulers existing instances of
-            each, including ones already registered with the given
-            address_info.
-        object_store_memory: The amount of memory (in bytes) to start the
-            object store with.
-        redis_max_memory: The max amount of memory (in bytes) to allow redis
-            to use, or None for no limit. Once the limit is exceeded, redis
-            will start LRU eviction of entries. This only applies to the
-            sharded redis tables (task and object tables).
-        collect_profiling_data: Whether to collect profiling data. Note that
-            profiling data cannot be LRU evicted, so if you set
-            redis_max_memory then profiling will also be disabled to prevent
-            it from consuming all available redis memory.
-        num_redis_shards: The number of Redis shards to start in addition to
-            the primary Redis shard.
-        redis_max_clients: If provided, attempt to configure Redis with this
-            maxclients number.
-        redis_password (str): Prevents external clients without the password
-            from connecting to Redis if provided.
-        worker_path (str): The path of the source code that will be run by the
-            worker.
+        ray_params (ray.params.RayParams): The RayParams instance. The
+            following parameters will be set to default values if it's None:
+            node_ip_address("127.0.0.1"), num_local_schedulers(1),
+            include_webui(False), worker_path(path of default_worker.py),
+            include_log_monitor(False)
         cleanup (bool): If cleanup is true, then the processes started here
             will be killed by services.cleanup() when the Python process that
             called this method exits.
-        redirect_worker_output: True if the stdout and stderr of worker
-            processes should be redirected to files.
-        redirect_output (bool): True if stdout and stderr for non-worker
-            processes should be redirected to files and false otherwise.
-        include_log_monitor (bool): If True, then start a log monitor to
-            monitor the log files for all processes on this node and push their
-            contents to Redis.
-        include_webui (bool): If True, then attempt to start the web UI. Note
-            that this is only possible with Python 3.
-        start_workers_from_local_scheduler (bool): If this flag is True, then
-            start the initial workers from the local scheduler. Else, start
-            them from Python.
-        resources: A dictionary mapping resource name to the quantity of that
-            resource.
-        plasma_directory: A directory where the Plasma memory mapped files will
-            be created.
-        huge_pages: Boolean flag indicating whether to start the Object
-            Store with hugetlbfs support. Requires plasma_directory.
-        autoscaling_config: path to autoscaling config file.
-        plasma_store_socket_name (str): If provided, it will specify the socket
-            name used by the plasma store.
-        raylet_socket_name (str): If provided, it will specify the socket path
-            used by the raylet process.
-        temp_dir (str): If provided, it will specify the root temporary
-            directory for the Ray process.
-        _internal_config (str): JSON configuration for overriding
-            RayConfig defaults. For testing purposes ONLY.
 
     Returns:
         A dictionary of the address information for the processes that were
             started.
     """
 
-    set_temp_root(temp_dir)
+    set_temp_root(ray_params.temp_dir)
 
     logger.info("Process STDOUT and STDERR is being redirected to {}.".format(
         get_logs_dir_path()))
 
-    config = json.loads(_internal_config) if _internal_config else None
+    config = json.loads(
+        ray_params._internal_config) if ray_params._internal_config else None
 
-    if resources is None:
-        resources = {}
-    if not isinstance(resources, list):
-        resources = num_local_schedulers * [resources]
+    ray_params.update_if_absent(
+        include_log_monitor=False,
+        resources={},
+        num_local_schedulers=1,
+        include_webui=False,
+        node_ip_address="127.0.0.1")
+    if not isinstance(ray_params.resources, list):
+        ray_params.resources = ray_params.num_local_schedulers * [
+            ray_params.resources
+        ]
 
-    if num_workers is not None:
+    if ray_params.num_workers is not None:
         raise Exception("The 'num_workers' argument is deprecated. Please use "
                         "'num_cpus' instead.")
     else:
         workers_per_local_scheduler = []
-        for resource_dict in resources:
+        for resource_dict in ray_params.resources:
             cpus = resource_dict.get("CPU")
             workers_per_local_scheduler.append(cpus if cpus is not None else
                                                multiprocessing.cpu_count())
 
-    if address_info is None:
-        address_info = {}
-    address_info["node_ip_address"] = node_ip_address
-
-    if worker_path is None:
-        worker_path = os.path.join(
+    ray_params.update_if_absent(
+        address_info={},
+        worker_path=os.path.join(
             os.path.dirname(os.path.abspath(__file__)),
-            "workers/default_worker.py")
+            "workers/default_worker.py"))
+    ray_params.address_info["node_ip_address"] = ray_params.node_ip_address
 
     # Start Redis if there isn't already an instance running. TODO(rkn): We are
     # suppressing the output of Redis because on Linux it prints a bunch of
     # warning messages when it starts up. Instead of suppressing the output, we
     # should address the warnings.
-    redis_address = address_info.get("redis_address")
-    redis_shards = address_info.get("redis_shards", [])
-    if redis_address is None:
-        redis_address, redis_shards = start_redis(
-            node_ip_address,
-            port=redis_port,
-            redis_shard_ports=redis_shard_ports,
-            num_redis_shards=num_redis_shards,
-            redis_max_clients=redis_max_clients,
+    ray_params.redis_address = ray_params.address_info.get("redis_address")
+    ray_params.redis_shards = ray_params.address_info.get("redis_shards", [])
+    if ray_params.redis_address is None:
+        ray_params.redis_address, ray_params.redis_shards = start_redis(
+            ray_params.node_ip_address,
+            port=ray_params.redis_port,
+            redis_shard_ports=ray_params.redis_shard_ports,
+            num_redis_shards=ray_params.num_redis_shards,
+            redis_max_clients=ray_params.redis_max_clients,
             redirect_output=True,
-            redirect_worker_output=redirect_worker_output,
+            redirect_worker_output=ray_params.redirect_worker_output,
             cleanup=cleanup,
-            password=redis_password,
-            redis_max_memory=redis_max_memory)
-        address_info["redis_address"] = redis_address
+            password=ray_params.redis_password,
+            redis_max_memory=ray_params.redis_max_memory)
+        ray_params.address_info["redis_address"] = ray_params.redis_address
         time.sleep(0.1)
 
         # Start monitoring the processes.
         monitor_stdout_file, monitor_stderr_file = new_monitor_log_file(
-            redirect_output)
+            ray_params.redirect_output)
         start_monitor(
-            redis_address,
-            node_ip_address,
+            ray_params.redis_address,
+            ray_params.node_ip_address,
             stdout_file=monitor_stdout_file,
             stderr_file=monitor_stderr_file,
             cleanup=cleanup,
-            autoscaling_config=autoscaling_config,
-            redis_password=redis_password)
+            autoscaling_config=ray_params.autoscaling_config,
+            redis_password=ray_params.redis_password)
         start_raylet_monitor(
-            redis_address,
+            ray_params.redis_address,
             stdout_file=monitor_stdout_file,
             stderr_file=monitor_stderr_file,
             cleanup=cleanup,
-            redis_password=redis_password,
+            redis_password=ray_params.redis_password,
             config=config)
-    if redis_shards == []:
+    if ray_params.redis_shards == []:
         # Get redis shards from primary redis instance.
-        redis_ip_address, redis_port = redis_address.split(":")
+        redis_ip_address, redis_port = ray_params.redis_address.split(":")
         redis_client = redis.StrictRedis(
-            host=redis_ip_address, port=redis_port, password=redis_password)
+            host=redis_ip_address,
+            port=redis_port,
+            password=ray_params.redis_password)
         redis_shards = redis_client.lrange("RedisShards", start=0, end=-1)
-        redis_shards = [ray.utils.decode(shard) for shard in redis_shards]
-        address_info["redis_shards"] = redis_shards
+        ray_params.redis_shards = [
+            ray.utils.decode(shard) for shard in redis_shards
+        ]
+        ray_params.address_info["redis_shards"] = ray_params.redis_shards
 
     # Start the log monitor, if necessary.
-    if include_log_monitor:
+    if ray_params.include_log_monitor:
         log_monitor_stdout_file, log_monitor_stderr_file = (
             new_log_monitor_log_file())
         start_log_monitor(
-            redis_address,
-            node_ip_address,
+            ray_params.redis_address,
+            ray_params.node_ip_address,
             stdout_file=log_monitor_stdout_file,
             stderr_file=log_monitor_stderr_file,
             cleanup=cleanup,
-            redis_password=redis_password)
+            redis_password=ray_params.redis_password)
 
     # Initialize with existing services.
-    if "object_store_addresses" not in address_info:
-        address_info["object_store_addresses"] = []
-    object_store_addresses = address_info["object_store_addresses"]
-    if "raylet_socket_names" not in address_info:
-        address_info["raylet_socket_names"] = []
-    raylet_socket_names = address_info["raylet_socket_names"]
+    if "object_store_addresses" not in ray_params.address_info:
+        ray_params.address_info["object_store_addresses"] = []
+    object_store_addresses = ray_params.address_info["object_store_addresses"]
+    if "raylet_socket_names" not in ray_params.address_info:
+        ray_params.address_info["raylet_socket_names"] = []
+    raylet_socket_names = ray_params.address_info["raylet_socket_names"]
 
     # Get the ports to use for the object managers if any are provided.
-    if not isinstance(object_manager_ports, list):
-        assert object_manager_ports is None or num_local_schedulers == 1
-        object_manager_ports = num_local_schedulers * [object_manager_ports]
-    assert len(object_manager_ports) == num_local_schedulers
-    if not isinstance(node_manager_ports, list):
-        assert node_manager_ports is None or num_local_schedulers == 1
-        node_manager_ports = num_local_schedulers * [node_manager_ports]
-    assert len(node_manager_ports) == num_local_schedulers
+    if not isinstance(ray_params.object_manager_ports, list):
+        assert (ray_params.object_manager_ports is None
+                or ray_params.num_local_schedulers == 1)
+        ray_params.object_manager_ports = (ray_params.num_local_schedulers *
+                                           [ray_params.object_manager_ports])
+    assert len(
+        ray_params.object_manager_ports) == ray_params.num_local_schedulers
+    if not isinstance(ray_params.node_manager_ports, list):
+        assert (ray_params.node_manager_ports is None
+                or ray_params.num_local_schedulers == 1)
+        ray_params.node_manager_ports = (
+            ray_params.num_local_schedulers * [ray_params.node_manager_ports])
+    assert len(
+        ray_params.node_manager_ports) == ray_params.num_local_schedulers
 
     # Start any object stores that do not yet exist.
-    for i in range(num_local_schedulers - len(object_store_addresses)):
+    for i in range(ray_params.num_local_schedulers -
+                   len(object_store_addresses)):
         # Start Plasma.
         plasma_store_stdout_file, plasma_store_stderr_file = (
-            new_plasma_store_log_file(i, redirect_output))
+            new_plasma_store_log_file(i, ray_params.redirect_output))
 
         object_store_address = start_plasma_store(
-            node_ip_address,
-            redis_address,
+            ray_params.node_ip_address,
+            ray_params.redis_address,
             store_stdout_file=plasma_store_stdout_file,
             store_stderr_file=plasma_store_stderr_file,
-            object_store_memory=object_store_memory,
+            object_store_memory=ray_params.object_store_memory,
             cleanup=cleanup,
-            plasma_directory=plasma_directory,
-            huge_pages=huge_pages,
-            plasma_store_socket_name=plasma_store_socket_name,
-            redis_password=redis_password)
+            plasma_directory=ray_params.plasma_directory,
+            huge_pages=ray_params.huge_pages,
+            plasma_store_socket_name=ray_params.plasma_store_socket_name,
+            redis_password=ray_params.redis_password)
         object_store_addresses.append(object_store_address)
         time.sleep(0.1)
 
     # Start any raylets that do not exist yet.
-    for i in range(len(raylet_socket_names), num_local_schedulers):
+    for raylet_index in range(
+            len(raylet_socket_names), ray_params.num_local_schedulers):
         raylet_stdout_file, raylet_stderr_file = new_raylet_log_file(
-            i, redirect_output=redirect_worker_output)
-        address_info["raylet_socket_names"].append(
+            raylet_index, redirect_output=ray_params.redirect_worker_output)
+        ray_params.address_info["raylet_socket_names"].append(
             start_raylet(
-                redis_address,
-                node_ip_address,
-                raylet_socket_name or get_raylet_socket_name(),
-                object_store_addresses[i],
-                worker_path,
-                object_manager_port=object_manager_ports[i],
-                node_manager_port=node_manager_ports[i],
-                resources=resources[i],
-                num_workers=workers_per_local_scheduler[i],
+                ray_params,
+                raylet_index,
+                ray_params.raylet_socket_name or get_raylet_socket_name(),
+                object_store_addresses[raylet_index],
+                num_workers=workers_per_local_scheduler[raylet_index],
                 stdout_file=raylet_stdout_file,
                 stderr_file=raylet_stderr_file,
                 cleanup=cleanup,
-                redis_password=redis_password,
-                collect_profiling_data=collect_profiling_data,
                 config=config))
 
     # Try to start the web UI.
-    if include_webui:
+    if ray_params.include_webui:
         ui_stdout_file, ui_stderr_file = new_webui_log_file()
-        address_info["webui_url"] = start_ui(
-            redis_address,
+        ray_params.address_info["webui_url"] = start_ui(
+            ray_params.redis_address,
             stdout_file=ui_stdout_file,
             stderr_file=ui_stderr_file,
             cleanup=cleanup)
     else:
-        address_info["webui_url"] = ""
+        ray_params.address_info["webui_url"] = ""
     # Return the addresses of the relevant processes.
-    return address_info
+    return ray_params.address_info
 
 
-def start_ray_node(node_ip_address,
-                   redis_address,
-                   object_manager_ports=None,
-                   node_manager_ports=None,
-                   num_workers=None,
-                   num_local_schedulers=1,
-                   object_store_memory=None,
-                   redis_password=None,
-                   worker_path=None,
-                   cleanup=True,
-                   redirect_worker_output=False,
-                   redirect_output=False,
-                   resources=None,
-                   plasma_directory=None,
-                   huge_pages=False,
-                   plasma_store_socket_name=None,
-                   raylet_socket_name=None,
-                   temp_dir=None,
-                   _internal_config=None):
+def start_ray_node(ray_params, cleanup=True):
     """Start the Ray processes for a single node.
 
     This assumes that the Ray processes on some master node have already been
     started.
 
     Args:
-        node_ip_address (str): The IP address of this node.
-        redis_address (str): The address of the Redis server.
-        object_manager_ports (list): A list of the ports to use for the object
-            managers. There should be one per object manager being started on
-            this node (typically just one).
-        node_manager_ports (list): A list of the ports to use for the node
-            managers. There should be one per node manager being started on
-            this node (typically just one).
-        num_workers (int): The number of workers to start.
-        num_local_schedulers (int): The number of local schedulers to start.
-            This is also the number of plasma stores and raylets to start.
-        object_store_memory (int): The maximum amount of memory (in bytes) to
-            let the plasma store use.
-        redis_password (str): Prevents external clients without the password
-            from connecting to Redis if provided.
-        worker_path (str): The path of the source code that will be run by the
-            worker.
+        ray_params (ray.params.RayParams): The RayParams instance. The
+            following parameters could be checked: node_ip_address,
+            redis_address, object_manager_ports, node_manager_ports,
+            num_workers, num_local_schedulers, object_store_memory,
+            redis_password, worker_path, cleanup, redirect_worker_output,
+            redirect_output, resources, plasma_directory, huge_pages,
+            plasma_store_socket_name, raylet_socket_name, temp_dir,
+            _internal_config
         cleanup (bool): If cleanup is true, then the processes started here
             will be killed by services.cleanup() when the Python process that
             called this method exits.
-        redirect_worker_output: True if the stdout and stderr of worker
-            processes should be redirected to files.
-        redirect_output (bool): True if stdout and stderr for non-worker
-            processes should be redirected to files and false otherwise.
-        resources: A dictionary mapping resource name to the available quantity
-            of that resource.
-        plasma_directory: A directory where the Plasma memory mapped files will
-            be created.
-        huge_pages: Boolean flag indicating whether to start the Object
-            Store with hugetlbfs support. Requires plasma_directory.
-        plasma_store_socket_name (str): If provided, it will specify the socket
-            name used by the plasma store.
-        raylet_socket_name (str): If provided, it will specify the socket path
-            used by the raylet process.
-        temp_dir (str): If provided, it will specify the root temporary
-            directory for the Ray process.
-        _internal_config (str): JSON configuration for overriding
-            RayConfig defaults. For testing purposes ONLY.
 
     Returns:
         A dictionary of the address information for the processes that were
             started.
     """
-    address_info = {
-        "redis_address": redis_address,
+    ray_params.address_info = {
+        "redis_address": ray_params.redis_address,
     }
-    return start_ray_processes(
-        address_info=address_info,
-        object_manager_ports=object_manager_ports,
-        node_manager_ports=node_manager_ports,
-        node_ip_address=node_ip_address,
-        num_workers=num_workers,
-        num_local_schedulers=num_local_schedulers,
-        object_store_memory=object_store_memory,
-        redis_password=redis_password,
-        worker_path=worker_path,
-        include_log_monitor=True,
-        cleanup=cleanup,
-        redirect_worker_output=redirect_worker_output,
-        redirect_output=redirect_output,
-        resources=resources,
-        plasma_directory=plasma_directory,
-        huge_pages=huge_pages,
-        plasma_store_socket_name=plasma_store_socket_name,
-        raylet_socket_name=raylet_socket_name,
-        temp_dir=temp_dir,
-        _internal_config=_internal_config)
+    ray_params.update(include_log_monitor=True)
+    return start_ray_processes(ray_params, cleanup=cleanup)
 
 
-def start_ray_head(address_info=None,
-                   object_manager_ports=None,
-                   node_manager_ports=None,
-                   node_ip_address="127.0.0.1",
-                   redis_port=None,
-                   redis_shard_ports=None,
-                   num_workers=None,
-                   num_local_schedulers=1,
-                   object_store_memory=None,
-                   redis_max_memory=None,
-                   collect_profiling_data=True,
-                   worker_path=None,
-                   cleanup=True,
-                   redirect_worker_output=False,
-                   redirect_output=False,
-                   start_workers_from_local_scheduler=True,
-                   resources=None,
-                   num_redis_shards=None,
-                   redis_max_clients=None,
-                   redis_password=None,
-                   include_webui=True,
-                   plasma_directory=None,
-                   huge_pages=False,
-                   autoscaling_config=None,
-                   plasma_store_socket_name=None,
-                   raylet_socket_name=None,
-                   temp_dir=None,
-                   _internal_config=None):
+def start_ray_head(ray_params, cleanup=True):
     """Start Ray in local mode.
 
     Args:
-        address_info (dict): A dictionary with address information for
-            processes that have already been started. If provided, address_info
-            will be modified to include processes that are newly started.
-        object_manager_ports (list): A list of the ports to use for the object
-            managers. There should be one per object manager being started on
-            this node (typically just one).
-        node_manager_ports (list): A list of the ports to use for the node
-            managers. There should be one per node manager being started on
-            this node (typically just one).
-        node_ip_address (str): The IP address of this node.
-        redis_port (int): The port that the primary Redis shard should listen
-            to. If None, then a random port will be chosen. If the key
-            "redis_address" is in address_info, then this argument will be
-            ignored.
-        redis_shard_ports: A list of the ports to use for the non-primary Redis
-            shards.
-        num_workers (int): The number of workers to start.
-        num_local_schedulers (int): The total number of local schedulers
-            required. This is also the total number of object stores required.
-            This method will start new instances of local schedulers and object
-            stores until there are at least num_local_schedulers existing
-            instances of each, including ones already registered with the given
-            address_info.
-        object_store_memory: The amount of memory (in bytes) to start the
-            object store with.
-        redis_max_memory: The max amount of memory (in bytes) to allow redis
-            to use, or None for no limit. Once the limit is exceeded, redis
-            will start LRU eviction of entries. This only applies to the
-            sharded redis tables (task and object tables).
-        collect_profiling_data: Whether to collect profiling data from workers.
-        worker_path (str): The path of the source code that will be run by the
-            worker.
+        ray_params (ray.params.RayParams): The RayParams instance. The
+            following parameters could be checked: address_info,
+            object_manager_ports, node_manager_ports, node_ip_address,
+            redis_port, redis_shard_ports, num_workers, num_local_schedulers,
+            object_store_memory, redis_max_memory, collect_profiling_data,
+            worker_path, cleanup, redirect_worker_output, redirect_output,
+            start_workers_from_local_scheduler, resources, num_redis_shards,
+            redis_max_clients, redis_password, include_webui, huge_pages,
+            plasma_directory, autoscaling_config, plasma_store_socket_name,
+            raylet_socket_name, temp_dir, _internal_config
         cleanup (bool): If cleanup is true, then the processes started here
             will be killed by services.cleanup() when the Python process that
             called this method exits.
-        redirect_worker_output: True if the stdout and stderr of worker
-            processes should be redirected to files.
-        redirect_output (bool): True if stdout and stderr for non-worker
-            processes should be redirected to files and false otherwise.
-        start_workers_from_local_scheduler (bool): If this flag is True, then
-            start the initial workers from the local scheduler. Else, start
-            them from Python.
-        resources: A dictionary mapping resource name to the available quantity
-            of that resource.
-        num_redis_shards: The number of Redis shards to start in addition to
-            the primary Redis shard.
-        redis_max_clients: If provided, attempt to configure Redis with this
-            maxclients number.
-        redis_password (str): Prevents external clients without the password
-            from connecting to Redis if provided.
-        include_webui: True if the UI should be started and false otherwise.
-        plasma_directory: A directory where the Plasma memory mapped files will
-            be created.
-        huge_pages: Boolean flag indicating whether to start the Object
-            Store with hugetlbfs support. Requires plasma_directory.
-        autoscaling_config: path to autoscaling config file.
-        plasma_store_socket_name (str): If provided, it will specify the socket
-            name used by the plasma store.
-        raylet_socket_name (str): If provided, it will specify the socket path
-            used by the raylet process.
-        temp_dir (str): If provided, it will specify the root temporary
-            directory for the Ray process.
-        _internal_config (str): JSON configuration for overriding
-            RayConfig defaults. For testing purposes ONLY.
 
     Returns:
         A dictionary of the address information for the processes that were
             started.
     """
-    num_redis_shards = 1 if num_redis_shards is None else num_redis_shards
-    return start_ray_processes(
-        address_info=address_info,
-        object_manager_ports=object_manager_ports,
-        node_manager_ports=node_manager_ports,
-        node_ip_address=node_ip_address,
-        redis_port=redis_port,
-        redis_shard_ports=redis_shard_ports,
-        num_workers=num_workers,
-        num_local_schedulers=num_local_schedulers,
-        object_store_memory=object_store_memory,
-        redis_max_memory=redis_max_memory,
-        collect_profiling_data=collect_profiling_data,
-        worker_path=worker_path,
-        cleanup=cleanup,
-        redirect_worker_output=redirect_worker_output,
-        redirect_output=redirect_output,
-        include_log_monitor=True,
-        include_webui=include_webui,
-        start_workers_from_local_scheduler=start_workers_from_local_scheduler,
-        resources=resources,
-        num_redis_shards=num_redis_shards,
-        redis_max_clients=redis_max_clients,
-        redis_password=redis_password,
-        plasma_directory=plasma_directory,
-        huge_pages=huge_pages,
-        autoscaling_config=autoscaling_config,
-        plasma_store_socket_name=plasma_store_socket_name,
-        raylet_socket_name=raylet_socket_name,
-        temp_dir=temp_dir,
-        _internal_config=_internal_config)
+    ray_params.update_if_absent(num_redis_shards=1, include_webui=True)
+    ray_params.update(include_log_monitor=True)
+    return start_ray_processes(ray_params, cleanup=cleanup)
diff --git a/python/ray/test/cluster_utils.py b/python/ray/test/cluster_utils.py
index 146000dae..9c98c57e6 100644
--- a/python/ray/test/cluster_utils.py
+++ b/python/ray/test/cluster_utils.py
@@ -7,6 +7,7 @@ import logging
 import time
 
 import ray
+from ray.parameter import RayParams
 import ray.services as services
 
 logger = logging.getLogger(__name__)
@@ -73,19 +74,18 @@ class Cluster(object):
             Node object of the added Ray node.
         """
         node_kwargs = {
-            "cleanup": True,
             "resources": {
                 "CPU": 1
             },
             "object_store_memory": 100 * (2**20)  # 100 MB
         }
         node_kwargs.update(override_kwargs)
+        ray_params = RayParams(
+            node_ip_address=services.get_node_ip_address(), **node_kwargs)
 
         if self.head_node is None:
-            address_info = services.start_ray_head(
-                node_ip_address=services.get_node_ip_address(),
-                include_webui=False,
-                **node_kwargs)
+            ray_params.update(include_webui=False)
+            address_info = services.start_ray_head(ray_params, cleanup=True)
             self.redis_address = address_info["redis_address"]
             # TODO(rliaw): Find a more stable way than modifying global state.
             process_dict_copy = services.all_processes.copy()
@@ -94,9 +94,8 @@ class Cluster(object):
             node = Node(address_info, process_dict_copy)
             self.head_node = node
         else:
-            address_info = services.start_ray_node(
-                services.get_node_ip_address(), self.redis_address,
-                **node_kwargs)
+            ray_params.update(redis_address=self.redis_address)
+            address_info = services.start_ray_node(ray_params, cleanup=True)
             # TODO(rliaw): Find a more stable way than modifying global state.
             process_dict_copy = services.all_processes.copy()
             for key in services.all_processes:
diff --git a/python/ray/worker.py b/python/ray/worker.py
index 9cdf3bbc9..aa45719ec 100644
--- a/python/ray/worker.py
+++ b/python/ray/worker.py
@@ -37,6 +37,7 @@ import ray.ray_constants as ray_constants
 from ray import import_thread
 from ray import profiling
 from ray.function_manager import (FunctionActorManager, FunctionDescriptor)
+from ray.parameter import RayParams
 from ray.utils import (
     check_oversized_pickle,
     is_cython,
@@ -1269,33 +1270,7 @@ def _normalize_resource_arguments(num_cpus, num_gpus, resources,
     return new_resources
 
 
-def _init(address_info=None,
-          start_ray_local=False,
-          object_id_seed=None,
-          num_workers=None,
-          num_local_schedulers=None,
-          object_store_memory=None,
-          redis_max_memory=None,
-          collect_profiling_data=True,
-          local_mode=False,
-          driver_mode=None,
-          redirect_worker_output=False,
-          redirect_output=True,
-          start_workers_from_local_scheduler=True,
-          num_cpus=None,
-          num_gpus=None,
-          resources=None,
-          num_redis_shards=None,
-          redis_max_clients=None,
-          redis_password=None,
-          plasma_directory=None,
-          huge_pages=False,
-          include_webui=True,
-          driver_id=None,
-          plasma_store_socket_name=None,
-          raylet_socket_name=None,
-          temp_dir=None,
-          _internal_config=None):
+def _init(ray_params, driver_id=None):
     """Helper method to connect to an existing Ray cluster or start a new one.
 
     This method handles two cases. Either a Ray cluster already exists and we
@@ -1303,67 +1278,17 @@ def _init(address_info=None,
     with a Ray cluster and attach to the newly started cluster.
 
     Args:
-        address_info (dict): A dictionary with address information for
-            processes in a partially-started Ray cluster. If
-            start_ray_local=True, any processes not in this dictionary will be
-            started. If provided, an updated address_info dictionary will be
-            returned to include processes that are newly started.
-        start_ray_local (bool): If True then this will start any processes not
-            already in address_info, including Redis, a global scheduler, local
-            scheduler(s), object store(s), and worker(s). It will also kill
-            these processes when Python exits. If False, this will attach to an
-            existing Ray cluster.
-        object_id_seed (int): Used to seed the deterministic generation of
-            object IDs. The same value can be used across multiple runs of the
-            same job in order to generate the object IDs in a consistent
-            manner. However, the same ID should not be used for different jobs.
-        num_local_schedulers (int): The number of local schedulers to start.
-            This is only provided if start_ray_local is True.
-        object_store_memory: The maximum amount of memory (in bytes) to
-            allow the object store to use.
-        redis_max_memory: The max amount of memory (in bytes) to allow redis
-            to use, or None for no limit. Once the limit is exceeded, redis
-            will start LRU eviction of entries. This only applies to the
-            sharded redis tables (task and object tables).
-        collect_profiling_data: Whether to collect profiling data from workers.
-        local_mode (bool): True if the code should be executed serially
-            without Ray. This is useful for debugging.
-        redirect_worker_output: True if the stdout and stderr of worker
-            processes should be redirected to files.
-        redirect_output (bool): True if stdout and stderr for non-worker
-            processes should be redirected to files and false otherwise.
-        start_workers_from_local_scheduler (bool): If this flag is True, then
-            start the initial workers from the local scheduler. Else, start
-            them from Python. The latter case is for debugging purposes only.
-        num_cpus (int): Number of cpus the user wishes all local schedulers to
-            be configured with.
-        num_gpus (int): Number of gpus the user wishes all local schedulers to
-            be configured with. If unspecified, Ray will attempt to autodetect
-            the number of GPUs available on the node (note that autodetection
-            currently only works for Nvidia GPUs).
-        resources: A dictionary mapping resource names to the quantity of that
-            resource available.
-        num_redis_shards: The number of Redis shards to start in addition to
-            the primary Redis shard.
-        redis_max_clients: If provided, attempt to configure Redis with this
-            maxclients number.
-        redis_password (str): Prevents external clients without the password
-            from connecting to Redis if provided.
-        plasma_directory: A directory where the Plasma memory mapped files will
-            be created.
-        huge_pages: Boolean flag indicating whether to start the Object
-            Store with hugetlbfs support. Requires plasma_directory.
-        include_webui: Boolean flag indicating whether to start the web
-            UI, which is a Jupyter notebook.
+        ray_params (ray.params.RayParams): The RayParams instance. The
+            following parameters could be checked: address_info,
+            start_ray_local, object_id_seed, num_workers,
+            num_local_schedulers, object_store_memory, redis_max_memory,
+            collect_profiling_data, local_mode, redirect_worker_output,
+            driver_mode, redirect_output, start_workers_from_local_scheduler,
+            num_cpus, num_gpus, resources, num_redis_shards,
+            redis_max_clients, redis_password, plasma_directory, huge_pages,
+            include_webui, driver_id, plasma_store_socket_name, temp_dir,
+            raylet_socket_name, _internal_config
         driver_id: The ID of driver.
-        plasma_store_socket_name (str): If provided, it will specify the socket
-            name used by the plasma store.
-        raylet_socket_name (str): If provided, it will specify the socket path
-            used by the raylet process.
-        temp_dir (str): If provided, it will specify the root temporary
-            directory for the Ray process.
-        _internal_config (str): JSON configuration for overriding
-            RayConfig defaults. For testing purposes ONLY.
 
     Returns:
         Address information about the started processes.
@@ -1372,157 +1297,137 @@ def _init(address_info=None,
         Exception: An exception is raised if an inappropriate combination of
             arguments is passed in.
     """
-    if driver_mode is not None:
+    if ray_params.driver_mode is not None:
         raise Exception("The 'driver_mode' argument has been deprecated. "
                         "To run Ray in local mode, pass in local_mode=True.")
-    if local_mode:
-        driver_mode = LOCAL_MODE
+    if ray_params.local_mode:
+        ray_params.driver_mode = LOCAL_MODE
     else:
-        driver_mode = SCRIPT_MODE
+        ray_params.driver_mode = SCRIPT_MODE
 
-    if redis_max_memory and collect_profiling_data:
+    if ray_params.redis_max_memory and ray_params.collect_profiling_data:
         logger.warning(
             "Profiling data cannot be LRU evicted, so it is disabled "
             "when redis_max_memory is set.")
-        collect_profiling_data = False
+        ray_params.collect_profiling_data = False
 
     # Get addresses of existing services.
-    if address_info is None:
-        address_info = {}
+    if ray_params.address_info is None:
+        ray_params.address_info = {}
     else:
-        assert isinstance(address_info, dict)
-    node_ip_address = address_info.get("node_ip_address")
-    redis_address = address_info.get("redis_address")
+        assert isinstance(ray_params.address_info, dict)
+    ray_params.node_ip_address = ray_params.address_info.get("node_ip_address")
+    ray_params.redis_address = ray_params.address_info.get("redis_address")
 
     # Start any services that do not yet exist.
-    if driver_mode == LOCAL_MODE:
+    if ray_params.driver_mode == LOCAL_MODE:
         # If starting Ray in LOCAL_MODE, don't start any other processes.
         pass
-    elif start_ray_local:
+    elif ray_params.start_ray_local:
         # In this case, we launch a scheduler, a new object store, and some
         # workers, and we connect to them. We do not launch any processes that
         # are already registered in address_info.
-        if node_ip_address is None:
-            node_ip_address = ray.services.get_node_ip_address()
+        ray_params.update_if_absent(
+            node_ip_address=ray.services.get_node_ip_address())
         # Use 1 local scheduler if num_local_schedulers is not provided. If
         # existing local schedulers are provided, use that count as
         # num_local_schedulers.
-        if num_local_schedulers is None:
-            num_local_schedulers = 1
+        ray_params.update_if_absent(num_local_schedulers=1)
         # Use 1 additional redis shard if num_redis_shards is not provided.
-        num_redis_shards = 1 if num_redis_shards is None else num_redis_shards
+        ray_params.update_if_absent(num_redis_shards=1)
 
         # Stick the CPU and GPU resources into the resource dictionary.
-        resources = _normalize_resource_arguments(
-            num_cpus, num_gpus, resources, num_local_schedulers)
+        ray_params.resources = _normalize_resource_arguments(
+            ray_params.num_cpus, ray_params.num_gpus, ray_params.resources,
+            ray_params.num_local_schedulers)
 
         # Start the scheduler, object store, and some workers. These will be
         # killed by the call to shutdown(), which happens when the Python
         # script exits.
-        address_info = services.start_ray_head(
-            address_info=address_info,
-            node_ip_address=node_ip_address,
-            num_workers=num_workers,
-            num_local_schedulers=num_local_schedulers,
-            object_store_memory=object_store_memory,
-            redis_max_memory=redis_max_memory,
-            collect_profiling_data=collect_profiling_data,
-            redirect_worker_output=redirect_worker_output,
-            redirect_output=redirect_output,
-            start_workers_from_local_scheduler=(
-                start_workers_from_local_scheduler),
-            resources=resources,
-            num_redis_shards=num_redis_shards,
-            redis_max_clients=redis_max_clients,
-            redis_password=redis_password,
-            plasma_directory=plasma_directory,
-            huge_pages=huge_pages,
-            include_webui=include_webui,
-            plasma_store_socket_name=plasma_store_socket_name,
-            raylet_socket_name=raylet_socket_name,
-            temp_dir=temp_dir,
-            _internal_config=_internal_config)
+        ray_params.address_info = services.start_ray_head(ray_params)
     else:
-        if redis_address is None:
+        if ray_params.redis_address is None:
             raise Exception("When connecting to an existing cluster, "
                             "redis_address must be provided.")
-        if num_workers is not None:
+        if ray_params.num_workers is not None:
             raise Exception("When connecting to an existing cluster, "
                             "num_workers must not be provided.")
-        if num_local_schedulers is not None:
+        if ray_params.num_local_schedulers is not None:
             raise Exception("When connecting to an existing cluster, "
                             "num_local_schedulers must not be provided.")
-        if num_cpus is not None or num_gpus is not None:
+        if ray_params.num_cpus is not None or ray_params.num_gpus is not None:
             raise Exception("When connecting to an existing cluster, num_cpus "
                             "and num_gpus must not be provided.")
-        if resources is not None:
+        if ray_params.resources is not None:
             raise Exception("When connecting to an existing cluster, "
                             "resources must not be provided.")
-        if num_redis_shards is not None:
+        if ray_params.num_redis_shards is not None:
             raise Exception("When connecting to an existing cluster, "
                             "num_redis_shards must not be provided.")
-        if redis_max_clients is not None:
+        if ray_params.redis_max_clients is not None:
             raise Exception("When connecting to an existing cluster, "
                             "redis_max_clients must not be provided.")
-        if object_store_memory is not None:
+        if ray_params.object_store_memory is not None:
             raise Exception("When connecting to an existing cluster, "
                             "object_store_memory must not be provided.")
-        if redis_max_memory is not None:
+        if ray_params.redis_max_memory is not None:
             raise Exception("When connecting to an existing cluster, "
                             "redis_max_memory must not be provided.")
-        if plasma_directory is not None:
+        if ray_params.plasma_directory is not None:
             raise Exception("When connecting to an existing cluster, "
                             "plasma_directory must not be provided.")
-        if huge_pages:
+        if ray_params.huge_pages:
             raise Exception("When connecting to an existing cluster, "
                             "huge_pages must not be provided.")
-        if temp_dir is not None:
+        if ray_params.temp_dir is not None:
             raise Exception("When connecting to an existing cluster, "
                             "temp_dir must not be provided.")
-        if plasma_store_socket_name is not None:
+        if ray_params.plasma_store_socket_name is not None:
             raise Exception("When connecting to an existing cluster, "
                             "plasma_store_socket_name must not be provided.")
-        if raylet_socket_name is not None:
+        if ray_params.raylet_socket_name is not None:
             raise Exception("When connecting to an existing cluster, "
                             "raylet_socket_name must not be provided.")
-        if _internal_config is not None:
+        if ray_params._internal_config is not None:
             raise Exception("When connecting to an existing cluster, "
                             "_internal_config must not be provided.")
 
         # Get the node IP address if one is not provided.
-        if node_ip_address is None:
-            node_ip_address = services.get_node_ip_address(redis_address)
+        ray_params.update_if_absent(
+            node_ip_address=services.get_node_ip_address(
+                ray_params.redis_address))
         # Get the address info of the processes to connect to from Redis.
-        address_info = get_address_info_from_redis(
-            redis_address, node_ip_address, redis_password=redis_password)
+        ray_params.address_info = get_address_info_from_redis(
+            ray_params.redis_address,
+            ray_params.node_ip_address,
+            redis_password=ray_params.redis_password)
 
     # Connect this driver to Redis, the object store, and the local scheduler.
     # Choose the first object store and local scheduler if there are multiple.
     # The corresponding call to disconnect will happen in the call to
     # shutdown() when the Python script exits.
-    if driver_mode == LOCAL_MODE:
+    if ray_params.driver_mode == LOCAL_MODE:
         driver_address_info = {}
     else:
         driver_address_info = {
-            "node_ip_address": node_ip_address,
-            "redis_address": address_info["redis_address"],
-            "store_socket_name": address_info["object_store_addresses"][0],
-            "webui_url": address_info["webui_url"],
+            "node_ip_address": ray_params.node_ip_address,
+            "redis_address": ray_params.address_info["redis_address"],
+            "store_socket_name": ray_params.address_info[
+                "object_store_addresses"][0],
+            "webui_url": ray_params.address_info["webui_url"],
         }
         driver_address_info["raylet_socket_name"] = (
-            address_info["raylet_socket_names"][0])
+            ray_params.address_info["raylet_socket_names"][0])
 
     # We only pass `temp_dir` to a worker (WORKER_MODE).
     # It can't be a worker here.
     connect(
+        ray_params,
         driver_address_info,
-        object_id_seed=object_id_seed,
-        mode=driver_mode,
+        mode=ray_params.driver_mode,
         worker=global_worker,
-        driver_id=driver_id,
-        redis_password=redis_password,
-        collect_profiling_data=collect_profiling_data)
-    return address_info
+        driver_id=driver_id)
+    return ray_params.address_info
 
 
 def init(redis_address=None,
@@ -1674,7 +1579,7 @@ def init(redis_address=None,
         redis_address = services.address_to_ip(redis_address)
 
     info = {"node_ip_address": node_ip_address, "redis_address": redis_address}
-    ret = _init(
+    ray_params = RayParams(
         address_info=info,
         start_ray_local=(redis_address is None),
         num_workers=num_workers,
@@ -1695,11 +1600,12 @@ def init(redis_address=None,
         object_store_memory=object_store_memory,
         redis_max_memory=redis_max_memory,
         collect_profiling_data=collect_profiling_data,
-        driver_id=driver_id,
         plasma_store_socket_name=plasma_store_socket_name,
         raylet_socket_name=raylet_socket_name,
         temp_dir=temp_dir,
-        _internal_config=_internal_config)
+        _internal_config=_internal_config,
+    )
+    ret = _init(ray_params, driver_id=driver_id)
     for hook in _post_init_hooks:
         hook()
     return ret
@@ -1900,26 +1806,23 @@ def print_error_messages(worker):
         pass
 
 
-def connect(info,
-            object_id_seed=None,
+def connect(ray_params,
+            info,
             mode=WORKER_MODE,
             worker=global_worker,
-            driver_id=None,
-            redis_password=None,
-            collect_profiling_data=True):
+            driver_id=None):
     """Connect this worker to the local scheduler, to Plasma, and to Redis.
 
     Args:
+        ray_params (ray.params.RayParams): The RayParams instance. The
+            following parameters could be checked: object_id_seed,
+            redis_password, collect_profiling_data
         info (dict): A dictionary with address of the Redis server and the
             sockets of the plasma store and raylet.
-        object_id_seed: A seed to use to make the generation of object IDs
-            deterministic.
         mode: The mode of the worker. One of SCRIPT_MODE, WORKER_MODE, and
             LOCAL_MODE.
+        worker: The ray.Worker instance.
         driver_id: The ID of driver. If it's None, then we will generate one.
-        redis_password (str): Prevents external clients without the password
-            from connecting to Redis if provided.
-        collect_profiling_data: Whether to collect profiling data from workers.
     """
     # Do some basic checking to make sure we didn't call ray.init twice.
     error_message = "Perhaps you called ray.init twice by accident?"
@@ -1929,7 +1832,7 @@ def connect(info,
     # Enable nice stack traces on SIGSEGV etc.
     faulthandler.enable(all_threads=False)
 
-    if collect_profiling_data:
+    if ray_params.collect_profiling_data:
         worker.profiler = profiling.Profiler(worker)
     else:
         worker.profiler = profiling.NoopProfiler()
@@ -1977,7 +1880,7 @@ def connect(info,
         redis.StrictRedis(
             host=redis_ip_address,
             port=int(redis_port),
-            password=redis_password))
+            password=ray_params.redis_password))
 
     # For driver's check that the version information matches the version
     # information that the Ray cluster was started with.
@@ -2015,11 +1918,13 @@ def connect(info,
             services.record_log_files_in_redis(
                 info["redis_address"],
                 info["node_ip_address"], [log_stdout_file, log_stderr_file],
-                password=redis_password)
+                password=ray_params.redis_password)
 
     # Create an object for interfacing with the global state.
     global_state._initialize_global_state(
-        redis_ip_address, int(redis_port), redis_password=redis_password)
+        redis_ip_address,
+        int(redis_port),
+        redis_password=ray_params.redis_password)
 
     # Register the worker with Redis.
     if mode == SCRIPT_MODE:
@@ -2068,8 +1973,8 @@ def connect(info,
         # the user's random number generator). Otherwise, set the current task
         # ID randomly to avoid object ID collisions.
         numpy_state = np.random.get_state()
-        if object_id_seed is not None:
-            np.random.seed(object_id_seed)
+        if ray_params.object_id_seed is not None:
+            np.random.seed(ray_params.object_id_seed)
         else:
             # Try to use true randomness.
             np.random.seed(None)
diff --git a/python/ray/workers/default_worker.py b/python/ray/workers/default_worker.py
index dc1085783..f165be673 100644
--- a/python/ray/workers/default_worker.py
+++ b/python/ray/workers/default_worker.py
@@ -8,6 +8,7 @@ import traceback
 
 import ray
 import ray.actor
+from ray.parameter import RayParams
 import ray.ray_constants as ray_constants
 import ray.tempfile_services as tempfile_services
 
@@ -35,11 +36,6 @@ parser.add_argument(
     required=True,
     type=str,
     help="the object store's name")
-parser.add_argument(
-    "--object-store-manager-name",
-    required=False,
-    type=str,
-    help="the object store manager's name")
 parser.add_argument(
     "--raylet-name", required=False, type=str, help="the raylet's name")
 parser.add_argument(
@@ -75,7 +71,6 @@ if __name__ == "__main__":
         "redis_address": args.redis_address,
         "redis_password": args.redis_password,
         "store_socket_name": args.object_store_name,
-        "manager_socket_name": args.object_store_manager_name,
         "raylet_socket_name": args.raylet_name,
     }
 
@@ -86,11 +81,15 @@ if __name__ == "__main__":
     # Override the temporary directory.
     tempfile_services.set_temp_root(args.temp_dir)
 
-    ray.worker.connect(
-        info,
-        mode=ray.WORKER_MODE,
+    ray_params = RayParams(
+        node_ip_address=args.node_ip_address,
+        redis_address=args.redis_address,
         redis_password=args.redis_password,
-        collect_profiling_data=args.collect_profiling_data)
+        plasma_store_socket_name=args.object_store_name,
+        raylet_socket_name=args.raylet_name,
+        temp_dir=args.temp_dir)
+
+    ray.worker.connect(ray_params, info, mode=ray.WORKER_MODE)
 
     error_explanation = """
   This error is unexpected and should not have happened. Somehow a worker
diff --git a/test/actor_test.py b/test/actor_test.py
index 9ce1bf8fb..29e3b1085 100644
--- a/test/actor_test.py
+++ b/test/actor_test.py
@@ -13,6 +13,7 @@ import sys
 import time
 
 import ray
+from ray.parameter import RayParams
 import ray.ray_constants as ray_constants
 import ray.test.test_utils
 import ray.test.cluster_utils
@@ -742,10 +743,11 @@ def test_actors_on_nodes_with_no_cpus(ray_start_regular):
 
 def test_actor_load_balancing(shutdown_only):
     num_local_schedulers = 3
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_cpus=1,
         num_local_schedulers=num_local_schedulers)
+    ray.worker._init(ray_params)
 
     @ray.remote
     class Actor1(object):
@@ -788,11 +790,12 @@ def test_actor_load_balancing(shutdown_only):
 def test_actor_gpus(shutdown_only):
     num_local_schedulers = 3
     num_gpus_per_scheduler = 4
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
         num_cpus=(num_local_schedulers * [10 * num_gpus_per_scheduler]),
         num_gpus=(num_local_schedulers * [num_gpus_per_scheduler]))
+    ray.worker._init(ray_params)
 
     @ray.remote(num_gpus=1)
     class Actor1(object):
@@ -830,11 +833,12 @@ def test_actor_gpus(shutdown_only):
 def test_actor_multiple_gpus(shutdown_only):
     num_local_schedulers = 3
     num_gpus_per_scheduler = 5
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
         num_cpus=(num_local_schedulers * [10 * num_gpus_per_scheduler]),
         num_gpus=(num_local_schedulers * [num_gpus_per_scheduler]))
+    ray.worker._init(ray_params)
 
     @ray.remote(num_gpus=2)
     class Actor1(object):
@@ -900,11 +904,12 @@ def test_actor_multiple_gpus(shutdown_only):
 def test_actor_different_numbers_of_gpus(shutdown_only):
     # Test that we can create actors on two nodes that have different
     # numbers of GPUs.
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=3,
         num_cpus=[10, 10, 10],
         num_gpus=[0, 5, 10])
+    ray.worker._init(ray_params)
 
     @ray.remote(num_gpus=1)
     class Actor1(object):
@@ -940,7 +945,7 @@ def test_actor_different_numbers_of_gpus(shutdown_only):
 def test_actor_multiple_gpus_from_multiple_tasks(shutdown_only):
     num_local_schedulers = 5
     num_gpus_per_scheduler = 5
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
         redirect_output=True,
@@ -949,6 +954,7 @@ def test_actor_multiple_gpus_from_multiple_tasks(shutdown_only):
         _internal_config=json.dumps({
             "num_heartbeats_timeout": 1000
         }))
+    ray.worker._init(ray_params)
 
     @ray.remote
     def create_actors(i, n):
@@ -1020,11 +1026,12 @@ def test_actor_multiple_gpus_from_multiple_tasks(shutdown_only):
 def test_actors_and_tasks_with_gpus(shutdown_only):
     num_local_schedulers = 3
     num_gpus_per_scheduler = 6
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
         num_cpus=num_gpus_per_scheduler,
         num_gpus=(num_local_schedulers * [num_gpus_per_scheduler]))
+    ray.worker._init(ray_params)
 
     def check_intervals_non_overlapping(list_of_intervals):
         for i in range(len(list_of_intervals)):
@@ -1387,11 +1394,12 @@ def test_reconstruction_suppression(head_node_cluster):
 def setup_counter_actor(test_checkpoint=False,
                         save_exception=False,
                         resume_exception=False):
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=2,
         num_cpus=1,
         redirect_output=True)
+    ray.worker._init(ray_params)
 
     # Only set the checkpoint interval if we're testing with checkpointing.
     checkpoint_interval = -1
@@ -1721,11 +1729,12 @@ def test_checkpoint_distributed_handle(shutdown_only):
 
 def _test_nondeterministic_reconstruction(num_forks, num_items_per_fork,
                                           num_forks_to_wait):
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=2,
         num_cpus=1,
         redirect_output=True)
+    ray.worker._init(ray_params)
 
     # Make a shared queue.
     @ray.remote
@@ -2019,7 +2028,7 @@ def test_lifetime_and_transient_resources(ray_start_regular):
 
 
 def test_custom_label_placement(shutdown_only):
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=2,
         num_cpus=2,
@@ -2028,6 +2037,7 @@ def test_custom_label_placement(shutdown_only):
         }, {
             "CustomResource2": 2
         }])
+    ray.worker._init(ray_params)
 
     @ray.remote(resources={"CustomResource1": 1})
     class ResourceActor1(object):
diff --git a/test/array_test.py b/test/array_test.py
index e5838f89d..54f1ebfc2 100644
--- a/test/array_test.py
+++ b/test/array_test.py
@@ -10,6 +10,7 @@ import sys
 import ray
 import ray.experimental.array.remote as ra
 import ray.experimental.array.distributed as da
+from ray.parameter import RayParams
 
 if sys.version_info >= (3, 0):
     from importlib import reload
@@ -74,8 +75,9 @@ def ray_start_two_nodes():
     ]:
         reload(module)
     # Start the Ray processes.
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True, num_local_schedulers=2, num_cpus=[10, 10])
+    ray.worker._init(ray_params)
     yield None
     # The code after the yield will run as teardown code.
     ray.shutdown()
diff --git a/test/component_failures_test.py b/test/component_failures_test.py
index 30071b3c1..d42f5e9bb 100644
--- a/test/component_failures_test.py
+++ b/test/component_failures_test.py
@@ -12,6 +12,7 @@ import numpy as np
 import pytest
 
 import ray
+from ray.parameter import RayParams
 from ray.test.cluster_utils import Cluster
 from ray.test.test_utils import run_string_as_driver_nonblocking
 
@@ -19,11 +20,9 @@ from ray.test.test_utils import run_string_as_driver_nonblocking
 @pytest.fixture
 def ray_start_workers_separate():
     # Start the Ray processes.
-    ray.worker._init(
-        num_cpus=1,
-        start_workers_from_local_scheduler=False,
-        start_ray_local=True,
-        redirect_output=True)
+    ray_params = RayParams(
+        num_cpus=1, start_ray_local=True, redirect_output=True)
+    ray.worker._init(ray_params)
     yield None
     # The code after the yield will run as teardown code.
     ray.shutdown()
@@ -239,12 +238,12 @@ def ray_start_workers_separate_multinode(request):
     num_local_schedulers = request.param[0]
     num_initial_workers = request.param[1]
     # Start the Ray processes.
-    ray.worker._init(
+    ray_params = RayParams(
         num_local_schedulers=num_local_schedulers,
-        start_workers_from_local_scheduler=False,
         start_ray_local=True,
         num_cpus=[num_initial_workers] * num_local_schedulers,
         redirect_output=True)
+    ray.worker._init(ray_params)
     yield num_local_schedulers, num_initial_workers
     # The code after the yield will run as teardown code.
     ray.shutdown()
@@ -282,7 +281,7 @@ def _test_component_failed(component_type):
     # Start with 4 workers and 4 cores.
     num_local_schedulers = 4
     num_workers_per_scheduler = 8
-    ray.worker._init(
+    ray_params = RayParams(
         num_local_schedulers=num_local_schedulers,
         start_ray_local=True,
         num_cpus=[num_workers_per_scheduler] * num_local_schedulers,
@@ -291,6 +290,7 @@ def _test_component_failed(component_type):
             "initial_reconstruction_timeout_milliseconds": 1000,
             "num_heartbeats_timeout": 10,
         }))
+    ray.worker._init(ray_params)
 
     # Submit many tasks with many dependencies.
     @ray.remote
diff --git a/test/failure_test.py b/test/failure_test.py
index 3efb9bc69..e3b5223bc 100644
--- a/test/failure_test.py
+++ b/test/failure_test.py
@@ -11,6 +11,7 @@ import tempfile
 import threading
 import time
 
+from ray.parameter import RayParams
 import ray.ray_constants as ray_constants
 from ray.utils import _random_string
 import pytest
@@ -573,13 +574,14 @@ def test_warning_for_infeasible_zero_cpu_actor(shutdown_only):
 @pytest.fixture
 def ray_start_two_nodes():
     # Start the Ray processes.
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=2,
         num_cpus=0,
         _internal_config=json.dumps({
             "num_heartbeats_timeout": 40
         }))
+    ray.worker._init(ray_params)
     yield None
     # The code after the yield will run as teardown code.
     ray.shutdown()
diff --git a/test/runtest.py b/test/runtest.py
index 91862023b..7ed2ab204 100644
--- a/test/runtest.py
+++ b/test/runtest.py
@@ -18,6 +18,7 @@ import numpy as np
 import pytest
 
 import ray
+from ray.parameter import RayParams
 import ray.ray_constants as ray_constants
 import ray.test.cluster_utils
 import ray.test.test_utils
@@ -308,10 +309,8 @@ def test_python_workers(shutdown_only):
     # instead of the local scheduler. This codepath is for debugging
     # purposes only.
     num_workers = 4
-    ray.worker._init(
-        num_cpus=num_workers,
-        start_workers_from_local_scheduler=False,
-        start_ray_local=True)
+    ray_params = RayParams(num_cpus=num_workers, start_ray_local=True)
+    ray.worker._init(ray_params)
 
     @ray.remote
     def f(x):
@@ -1263,7 +1262,7 @@ def test_free_objects_multi_node(shutdown_only):
     # workers and the plasma client holding the deletion target
     # may not be flushed.
     config = json.dumps({"object_manager_repeated_push_delay_ms": 1000})
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=3,
         num_cpus=[1, 1, 1],
@@ -1275,6 +1274,7 @@ def test_free_objects_multi_node(shutdown_only):
             "Custom2": 1
         }],
         _internal_config=config)
+    ray.worker._init(ray_params)
 
     @ray.remote(resources={"Custom0": 1})
     class ActorOnNode0(object):
@@ -1719,8 +1719,9 @@ def test_zero_cpus(shutdown_only):
 
 
 def test_zero_cpus_actor(shutdown_only):
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True, num_local_schedulers=2, num_cpus=[0, 2])
+    ray.worker._init(ray_params)
 
     local_plasma = ray.worker.global_worker.plasma_client.store_socket_name
 
@@ -1789,11 +1790,12 @@ def test_multiple_local_schedulers(shutdown_only):
     # This test will define a bunch of tasks that can only be assigned to
     # specific local schedulers, and we will check that they are assigned
     # to the correct local schedulers.
-    address_info = ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=3,
         num_cpus=[11, 5, 10],
         num_gpus=[0, 5, 1])
+    address_info = ray.worker._init(ray_params)
 
     # Define a bunch of remote functions that all return the socket name of
     # the plasma store. Since there is a one-to-one correspondence between
@@ -1897,7 +1899,7 @@ def test_multiple_local_schedulers(shutdown_only):
 
 
 def test_custom_resources(shutdown_only):
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=2,
         num_cpus=[3, 3],
@@ -1906,6 +1908,7 @@ def test_custom_resources(shutdown_only):
         }, {
             "CustomResource": 1
         }])
+    ray.worker._init(ray_params)
 
     @ray.remote
     def f():
@@ -1938,7 +1941,7 @@ def test_custom_resources(shutdown_only):
 
 
 def test_two_custom_resources(shutdown_only):
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=2,
         num_cpus=[3, 3],
@@ -1949,6 +1952,7 @@ def test_two_custom_resources(shutdown_only):
             "CustomResource1": 3,
             "CustomResource2": 4
         }])
+    ray.worker._init(ray_params)
 
     @ray.remote(resources={"CustomResource1": 1})
     def f():
@@ -2149,10 +2153,11 @@ def test_load_balancing(shutdown_only):
     # schedulers in a roughly equal manner.
     num_local_schedulers = 3
     num_cpus = 7
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
         num_cpus=num_cpus)
+    ray.worker._init(ray_params)
 
     @ray.remote
     def f():
@@ -2168,10 +2173,11 @@ def test_load_balancing_with_dependencies(shutdown_only):
     # schedulers in a roughly equal manner even when the tasks have
     # dependencies.
     num_local_schedulers = 3
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
         num_cpus=1)
+    ray.worker._init(ray_params)
 
     @ray.remote
     def f(x):
diff --git a/test/stress_tests.py b/test/stress_tests.py
index 3771f5805..22edf4a6b 100644
--- a/test/stress_tests.py
+++ b/test/stress_tests.py
@@ -9,6 +9,7 @@ import pytest
 import time
 
 import ray
+from ray.parameter import RayParams
 import ray.tempfile_services
 import ray.ray_constants as ray_constants
 
@@ -36,10 +37,11 @@ def ray_start_combination(request):
     num_local_schedulers = request.param[0]
     num_workers_per_scheduler = request.param[1]
     # Start the Ray processes.
-    ray.worker._init(
+    ray_params = RayParams(
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
         num_cpus=10)
+    ray.worker._init(ray_params)
     yield num_local_schedulers, num_workers_per_scheduler
     # The code after the yield will run as teardown code.
     ray.shutdown()
@@ -212,7 +214,7 @@ def ray_start_reconstruction(request):
         "redis_shards": redis_shards,
         "object_store_addresses": plasma_addresses
     }
-    ray.worker._init(
+    ray_params = RayParams(
         address_info=address_info,
         start_ray_local=True,
         num_local_schedulers=num_local_schedulers,
@@ -221,6 +223,7 @@ def ray_start_reconstruction(request):
         _internal_config=json.dumps({
             "initial_reconstruction_timeout_milliseconds": 200
         }))
+    ray.worker._init(ray_params)
 
     yield (redis_ip_address, redis_port, plasma_store_memory,
            num_local_schedulers)