diff --git a/java/test/src/main/resources/test_cross_language_invocation.py b/java/test/src/main/resources/test_cross_language_invocation.py
index 53f38e86a..8b06aa661 100644
--- a/java/test/src/main/resources/test_cross_language_invocation.py
+++ b/java/test/src/main/resources/test_cross_language_invocation.py
@@ -1,8 +1,6 @@
 # This file is used by CrossLanguageInvocationTest.java to test cross-language
 # invocation.
 
-import six
-
 import ray
 
 
@@ -63,4 +61,4 @@ class Counter(object):
 
     def increase(self, delta):
         self.value += int(delta)
-        return str(self.value).encode("utf-8") if six.PY3 else str(self.value)
+        return str(self.value).encode("utf-8")
diff --git a/python/ray/actor.py b/python/ray/actor.py
index d615b5651..879498725 100644
--- a/python/ray/actor.py
+++ b/python/ray/actor.py
@@ -1,7 +1,6 @@
 import copy
 import inspect
 import logging
-import six
 import weakref
 
 from abc import ABCMeta, abstractmethod
@@ -963,7 +962,7 @@ Checkpoint = namedtuple(
 """A namedtuple that represents a checkpoint."""
 
 
-class Checkpointable(six.with_metaclass(ABCMeta, object)):
+class Checkpointable(metaclass=ABCMeta):
     """An interface that indicates an actor can be checkpointed."""
 
     @abstractmethod
diff --git a/python/ray/tests/test_advanced.py b/python/ray/tests/test_advanced.py
index fa3707c7e..cc1afe75f 100644
--- a/python/ray/tests/test_advanced.py
+++ b/python/ray/tests/test_advanced.py
@@ -3,7 +3,6 @@ from concurrent.futures import ThreadPoolExecutor
 import json
 import logging
 import random
-import six
 import sys
 import threading
 import time
@@ -365,10 +364,6 @@ def test_illegal_api_calls(ray_start_regular):
         ray.get(3)
 
 
-# TODO(hchen): This test currently doesn't work in Python 2. This is likely
-# because plasma client isn't thread-safe. This needs to be fixed from the
-# Arrow side. See #4107 for relevant discussions.
-@pytest.mark.skipif(six.PY2, reason="Doesn't work in Python 2.")
 def test_multithreading(ray_start_2_cpus):
     # This test requires at least 2 CPUs to finish since the worker does not
     # release resources when joining the threads.
diff --git a/python/ray/tune/examples/durable_trainable_example.py b/python/ray/tune/examples/durable_trainable_example.py
index 97bad8af1..ab7bb13d8 100644
--- a/python/ray/tune/examples/durable_trainable_example.py
+++ b/python/ray/tune/examples/durable_trainable_example.py
@@ -8,7 +8,7 @@ from ray import tune
 from ray.tune import DurableTrainable
 from ray.tune.sync_client import get_sync_client
 
-import cloudpickle
+from ray import cloudpickle
 
 logger = logging.getLogger(__name__)
 
diff --git a/python/ray/tune/experiment.py b/python/ray/tune/experiment.py
index dc2b940ed..78c85ce1b 100644
--- a/python/ray/tune/experiment.py
+++ b/python/ray/tune/experiment.py
@@ -1,7 +1,6 @@
 import copy
 import logging
 import os
-import six
 
 from ray.tune.error import TuneError
 from ray.tune.registry import register_trainable, get_trainable_cls
@@ -189,7 +188,7 @@ class Experiment:
             A string representing the trainable identifier.
         """
 
-        if isinstance(run_object, six.string_types):
+        if isinstance(run_object, str):
             return run_object
         elif isinstance(run_object, sample_from):
             logger.warning("Not registering trainable. Resolving as variant.")
diff --git a/python/ray/tune/tune.py b/python/ray/tune/tune.py
index 6b14a60db..a96be49ad 100644
--- a/python/ray/tune/tune.py
+++ b/python/ray/tune/tune.py
@@ -1,5 +1,4 @@
 import logging
-import six
 
 from ray.tune.error import TuneError
 from ray.tune.experiment import convert_to_experiment_list, Experiment
@@ -41,7 +40,7 @@ def _make_scheduler(args):
 
 
 def _check_default_resources_override(run_identifier):
-    if not isinstance(run_identifier, six.string_types):
+    if not isinstance(run_identifier, str):
         # If obscure dtype, assume it is overriden.
         return True
     trainable_cls = get_trainable_cls(run_identifier)
diff --git a/python/ray/utils.py b/python/ray/utils.py
index f796fce51..8ae693bbf 100644
--- a/python/ray/utils.py
+++ b/python/ray/utils.py
@@ -5,7 +5,6 @@ import inspect
 import logging
 import numpy as np
 import os
-import six
 import subprocess
 import sys
 import tempfile
@@ -223,32 +222,14 @@ def decode(byte_str, allow_none=False):
 def ensure_str(s, encoding="utf-8", errors="strict"):
     """Coerce *s* to `str`.
 
-    To keep six with lower version, see Issue 4169, we copy this function
-    from six == 1.12.0.
-
-    TODO(yuhguo): remove this function when six >= 1.12.0.
-
-    For Python 2:
-      - `unicode` -> encoded to `str`
-      - `str` -> `str`
-
-    For Python 3:
       - `str` -> `str`
       - `bytes` -> decoded to `str`
     """
-    if six.PY3:
-        text_type = str
-        binary_type = bytes
+    if isinstance(s, str):
+        return s
     else:
-        text_type = unicode  # noqa: F821
-        binary_type = str
-    if not isinstance(s, (text_type, binary_type)):
-        raise TypeError("not expecting type '%s'" % type(s))
-    if six.PY2 and isinstance(s, text_type):
-        s = s.encode(encoding, errors)
-    elif six.PY3 and isinstance(s, binary_type):
-        s = s.decode(encoding, errors)
-    return s
+        assert isinstance(s, bytes)
+        return s.decode(encoding, errors)
 
 
 def binary_to_object_id(binary_object_id):
diff --git a/python/setup.py b/python/setup.py
index 985f4d608..e84cfe59f 100644
--- a/python/setup.py
+++ b/python/setup.py
@@ -171,24 +171,9 @@ def find_version(*filepath):
 
 
 requires = [
-    "numpy >= 1.16",
-    "filelock",
-    "jsonschema",
-    "click",
-    "colorama",
-    "packaging",
-    "pyyaml",
-    "redis>=3.3.2",
-    # NOTE: Don't upgrade the version of six! Doing so causes installation
-    # problems. See https://github.com/ray-project/ray/issues/4169.
-    "six >= 1.0.0",
-    "faulthandler;python_version<'3.3'",
-    "protobuf >= 3.8.0",
-    "cloudpickle",
-    "py-spy >= 0.2.0",
-    "aiohttp",
-    "google",
-    "grpcio"
+    "numpy >= 1.16", "filelock", "jsonschema", "click", "colorama",
+    "packaging", "pyyaml", "redis >= 3.3.2", "protobuf >= 3.8.0",
+    "py-spy >= 0.2.0", "aiohttp", "google", "grpcio"
 ]
 
 setup(
diff --git a/rllib/agents/trainer.py b/rllib/agents/trainer.py
index c952cb376..1a74eacbd 100644
--- a/rllib/agents/trainer.py
+++ b/rllib/agents/trainer.py
@@ -4,7 +4,6 @@ import logging
 import math
 import os
 import pickle
-import six
 import time
 import tempfile
 
@@ -1022,7 +1021,7 @@ class Trainer(Trainable):
             self.optimizer.restore(state["optimizer"])
 
     def _register_if_needed(self, env_object):
-        if isinstance(env_object, six.string_types):
+        if isinstance(env_object, str):
             return env_object
         elif isinstance(env_object, type):
             name = env_object.__name__
diff --git a/rllib/env/policy_server_input.py b/rllib/env/policy_server_input.py
index efb1ea996..f09fbf975 100644
--- a/rllib/env/policy_server_input.py
+++ b/rllib/env/policy_server_input.py
@@ -1,5 +1,5 @@
 import logging
-import six.moves.queue as queue
+import queue
 import threading
 import traceback
 
diff --git a/rllib/evaluation/sampler.py b/rllib/evaluation/sampler.py
index d39caacd5..7f413b0c5 100644
--- a/rllib/evaluation/sampler.py
+++ b/rllib/evaluation/sampler.py
@@ -1,7 +1,7 @@
 from collections import defaultdict, namedtuple
 import logging
 import numpy as np
-import six.moves.queue as queue
+import queue
 import threading
 import time
 
diff --git a/rllib/offline/json_reader.py b/rllib/offline/json_reader.py
index c1eefdd00..c2f0ac6ba 100644
--- a/rllib/offline/json_reader.py
+++ b/rllib/offline/json_reader.py
@@ -3,8 +3,7 @@ import json
 import logging
 import os
 import random
-import six
-from six.moves.urllib.parse import urlparse
+from urllib.parse import urlparse
 
 try:
     from smart_open import smart_open
@@ -39,7 +38,7 @@ class JsonReader(InputReader):
         """
 
         self.ioctx = ioctx or IOContext()
-        if isinstance(inputs, six.string_types):
+        if isinstance(inputs, str):
             inputs = os.path.abspath(os.path.expanduser(inputs))
             if os.path.isdir(inputs):
                 inputs = os.path.join(inputs, "*.json")
diff --git a/rllib/policy/sample_batch.py b/rllib/policy/sample_batch.py
index bb0c0ccc5..8c98ab350 100644
--- a/rllib/policy/sample_batch.py
+++ b/rllib/policy/sample_batch.py
@@ -1,4 +1,3 @@
-import six
 import collections
 import numpy as np
 
@@ -49,7 +48,7 @@ class SampleBatch:
         self.data = dict(*args, **kwargs)
         lengths = []
         for k, v in self.data.copy().items():
-            assert isinstance(k, six.string_types), self
+            assert isinstance(k, str), self
             lengths.append(len(v))
             self.data[k] = np.array(v, copy=False)
         if not lengths:
diff --git a/streaming/python/function.py b/streaming/python/function.py
index e94535d0a..fceb819a5 100644
--- a/streaming/python/function.py
+++ b/streaming/python/function.py
@@ -4,7 +4,7 @@ import sys
 from abc import ABC, abstractmethod
 import typing
 
-import cloudpickle
+from ray import cloudpickle
 from ray.streaming.runtime import gateway_client
 
 
diff --git a/streaming/python/partition.py b/streaming/python/partition.py
index a9f9d92d6..b6a9b87fe 100644
--- a/streaming/python/partition.py
+++ b/streaming/python/partition.py
@@ -1,7 +1,7 @@
 import importlib
 from abc import ABC, abstractmethod
 
-import cloudpickle
+from ray import cloudpickle
 from ray.streaming.runtime import gateway_client