diff --git a/sdks/python/apache_beam/coders/coders.py b/sdks/python/apache_beam/coders/coders.py
index 66f4bff16a85..1bf0fa8c5efd 100644
--- a/sdks/python/apache_beam/coders/coders.py
+++ b/sdks/python/apache_beam/coders/coders.py
@@ -24,7 +24,6 @@
 from __future__ import absolute_import
 
 import base64
-import sys
 from builtins import object
 from typing import TYPE_CHECKING
 from typing import Any
@@ -426,19 +425,8 @@ def to_type_hint(self):
 
 class ToBytesCoder(Coder):
   """A default string coder used if no sink coder is specified."""
-
-  if sys.version_info.major == 2:
-
-    def encode(self, value):
-      # pylint: disable=unicode-builtin
-      return (
-          value.encode('utf-8') if isinstance(value, unicode)  # noqa: F821
-          else str(value))
-
-  else:
-
-    def encode(self, value):
-      return value if isinstance(value, bytes) else str(value).encode('utf-8')
+  def encode(self, value):
+    return value if isinstance(value, bytes) else str(value).encode('utf-8')
 
   def decode(self, _):
     raise NotImplementedError('ToBytesCoder cannot be used for decoding.')
diff --git a/sdks/python/apache_beam/coders/coders_test_common.py b/sdks/python/apache_beam/coders/coders_test_common.py
index f4237d5c54e0..38e67fc6480a 100644
--- a/sdks/python/apache_beam/coders/coders_test_common.py
+++ b/sdks/python/apache_beam/coders/coders_test_common.py
@@ -22,7 +22,6 @@
 
 import logging
 import math
-import sys
 import unittest
 from builtins import range
 from typing import Any
@@ -97,9 +96,6 @@ class CodersTest(unittest.TestCase):
   def setUpClass(cls):
     cls.seen = set()
     cls.seen_nested = set()
-    # Method has been renamed in Python 3
-    if sys.version_info[0] < 3:
-      cls.assertCountEqual = cls.assertItemsEqual
 
   @classmethod
   def tearDownClass(cls):
diff --git a/sdks/python/apache_beam/coders/slow_stream.py b/sdks/python/apache_beam/coders/slow_stream.py
index a41c578de899..5840e241e1c4 100644
--- a/sdks/python/apache_beam/coders/slow_stream.py
+++ b/sdks/python/apache_beam/coders/slow_stream.py
@@ -24,7 +24,6 @@
 from __future__ import absolute_import
 
 import struct
-import sys
 from builtins import chr
 from builtins import object
 from typing import List
@@ -129,19 +128,6 @@ def __init__(self, data):
     self.data = data
     self.pos = 0
 
-    # The behavior of looping over a byte-string and obtaining byte characters
-    # has been changed between python 2 and 3.
-    # b = b'\xff\x01'
-    # Python 2:
-    # b[0] = '\xff'
-    # ord(b[0]) = 255
-    # Python 3:
-    # b[0] = 255
-    if sys.version_info[0] >= 3:
-      self.read_byte = self.read_byte_py3
-    else:
-      self.read_byte = self.read_byte_py2
-
   def size(self):
     return len(self.data) - self.pos
 
@@ -154,13 +140,7 @@ def read_all(self, nested):
     # type: (bool) -> bytes
     return self.read(self.read_var_int64() if nested else self.size())
 
-  def read_byte_py2(self):
-    # type: () -> int
-    self.pos += 1
-    # mypy tests against python 3.x, where this is an error:
-    return ord(self.data[self.pos - 1])  # type: ignore[arg-type]
-
-  def read_byte_py3(self):
+  def read_byte(self):
     # type: () -> int
     self.pos += 1
     return self.data[self.pos - 1]
diff --git a/sdks/python/apache_beam/dataframe/doctests_test.py b/sdks/python/apache_beam/dataframe/doctests_test.py
index 52d70fe57c4a..d3030fb1597d 100644
--- a/sdks/python/apache_beam/dataframe/doctests_test.py
+++ b/sdks/python/apache_beam/dataframe/doctests_test.py
@@ -18,7 +18,6 @@
 
 import doctest
 import os
-import sys
 import tempfile
 import unittest
 
@@ -146,7 +145,6 @@ def foo(x):
 '''
 
 
-@unittest.skipIf(sys.version_info <= (3, ), 'Requires contextlib.ExitStack.')
 class DoctestTest(unittest.TestCase):
   def test_good(self):
     result = doctests.teststring(SAMPLE_DOCTEST, report=False)
diff --git a/sdks/python/apache_beam/dataframe/frame_base.py b/sdks/python/apache_beam/dataframe/frame_base.py
index ff4ad14b924c..b5b7aface6c8 100644
--- a/sdks/python/apache_beam/dataframe/frame_base.py
+++ b/sdks/python/apache_beam/dataframe/frame_base.py
@@ -18,7 +18,6 @@
 
 import functools
 import inspect
-import sys
 from typing import Any
 from typing import Callable
 from typing import Dict
@@ -31,17 +30,8 @@
 from apache_beam.dataframe import expressions
 from apache_beam.dataframe import partitionings
 
-# pylint: disable=deprecated-method
-if sys.version_info < (3, ):
-  _getargspec = inspect.getargspec
-
-  def _unwrap(func):
-    while hasattr(func, '__wrapped__'):
-      func = func.__wrapped__
-    return func
-else:
-  _getargspec = inspect.getfullargspec
-  _unwrap = inspect.unwrap
+_getargspec = inspect.getfullargspec
+_unwrap = inspect.unwrap
 
 
 class DeferredBase(object):
diff --git a/sdks/python/apache_beam/dataframe/io_test.py b/sdks/python/apache_beam/dataframe/io_test.py
index 06fe8dbe78de..6ae16f2f03fe 100644
--- a/sdks/python/apache_beam/dataframe/io_test.py
+++ b/sdks/python/apache_beam/dataframe/io_test.py
@@ -23,7 +23,6 @@
 import os
 import platform
 import shutil
-import sys
 import tempfile
 import unittest
 from io import BytesIO
@@ -67,7 +66,6 @@ def read_all_lines(self, pattern):
         for line in fin:
           yield line.rstrip('\n')
 
-  @unittest.skipIf(sys.version_info[0] < 3, 'unicode issues')
   def test_read_write_csv(self):
     input = self.temp_dir({'1.csv': 'a,b\n1,2\n', '2.csv': 'a,b\n3,4\n'})
     output = self.temp_dir()
diff --git a/sdks/python/apache_beam/dataframe/pandas_doctests_test.py b/sdks/python/apache_beam/dataframe/pandas_doctests_test.py
index e4b5a34f0d6d..15cd988371c4 100644
--- a/sdks/python/apache_beam/dataframe/pandas_doctests_test.py
+++ b/sdks/python/apache_beam/dataframe/pandas_doctests_test.py
@@ -25,8 +25,6 @@
 from apache_beam.dataframe.pandas_top_level_functions import _is_top_level_function
 
 
-@unittest.skipIf(sys.version_info <= (3, ), 'Requires contextlib.ExitStack.')
-@unittest.skipIf(sys.version_info < (3, 6), 'Nondeterministic dict ordering.')
 @unittest.skipIf(sys.platform == 'win32', '[BEAM-10626]')
 class DoctestTest(unittest.TestCase):
   def test_dataframe_tests(self):
diff --git a/sdks/python/apache_beam/examples/snippets/snippets_test.py b/sdks/python/apache_beam/examples/snippets/snippets_test.py
index 766f1fdccb3d..d1a48e301070 100644
--- a/sdks/python/apache_beam/examples/snippets/snippets_test.py
+++ b/sdks/python/apache_beam/examples/snippets/snippets_test.py
@@ -28,7 +28,6 @@
 import logging
 import math
 import os
-import sys
 import tempfile
 import time
 import unittest
@@ -497,12 +496,6 @@ def expand(self, pcoll):
       return pcoll | 'DummyWriteForTesting' >> beam.ParDo(
           SnippetsTest.DummyWriteTransform.WriteDoFn(self.file_to_write))
 
-  @classmethod
-  def setUpClass(cls):
-    # Method has been renamed in Python 3
-    if sys.version_info[0] < 3:
-      cls.assertCountEqual = cls.assertItemsEqual
-
   def setUp(self):
     self.old_read_from_text = beam.io.ReadFromText
     self.old_write_to_text = beam.io.WriteToText
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py b/sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py
index 4f7ec2f2056a..3075388ad470 100644
--- a/sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py
+++ b/sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py
@@ -21,7 +21,6 @@
 from __future__ import absolute_import
 from __future__ import print_function
 
-import sys
 import unittest
 
 import mock
@@ -67,32 +66,16 @@ def check_percentages(actual):
     str)
 # pylint: enable=line-too-long
 class CombineGloballyTest(unittest.TestCase):
-  # TODO: Remove this after Python 2 deprecation.
-  # https://issues.apache.org/jira/browse/BEAM-8124
-  @unittest.skipIf(
-      sys.version_info[0] == 2, 'Python 2 renders sets in a non-compatible way')
   def test_combineglobally_function(self):
     combineglobally.combineglobally_function(check_common_items)
 
-  # TODO: Remove this after Python 2 deprecation.
-  # https://issues.apache.org/jira/browse/BEAM-8124
-  @unittest.skipIf(
-      sys.version_info[0] == 2, 'Python 2 renders sets in a non-compatible way')
   def test_combineglobally_lambda(self):
     combineglobally.combineglobally_lambda(check_common_items)
 
-  # TODO: Remove this after Python 2 deprecation.
-  # https://issues.apache.org/jira/browse/BEAM-8124
-  @unittest.skipIf(
-      sys.version_info[0] == 2, 'Python 2 renders sets in a non-compatible way')
   def test_combineglobally_multiple_arguments(self):
     combineglobally.combineglobally_multiple_arguments(
         check_common_items_with_exceptions)
 
-  # TODO: Remove this after Python 2 deprecation.
-  # https://issues.apache.org/jira/browse/BEAM-8124
-  @unittest.skipIf(
-      sys.version_info[0] == 2, 'Python 2 renders sets in a non-compatible way')
   def test_combineglobally_side_inputs_singleton(self):
     combineglobally.combineglobally_side_inputs_singleton(
         check_common_items_with_exceptions)
@@ -107,10 +90,6 @@ def test_combineglobally_side_inputs_singleton(self):
   #   combineglobally.combineglobally_side_inputs_dict(
   #       check_custom_common_items)
 
-  # TODO: Remove this after Python 2 deprecation.
-  # https://issues.apache.org/jira/browse/BEAM-8124
-  @unittest.skipIf(
-      sys.version_info[0] == 2, 'Python 2 renders sets in a non-compatible way')
   def test_combineglobally_combinefn(self):
     combineglobally.combineglobally_combinefn(check_percentages)
 
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/aggregation/groupby_test.py b/sdks/python/apache_beam/examples/snippets/transforms/aggregation/groupby_test.py
index 879446a69620..029d9d5b97ec 100644
--- a/sdks/python/apache_beam/examples/snippets/transforms/aggregation/groupby_test.py
+++ b/sdks/python/apache_beam/examples/snippets/transforms/aggregation/groupby_test.py
@@ -25,7 +25,6 @@
 from __future__ import division
 from __future__ import print_function
 
-import sys
 import typing
 import unittest
 
@@ -120,7 +119,6 @@ def test_groupby_two_exprs(self):
       ]
       assert_that(grouped | beam.MapTuple(normalize_kv), equal_to(expected))
 
-  @unittest.skipIf(sys.version_info[0] < 3, 'bad comparison op')
   def test_group_by_attr(self):
     # [START groupby_attr]
     with beam.Pipeline() as p:
@@ -145,7 +143,6 @@ def test_group_by_attr(self):
       ]
       assert_that(grouped | beam.MapTuple(normalize_kv), equal_to(expected))
 
-  @unittest.skipIf(sys.version_info[0] < 3, 'bad comparison op')
   def test_group_by_attr_expr(self):
     # [START groupby_attr_expr]
     with beam.Pipeline() as p:
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/aggregation/top_test.py b/sdks/python/apache_beam/examples/snippets/transforms/aggregation/top_test.py
index 2bbb759c0297..4781d819f3e8 100644
--- a/sdks/python/apache_beam/examples/snippets/transforms/aggregation/top_test.py
+++ b/sdks/python/apache_beam/examples/snippets/transforms/aggregation/top_test.py
@@ -19,7 +19,6 @@
 from __future__ import absolute_import
 from __future__ import print_function
 
-import sys
 import unittest
 
 import mock
@@ -98,12 +97,6 @@ def test_top_smallest_per_key(self):
   def test_top_of(self):
     top.top_of(check_shortest_elements)
 
-  # TODO: Remove this after Python 2 deprecation.
-  # https://issues.apache.org/jira/browse/BEAM-8124
-  @unittest.skipIf(
-      sys.version_info[0] == 2,
-      'nosetests in Python 2 uses ascii instead of utf-8 in '
-      'the Top.PerKey transform and causes this to fail')
   def test_top_per_key(self):
     top.top_per_key(check_shortest_elements_per_key)
 
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/elementwise/pardo_test.py b/sdks/python/apache_beam/examples/snippets/transforms/elementwise/pardo_test.py
index 8498ab9be8b2..9882cd481d73 100644
--- a/sdks/python/apache_beam/examples/snippets/transforms/elementwise/pardo_test.py
+++ b/sdks/python/apache_beam/examples/snippets/transforms/elementwise/pardo_test.py
@@ -21,9 +21,8 @@
 from __future__ import absolute_import
 from __future__ import print_function
 
-import platform
-import sys
 import unittest
+from io import StringIO
 
 import mock
 
@@ -34,13 +33,6 @@
 
 from . import pardo
 
-# TODO: Remove this after Python 2 deprecation.
-# https://issues.apache.org/jira/browse/BEAM-8124
-if sys.version_info[0] == 2:
-  from io import BytesIO as StringIO
-else:
-  from io import StringIO
-
 
 def check_plants(actual):
   expected = '''[START plants]
@@ -96,11 +88,6 @@ class ParDoTest(unittest.TestCase):
   def test_pardo_dofn(self):
     pardo.pardo_dofn(check_plants)
 
-  # TODO: Remove this after Python 2 deprecation.
-  # https://issues.apache.org/jira/browse/BEAM-8124
-  @unittest.skipIf(
-      sys.version_info[0] == 2 and platform.system() == 'Windows',
-      'Python 2 on Windows uses `long` rather than `int`')
   def test_pardo_dofn_params(self):
     pardo.pardo_dofn_params(check_dofn_params)
 
diff --git a/sdks/python/apache_beam/internal/pickler.py b/sdks/python/apache_beam/internal/pickler.py
index 395d511dd8e6..9f0c123daba3 100644
--- a/sdks/python/apache_beam/internal/pickler.py
+++ b/sdks/python/apache_beam/internal/pickler.py
@@ -55,13 +55,9 @@ def __exit__(self, *unused_exc_info):
     pass
 
 
-if sys.version_info[0] > 2:
-  # Pickling, especially unpickling, causes broken module imports on Python 3
-  # if executed concurrently, see: BEAM-8651, http://bugs.python.org/issue38884.
-  _pickle_lock_unless_py2 = threading.RLock()
-else:
-  # Avoid slow reentrant locks on Py2. See: https://bugs.python.org/issue3001.
-  _pickle_lock_unless_py2 = _NoOpContextManager()
+# Pickling, especially unpickling, causes broken module imports on Python 3
+# if executed concurrently, see: BEAM-8651, http://bugs.python.org/issue38884.
+_pickle_lock_unless_py2 = threading.RLock()
 # Dill 0.28.0 renamed dill.dill to dill._dill:
 # https://github.com/uqfoundation/dill/commit/f0972ecc7a41d0b8acada6042d557068cac69baa
 # TODO: Remove this once Beam depends on dill >= 0.2.8