diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobInvoker.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobInvoker.java
index 24edb28..b6e040c 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobInvoker.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobInvoker.java
@@ -70,9 +70,6 @@
     }
 
     PortablePipelineOptions portableOptions = flinkOptions.as(PortablePipelineOptions.class);
-    if (portableOptions.getSdkWorkerParallelism() == 0L) {
-      portableOptions.setSdkWorkerParallelism(serverConfig.getSdkWorkerParallelism());
-    }
 
     PortablePipelineRunner pipelineRunner;
     if (portableOptions.getOutputExecutablePath() == null
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkJobServerDriverTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkJobServerDriverTest.java
index a90f7e6..1e345d0 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkJobServerDriverTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkJobServerDriverTest.java
@@ -43,7 +43,6 @@
     assertThat(config.getArtifactPort(), is(8098));
     assertThat(config.getExpansionPort(), is(8097));
     assertThat(config.getFlinkMasterUrl(), is("[auto]"));
-    assertThat(config.getSdkWorkerParallelism(), is(1L));
     assertThat(config.isCleanArtifactsPerJob(), is(true));
     FlinkJobServerDriver flinkJobServerDriver = FlinkJobServerDriver.fromConfig(config);
     assertThat(flinkJobServerDriver, is(not(nullValue())));
@@ -62,7 +61,6 @@
               "--expansion-port",
               "44",
               "--flink-master-url=jobmanager",
-              "--sdk-worker-parallelism=4",
               "--clean-artifacts-per-job=false",
             });
     FlinkJobServerDriver.FlinkServerConfiguration config =
@@ -72,7 +70,6 @@
     assertThat(config.getArtifactPort(), is(43));
     assertThat(config.getExpansionPort(), is(44));
     assertThat(config.getFlinkMasterUrl(), is("jobmanager"));
-    assertThat(config.getSdkWorkerParallelism(), is(4L));
     assertThat(config.isCleanArtifactsPerJob(), is(false));
   }
 
diff --git a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/JobServerDriver.java b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/JobServerDriver.java
index f8977ff..0c5bf94 100644
--- a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/JobServerDriver.java
+++ b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/JobServerDriver.java
@@ -94,15 +94,6 @@
         handler = ExplicitBooleanOptionHandler.class)
     private boolean cleanArtifactsPerJob = true;
 
-    @Option(
-        name = "--sdk-worker-parallelism",
-        usage =
-            "Default parallelism for SDK worker processes. This option is only applied when the "
-                + "pipeline option sdkWorkerParallelism is set to 0."
-                + "Default is 1, If 0, worker parallelism will be dynamically decided by runner."
-                + "See also: sdkWorkerParallelism Pipeline Option")
-    private long sdkWorkerParallelism = 1L;
-
     public String getHost() {
       return host;
     }
@@ -126,10 +117,6 @@
     public boolean isCleanArtifactsPerJob() {
       return cleanArtifactsPerJob;
     }
-
-    public long getSdkWorkerParallelism() {
-      return this.sdkWorkerParallelism;
-    }
   }
 
   protected static ServerFactory createJobServerFactory(ServerConfiguration configuration) {
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/test/TestTableProvider.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/test/TestTableProvider.java
index b9d7bd7..e65d903 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/test/TestTableProvider.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/test/TestTableProvider.java
@@ -32,12 +32,15 @@
 import org.apache.beam.sdk.extensions.sql.impl.BeamTableStatistics;
 import org.apache.beam.sdk.extensions.sql.meta.BaseBeamTable;
 import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTableFilter;
+import org.apache.beam.sdk.extensions.sql.meta.DefaultTableFilter;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
 import org.apache.beam.sdk.extensions.sql.meta.provider.InMemoryMetaTableProvider;
 import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.SchemaCoder;
+import org.apache.beam.sdk.schemas.transforms.Select;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.ParDo;
@@ -151,12 +154,37 @@
     }
 
     @Override
+    public PCollection<Row> buildIOReader(
+        PBegin begin, BeamSqlTableFilter filters, List<String> fieldNames) {
+      PCollection<Row> withAllFields = buildIOReader(begin);
+      if (fieldNames.isEmpty() && filters instanceof DefaultTableFilter) {
+        return withAllFields;
+      }
+
+      PCollection<Row> result = withAllFields;
+      if (!(filters instanceof DefaultTableFilter)) {
+        throw new RuntimeException("Unimplemented at the moment.");
+      }
+
+      if (!fieldNames.isEmpty()) {
+        result = result.apply(Select.fieldNames(fieldNames.toArray(new String[0])));
+      }
+
+      return result;
+    }
+
+    @Override
     public POutput buildIOWriter(PCollection<Row> input) {
       input.apply(ParDo.of(new CollectorFn(tableWithRows)));
       return PDone.in(input.getPipeline());
     }
 
     @Override
+    public boolean supportsProjects() {
+      return true;
+    }
+
+    @Override
     public Schema getSchema() {
       return tableWithRows.table.getSchema();
     }
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/test/TestTableProviderTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/test/TestTableProviderTest.java
new file mode 100644
index 0000000..c65d593
--- /dev/null
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/test/TestTableProviderTest.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql.meta.provider.test;
+
+import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.calcite.v1_20_0.com.google.common.collect.ImmutableList;
+import org.joda.time.Duration;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class TestTableProviderTest {
+  private static final Schema BASIC_SCHEMA =
+      Schema.builder().addInt32Field("id").addStringField("name").build();
+  private BeamSqlTable beamSqlTable;
+
+  @Rule public TestPipeline pipeline = TestPipeline.create();
+
+  @Before
+  public void buildUp() {
+    TestTableProvider tableProvider = new TestTableProvider();
+    Table table = getTable("tableName");
+    tableProvider.createTable(table);
+    tableProvider.addRows(
+        table.getName(), row(BASIC_SCHEMA, 1, "one"), row(BASIC_SCHEMA, 2, "two"));
+
+    beamSqlTable = tableProvider.buildBeamSqlTable(table);
+  }
+
+  @Test
+  public void testInMemoryTableProvider_returnsSelectedColumns() {
+    PCollection<Row> result =
+        beamSqlTable.buildIOReader(
+            pipeline.begin(),
+            beamSqlTable.constructFilter(ImmutableList.of()),
+            ImmutableList.of("name"));
+
+    PAssert.that(result)
+        .containsInAnyOrder(row(result.getSchema(), "one"), row(result.getSchema(), "two"));
+
+    pipeline.run().waitUntilFinish(Duration.standardMinutes(2));
+  }
+
+  @Test
+  public void testInMemoryTableProvider_withEmptySelectedColumns_returnsAllColumns() {
+    PCollection<Row> result =
+        beamSqlTable.buildIOReader(
+            pipeline.begin(), beamSqlTable.constructFilter(ImmutableList.of()), ImmutableList.of());
+
+    PAssert.that(result)
+        .containsInAnyOrder(row(result.getSchema(), 1, "one"), row(result.getSchema(), 2, "two"));
+
+    pipeline.run().waitUntilFinish(Duration.standardMinutes(2));
+  }
+
+  @Test
+  public void testInMemoryTableProvider_withAllSelectedColumns_returnsAllColumns() {
+    PCollection<Row> result =
+        beamSqlTable.buildIOReader(
+            pipeline.begin(),
+            beamSqlTable.constructFilter(ImmutableList.of()),
+            ImmutableList.of("name", "id")); // Note that order is ignored
+
+    // Selected columns are outputted in the same order they are listed in the schema.
+    PAssert.that(result)
+        .containsInAnyOrder(row(result.getSchema(), 1, "one"), row(result.getSchema(), 2, "two"));
+
+    pipeline.run().waitUntilFinish(Duration.standardMinutes(2));
+  }
+
+  @Test
+  public void testInMemoryTableProvider_withDuplicateSelectedColumns_returnsSelectedColumnsOnce() {
+    PCollection<Row> result =
+        beamSqlTable.buildIOReader(
+            pipeline.begin(),
+            beamSqlTable.constructFilter(ImmutableList.of()),
+            ImmutableList.of("name", "name"));
+
+    PAssert.that(result)
+        .containsInAnyOrder(row(result.getSchema(), "one"), row(result.getSchema(), "two"));
+
+    pipeline.run().waitUntilFinish(Duration.standardMinutes(2));
+  }
+
+  private static Row row(Schema schema, Object... objects) {
+    return Row.withSchema(schema).addValues(objects).build();
+  }
+
+  private static Table getTable(String name) {
+    return Table.builder()
+        .name(name)
+        .comment(name + " table")
+        .schema(BASIC_SCHEMA)
+        .type("test")
+        .build();
+  }
+}
diff --git a/sdks/java/io/amazon-web-services/src/main/java/org/apache/beam/sdk/io/aws/options/AwsModule.java b/sdks/java/io/amazon-web-services/src/main/java/org/apache/beam/sdk/io/aws/options/AwsModule.java
index 79f9db0..30add9b 100644
--- a/sdks/java/io/amazon-web-services/src/main/java/org/apache/beam/sdk/io/aws/options/AwsModule.java
+++ b/sdks/java/io/amazon-web-services/src/main/java/org/apache/beam/sdk/io/aws/options/AwsModule.java
@@ -36,7 +36,9 @@
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.core.type.WritableTypeId;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JsonDeserializer;
 import com.fasterxml.jackson.databind.JsonSerializer;
@@ -156,8 +158,9 @@
         SerializerProvider serializers,
         TypeSerializer typeSerializer)
         throws IOException {
-      typeSerializer.writeTypePrefixForObject(credentialsProvider, jsonGenerator);
-
+      WritableTypeId typeId =
+          typeSerializer.writeTypePrefix(
+              jsonGenerator, typeSerializer.typeId(credentialsProvider, JsonToken.START_OBJECT));
       if (credentialsProvider.getClass().equals(AWSStaticCredentialsProvider.class)) {
         jsonGenerator.writeStringField(
             AWS_ACCESS_KEY_ID, credentialsProvider.getCredentials().getAWSAccessKeyId());
@@ -197,7 +200,7 @@
         throw new IllegalArgumentException(
             "Unsupported AWS credentials provider type " + credentialsProvider.getClass());
       }
-      typeSerializer.writeTypeSuffixForObject(credentialsProvider, jsonGenerator);
+      typeSerializer.writeTypeSuffix(jsonGenerator, typeId);
     }
   }
 
diff --git a/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/options/AwsModule.java b/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/options/AwsModule.java
index 4adad67..79e8d15 100644
--- a/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/options/AwsModule.java
+++ b/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/options/AwsModule.java
@@ -20,7 +20,9 @@
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.core.type.WritableTypeId;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JsonDeserializer;
 import com.fasterxml.jackson.databind.JsonSerializer;
@@ -139,7 +141,9 @@
         SerializerProvider serializer,
         TypeSerializer typeSerializer)
         throws IOException {
-      typeSerializer.writeTypePrefixForObject(credentialsProvider, jsonGenerator);
+      WritableTypeId typeId =
+          typeSerializer.writeTypePrefix(
+              jsonGenerator, typeSerializer.typeId(credentialsProvider, JsonToken.START_OBJECT));
       if (credentialsProvider.getClass().equals(StaticCredentialsProvider.class)) {
         jsonGenerator.writeStringField(
             ACCESS_KEY_ID, credentialsProvider.resolveCredentials().accessKeyId());
@@ -149,7 +153,7 @@
         throw new IllegalArgumentException(
             "Unsupported AWS credentials provider type " + credentialsProvider.getClass());
       }
-      typeSerializer.writeTypeSuffixForObject(credentialsProvider, jsonGenerator);
+      typeSerializer.writeTypeSuffix(jsonGenerator, typeId);
     }
   }
 
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/__init__.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/__init__.py
deleted file mode 100644
index 6569e3f..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/__init__.py
+++ /dev/null
@@ -1,18 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/filter.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/filter.py
deleted file mode 100644
index 44b11b8..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/filter.py
+++ /dev/null
@@ -1,182 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-
-def filter_function(test=None):
-  # [START filter_function]
-  import apache_beam as beam
-
-  def is_perennial(plant):
-    return plant['duration'] == 'perennial'
-
-  with beam.Pipeline() as pipeline:
-    perennials = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Filter perennials' >> beam.Filter(is_perennial)
-        | beam.Map(print)
-    )
-    # [END filter_function]
-    if test:
-      test(perennials)
-
-
-def filter_lambda(test=None):
-  # [START filter_lambda]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    perennials = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Filter perennials' >> beam.Filter(
-            lambda plant: plant['duration'] == 'perennial')
-        | beam.Map(print)
-    )
-    # [END filter_lambda]
-    if test:
-      test(perennials)
-
-
-def filter_multiple_arguments(test=None):
-  # [START filter_multiple_arguments]
-  import apache_beam as beam
-
-  def has_duration(plant, duration):
-    return plant['duration'] == duration
-
-  with beam.Pipeline() as pipeline:
-    perennials = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Filter perennials' >> beam.Filter(has_duration, 'perennial')
-        | beam.Map(print)
-    )
-    # [END filter_multiple_arguments]
-    if test:
-      test(perennials)
-
-
-def filter_side_inputs_singleton(test=None):
-  # [START filter_side_inputs_singleton]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    perennial = pipeline | 'Perennial' >> beam.Create(['perennial'])
-
-    perennials = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Filter perennials' >> beam.Filter(
-            lambda plant, duration: plant['duration'] == duration,
-            duration=beam.pvalue.AsSingleton(perennial),
-        )
-        | beam.Map(print)
-    )
-    # [END filter_side_inputs_singleton]
-    if test:
-      test(perennials)
-
-
-def filter_side_inputs_iter(test=None):
-  # [START filter_side_inputs_iter]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    valid_durations = pipeline | 'Valid durations' >> beam.Create([
-        'annual',
-        'biennial',
-        'perennial',
-    ])
-
-    valid_plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'PERENNIAL'},
-        ])
-        | 'Filter valid plants' >> beam.Filter(
-            lambda plant, valid_durations: plant['duration'] in valid_durations,
-            valid_durations=beam.pvalue.AsIter(valid_durations),
-        )
-        | beam.Map(print)
-    )
-    # [END filter_side_inputs_iter]
-    if test:
-      test(valid_plants)
-
-
-def filter_side_inputs_dict(test=None):
-  # [START filter_side_inputs_dict]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    keep_duration = pipeline | 'Duration filters' >> beam.Create([
-        ('annual', False),
-        ('biennial', False),
-        ('perennial', True),
-    ])
-
-    perennials = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Filter plants by duration' >> beam.Filter(
-            lambda plant, keep_duration: keep_duration[plant['duration']],
-            keep_duration=beam.pvalue.AsDict(keep_duration),
-        )
-        | beam.Map(print)
-    )
-    # [END filter_side_inputs_dict]
-    if test:
-      test(perennials)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/filter_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/filter_test.py
deleted file mode 100644
index 02da146..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/filter_test.py
+++ /dev/null
@@ -1,80 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-import unittest
-
-import mock
-
-from apache_beam.examples.snippets.transforms.element_wise.filter import *
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.filter.print', lambda elem: elem)
-# pylint: enable=line-too-long
-class FilterTest(unittest.TestCase):
-  def __init__(self, methodName):
-    super(FilterTest, self).__init__(methodName)
-    # [START perennials]
-    perennials = [
-        {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-        {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-        {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-    ]
-    # [END perennials]
-    self.perennials_test = lambda actual: \
-        assert_that(actual, equal_to(perennials))
-
-    # [START valid_plants]
-    valid_plants = [
-        {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-        {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-        {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-        {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-    ]
-    # [END valid_plants]
-    self.valid_plants_test = lambda actual: \
-        assert_that(actual, equal_to(valid_plants))
-
-  def test_filter_function(self):
-    filter_function(self.perennials_test)
-
-  def test_filter_lambda(self):
-    filter_lambda(self.perennials_test)
-
-  def test_filter_multiple_arguments(self):
-    filter_multiple_arguments(self.perennials_test)
-
-  def test_filter_side_inputs_singleton(self):
-    filter_side_inputs_singleton(self.perennials_test)
-
-  def test_filter_side_inputs_iter(self):
-    filter_side_inputs_iter(self.valid_plants_test)
-
-  def test_filter_side_inputs_dict(self):
-    filter_side_inputs_dict(self.perennials_test)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/keys.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/keys.py
deleted file mode 100644
index 01c9d6b..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/keys.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-
-def keys(test=None):
-  # [START keys]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    icons = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            ('🍓', 'Strawberry'),
-            ('🥕', 'Carrot'),
-            ('🍆', 'Eggplant'),
-            ('🍅', 'Tomato'),
-            ('🥔', 'Potato'),
-        ])
-        | 'Keys' >> beam.Keys()
-        | beam.Map(print)
-    )
-    # [END keys]
-    if test:
-      test(icons)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/keys_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/keys_test.py
deleted file mode 100644
index 9cb4909..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/keys_test.py
+++ /dev/null
@@ -1,55 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-import unittest
-
-import mock
-
-from apache_beam.examples.snippets.transforms.element_wise.keys import *
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.keys.print', lambda elem: elem)
-# pylint: enable=line-too-long
-class KeysTest(unittest.TestCase):
-  def __init__(self, methodName):
-    super(KeysTest, self).__init__(methodName)
-    # [START icons]
-    icons = [
-        '🍓',
-        '🥕',
-        '🍆',
-        '🍅',
-        '🥔',
-    ]
-    # [END icons]
-    self.icons_test = lambda actual: assert_that(actual, equal_to(icons))
-
-  def test_keys(self):
-    keys(self.icons_test)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/kvswap.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/kvswap.py
deleted file mode 100644
index 2107fd5..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/kvswap.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-
-def kvswap(test=None):
-  # [START kvswap]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            ('🍓', 'Strawberry'),
-            ('🥕', 'Carrot'),
-            ('🍆', 'Eggplant'),
-            ('🍅', 'Tomato'),
-            ('🥔', 'Potato'),
-        ])
-        | 'Key-Value swap' >> beam.KvSwap()
-        | beam.Map(print)
-    )
-    # [END kvswap]
-    if test:
-      test(plants)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/kvswap_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/kvswap_test.py
deleted file mode 100644
index 85fa9dc..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/kvswap_test.py
+++ /dev/null
@@ -1,55 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-import unittest
-
-import mock
-
-from apache_beam.examples.snippets.transforms.element_wise.kvswap import *
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.kvswap.print', lambda elem: elem)
-# pylint: enable=line-too-long
-class KvSwapTest(unittest.TestCase):
-  def __init__(self, methodName):
-    super(KvSwapTest, self).__init__(methodName)
-    # [START plants]
-    plants = [
-        ('Strawberry', '🍓'),
-        ('Carrot', '🥕'),
-        ('Eggplant', '🍆'),
-        ('Tomato', '🍅'),
-        ('Potato', '🥔'),
-    ]
-    # [END plants]
-    self.plants_test = lambda actual: assert_that(actual, equal_to(plants))
-
-  def test_kvswap(self):
-    kvswap(self.plants_test)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/map.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/map.py
deleted file mode 100644
index 9defd47..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/map.py
+++ /dev/null
@@ -1,226 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-
-def map_simple(test=None):
-  # [START map_simple]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            '   🍓Strawberry   \n',
-            '   🥕Carrot   \n',
-            '   🍆Eggplant   \n',
-            '   🍅Tomato   \n',
-            '   🥔Potato   \n',
-        ])
-        | 'Strip' >> beam.Map(str.strip)
-        | beam.Map(print)
-    )
-    # [END map_simple]
-    if test:
-      test(plants)
-
-
-def map_function(test=None):
-  # [START map_function]
-  import apache_beam as beam
-
-  def strip_header_and_newline(text):
-    return text.strip('# \n')
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            '# 🍓Strawberry\n',
-            '# 🥕Carrot\n',
-            '# 🍆Eggplant\n',
-            '# 🍅Tomato\n',
-            '# 🥔Potato\n',
-        ])
-        | 'Strip header' >> beam.Map(strip_header_and_newline)
-        | beam.Map(print)
-    )
-    # [END map_function]
-    if test:
-      test(plants)
-
-
-def map_lambda(test=None):
-  # [START map_lambda]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            '# 🍓Strawberry\n',
-            '# 🥕Carrot\n',
-            '# 🍆Eggplant\n',
-            '# 🍅Tomato\n',
-            '# 🥔Potato\n',
-        ])
-        | 'Strip header' >> beam.Map(lambda text: text.strip('# \n'))
-        | beam.Map(print)
-    )
-    # [END map_lambda]
-    if test:
-      test(plants)
-
-
-def map_multiple_arguments(test=None):
-  # [START map_multiple_arguments]
-  import apache_beam as beam
-
-  def strip(text, chars=None):
-    return text.strip(chars)
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            '# 🍓Strawberry\n',
-            '# 🥕Carrot\n',
-            '# 🍆Eggplant\n',
-            '# 🍅Tomato\n',
-            '# 🥔Potato\n',
-        ])
-        | 'Strip header' >> beam.Map(strip, chars='# \n')
-        | beam.Map(print)
-    )
-    # [END map_multiple_arguments]
-    if test:
-      test(plants)
-
-
-def map_tuple(test=None):
-  # [START map_tuple]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            ('🍓', 'Strawberry'),
-            ('🥕', 'Carrot'),
-            ('🍆', 'Eggplant'),
-            ('🍅', 'Tomato'),
-            ('🥔', 'Potato'),
-        ])
-        | 'Format' >> beam.MapTuple(
-            lambda icon, plant: '{}{}'.format(icon, plant))
-        | beam.Map(print)
-    )
-    # [END map_tuple]
-    if test:
-      test(plants)
-
-
-def map_side_inputs_singleton(test=None):
-  # [START map_side_inputs_singleton]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    chars = pipeline | 'Create chars' >> beam.Create(['# \n'])
-
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            '# 🍓Strawberry\n',
-            '# 🥕Carrot\n',
-            '# 🍆Eggplant\n',
-            '# 🍅Tomato\n',
-            '# 🥔Potato\n',
-        ])
-        | 'Strip header' >> beam.Map(
-            lambda text, chars: text.strip(chars),
-            chars=beam.pvalue.AsSingleton(chars),
-        )
-        | beam.Map(print)
-    )
-    # [END map_side_inputs_singleton]
-    if test:
-      test(plants)
-
-
-def map_side_inputs_iter(test=None):
-  # [START map_side_inputs_iter]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    chars = pipeline | 'Create chars' >> beam.Create(['#', ' ', '\n'])
-
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            '# 🍓Strawberry\n',
-            '# 🥕Carrot\n',
-            '# 🍆Eggplant\n',
-            '# 🍅Tomato\n',
-            '# 🥔Potato\n',
-        ])
-        | 'Strip header' >> beam.Map(
-            lambda text, chars: text.strip(''.join(chars)),
-            chars=beam.pvalue.AsIter(chars),
-        )
-        | beam.Map(print)
-    )
-    # [END map_side_inputs_iter]
-    if test:
-      test(plants)
-
-
-def map_side_inputs_dict(test=None):
-  # [START map_side_inputs_dict]
-  import apache_beam as beam
-
-  def replace_duration(plant, durations):
-    plant['duration'] = durations[plant['duration']]
-    return plant
-
-  with beam.Pipeline() as pipeline:
-    durations = pipeline | 'Durations' >> beam.Create([
-        (0, 'annual'),
-        (1, 'biennial'),
-        (2, 'perennial'),
-    ])
-
-    plant_details = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 2},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 1},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 2},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 0},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 2},
-        ])
-        | 'Replace duration' >> beam.Map(
-            replace_duration,
-            durations=beam.pvalue.AsDict(durations),
-        )
-        | beam.Map(print)
-    )
-    # [END map_side_inputs_dict]
-    if test:
-      test(plant_details)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/map_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/map_test.py
deleted file mode 100644
index 5fcee8a..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/map_test.py
+++ /dev/null
@@ -1,90 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-import unittest
-
-import mock
-
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-from . import map
-
-
-def check_plants(actual):
-  # [START plants]
-  plants = [
-      '🍓Strawberry',
-      '🥕Carrot',
-      '🍆Eggplant',
-      '🍅Tomato',
-      '🥔Potato',
-  ]
-  # [END plants]
-  assert_that(actual, equal_to(plants))
-
-
-def check_plant_details(actual):
-  # [START plant_details]
-  plant_details = [
-      {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-      {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-      {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-      {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-      {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-  ]
-  # [END plant_details]
-  assert_that(actual, equal_to(plant_details))
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.map.print', lambda elem: elem)
-# pylint: enable=line-too-long
-class MapTest(unittest.TestCase):
-  def test_map_simple(self):
-    map.map_simple(check_plants)
-
-  def test_map_function(self):
-    map.map_function(check_plants)
-
-  def test_map_lambda(self):
-    map.map_lambda(check_plants)
-
-  def test_map_multiple_arguments(self):
-    map.map_multiple_arguments(check_plants)
-
-  def test_map_tuple(self):
-    map.map_tuple(check_plants)
-
-  def test_map_side_inputs_singleton(self):
-    map.map_side_inputs_singleton(check_plants)
-
-  def test_map_side_inputs_iter(self):
-    map.map_side_inputs_iter(check_plants)
-
-  def test_map_side_inputs_dict(self):
-    map.map_side_inputs_dict(check_plant_details)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/pardo.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/pardo.py
deleted file mode 100644
index 971e9f0..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/pardo.py
+++ /dev/null
@@ -1,126 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-from __future__ import unicode_literals
-
-
-def pardo_dofn(test=None):
-  # [START pardo_dofn]
-  import apache_beam as beam
-
-  class SplitWords(beam.DoFn):
-    def __init__(self, delimiter=','):
-      self.delimiter = delimiter
-
-    def process(self, text):
-      for word in text.split(self.delimiter):
-        yield word
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            '🍓Strawberry,🥕Carrot,🍆Eggplant',
-            '🍅Tomato,🥔Potato',
-        ])
-        | 'Split words' >> beam.ParDo(SplitWords(','))
-        | beam.Map(print)
-    )
-    # [END pardo_dofn]
-    if test:
-      test(plants)
-
-
-def pardo_dofn_params(test=None):
-  # pylint: disable=line-too-long
-  # [START pardo_dofn_params]
-  import apache_beam as beam
-
-  class AnalyzeElement(beam.DoFn):
-    def process(self, elem, timestamp=beam.DoFn.TimestampParam, window=beam.DoFn.WindowParam):
-      yield '\n'.join([
-          '# timestamp',
-          'type(timestamp) -> ' + repr(type(timestamp)),
-          'timestamp.micros -> ' + repr(timestamp.micros),
-          'timestamp.to_rfc3339() -> ' + repr(timestamp.to_rfc3339()),
-          'timestamp.to_utc_datetime() -> ' + repr(timestamp.to_utc_datetime()),
-          '',
-          '# window',
-          'type(window) -> ' + repr(type(window)),
-          'window.start -> {} ({})'.format(window.start, window.start.to_utc_datetime()),
-          'window.end -> {} ({})'.format(window.end, window.end.to_utc_datetime()),
-          'window.max_timestamp() -> {} ({})'.format(window.max_timestamp(), window.max_timestamp().to_utc_datetime()),
-      ])
-
-  with beam.Pipeline() as pipeline:
-    dofn_params = (
-        pipeline
-        | 'Create a single test element' >> beam.Create([':)'])
-        | 'Add timestamp (Spring equinox 2020)' >> beam.Map(
-            lambda elem: beam.window.TimestampedValue(elem, 1584675660))
-        | 'Fixed 30sec windows' >> beam.WindowInto(beam.window.FixedWindows(30))
-        | 'Analyze element' >> beam.ParDo(AnalyzeElement())
-        | beam.Map(print)
-    )
-    # [END pardo_dofn_params]
-    # pylint: enable=line-too-long
-    if test:
-      test(dofn_params)
-
-
-def pardo_dofn_methods(test=None):
-  # [START pardo_dofn_methods]
-  import apache_beam as beam
-
-  class DoFnMethods(beam.DoFn):
-    def __init__(self):
-      print('__init__')
-      self.window = beam.window.GlobalWindow()
-
-    def setup(self):
-      print('setup')
-
-    def start_bundle(self):
-      print('start_bundle')
-
-    def process(self, element, window=beam.DoFn.WindowParam):
-      self.window = window
-      yield '* process: ' + element
-
-    def finish_bundle(self):
-      yield beam.utils.windowed_value.WindowedValue(
-          value='* finish_bundle: 🌱🌳🌍',
-          timestamp=0,
-          windows=[self.window],
-      )
-
-    def teardown(self):
-      print('teardown')
-
-  with beam.Pipeline() as pipeline:
-    results = (
-        pipeline
-        | 'Create inputs' >> beam.Create(['🍓', '🥕', '🍆', '🍅', '🥔'])
-        | 'DoFn methods' >> beam.ParDo(DoFnMethods())
-        | beam.Map(print)
-    )
-    # [END pardo_dofn_methods]
-    if test:
-      return test(results)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/pardo_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/pardo_test.py
deleted file mode 100644
index a8de2d0..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/pardo_test.py
+++ /dev/null
@@ -1,120 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-from __future__ import unicode_literals
-
-import io
-import platform
-import sys
-import unittest
-
-import mock
-
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-from . import pardo
-
-
-def check_plants(actual):
-  # [START plants]
-  plants = [
-      '🍓Strawberry',
-      '🥕Carrot',
-      '🍆Eggplant',
-      '🍅Tomato',
-      '🥔Potato',
-  ]
-  # [END plants]
-  assert_that(actual, equal_to(plants))
-
-
-def check_dofn_params(actual):
-  # pylint: disable=line-too-long
-  dofn_params = '\n'.join('''[START dofn_params]
-# timestamp
-type(timestamp) -> <class 'apache_beam.utils.timestamp.Timestamp'>
-timestamp.micros -> 1584675660000000
-timestamp.to_rfc3339() -> '2020-03-20T03:41:00Z'
-timestamp.to_utc_datetime() -> datetime.datetime(2020, 3, 20, 3, 41)
-
-# window
-type(window) -> <class 'apache_beam.transforms.window.IntervalWindow'>
-window.start -> Timestamp(1584675660) (2020-03-20 03:41:00)
-window.end -> Timestamp(1584675690) (2020-03-20 03:41:30)
-window.max_timestamp() -> Timestamp(1584675689.999999) (2020-03-20 03:41:29.999999)
-[END dofn_params]'''.splitlines()[1:-1])
-  # pylint: enable=line-too-long
-  assert_that(actual, equal_to([dofn_params]))
-
-
-def check_dofn_methods(actual):
-  # Return the expected stdout to check the ordering of the called methods.
-  return '''[START results]
-__init__
-setup
-start_bundle
-* process: 🍓
-* process: 🥕
-* process: 🍆
-* process: 🍅
-* process: 🥔
-* finish_bundle: 🌱🌳🌍
-teardown
-[END results]'''.splitlines()[1:-1]
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.pardo.print', lambda elem: elem)
-# pylint: enable=line-too-long
-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] < 3 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)
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-@mock.patch('sys.stdout', new_callable=io.StringIO)
-class ParDoStdoutTest(unittest.TestCase):
-  def test_pardo_dofn_methods(self, mock_stdout):
-    expected = pardo.pardo_dofn_methods(check_dofn_methods)
-    actual = mock_stdout.getvalue().splitlines()
-
-    # For the stdout, check the ordering of the methods, not of the elements.
-    actual_stdout = [line.split(':')[0] for line in actual]
-    expected_stdout = [line.split(':')[0] for line in expected]
-    self.assertEqual(actual_stdout, expected_stdout)
-
-    # For the elements, ignore the stdout and just make sure all elements match.
-    actual_elements = {line for line in actual if line.startswith('*')}
-    expected_elements = {line for line in expected if line.startswith('*')}
-    self.assertEqual(actual_elements, expected_elements)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/partition.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/partition.py
deleted file mode 100644
index 6f839d4..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/partition.py
+++ /dev/null
@@ -1,136 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-
-def partition_function(test=None):
-  # [START partition_function]
-  import apache_beam as beam
-
-  durations = ['annual', 'biennial', 'perennial']
-
-  def by_duration(plant, num_partitions):
-    return durations.index(plant['duration'])
-
-  with beam.Pipeline() as pipeline:
-    annuals, biennials, perennials = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Partition' >> beam.Partition(by_duration, len(durations))
-    )
-    _ = (
-        annuals
-        | 'Annuals' >> beam.Map(lambda x: print('annual: ' + str(x)))
-    )
-    _ = (
-        biennials
-        | 'Biennials' >> beam.Map(lambda x: print('biennial: ' + str(x)))
-    )
-    _ = (
-        perennials
-        | 'Perennials' >> beam.Map(lambda x: print('perennial: ' + str(x)))
-    )
-    # [END partition_function]
-    if test:
-      test(annuals, biennials, perennials)
-
-
-def partition_lambda(test=None):
-  # [START partition_lambda]
-  import apache_beam as beam
-
-  durations = ['annual', 'biennial', 'perennial']
-
-  with beam.Pipeline() as pipeline:
-    annuals, biennials, perennials = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Partition' >> beam.Partition(
-            lambda plant, num_partitions: durations.index(plant['duration']),
-            len(durations),
-        )
-    )
-    _ = (
-        annuals
-        | 'Annuals' >> beam.Map(lambda x: print('annual: ' + str(x)))
-    )
-    _ = (
-        biennials
-        | 'Biennials' >> beam.Map(lambda x: print('biennial: ' + str(x)))
-    )
-    _ = (
-        perennials
-        | 'Perennials' >> beam.Map(lambda x: print('perennial: ' + str(x)))
-    )
-    # [END partition_lambda]
-    if test:
-      test(annuals, biennials, perennials)
-
-
-def partition_multiple_arguments(test=None):
-  # [START partition_multiple_arguments]
-  import apache_beam as beam
-  import json
-
-  def split_dataset(plant, num_partitions, ratio):
-    assert num_partitions == len(ratio)
-    bucket = sum(map(ord, json.dumps(plant))) % sum(ratio)
-    total = 0
-    for i, part in enumerate(ratio):
-      total += part
-      if bucket < total:
-        return i
-    return len(ratio) - 1
-
-  with beam.Pipeline() as pipeline:
-    train_dataset, test_dataset = (
-        pipeline
-        | 'Gardening plants' >> beam.Create([
-            {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-            {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-            {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-            {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-            {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-        ])
-        | 'Partition' >> beam.Partition(split_dataset, 2, ratio=[8, 2])
-    )
-    _ = (
-        train_dataset
-        | 'Train' >> beam.Map(lambda x: print('train: ' + str(x)))
-    )
-    _ = (
-        test_dataset
-        | 'Test'  >> beam.Map(lambda x: print('test: ' + str(x)))
-    )
-    # [END partition_multiple_arguments]
-    if test:
-      test(train_dataset, test_dataset)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/partition_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/partition_test.py
deleted file mode 100644
index 48f83da..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/partition_test.py
+++ /dev/null
@@ -1,84 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-import unittest
-
-import mock
-
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-from ..element_wise import partition
-
-
-def check_partitions(actual1, actual2, actual3):
-  # [START partitions]
-  annuals = [
-      {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-  ]
-  biennials = [
-      {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-  ]
-  perennials = [
-      {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-      {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-      {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-  ]
-  # [END partitions]
-  assert_that(actual1, equal_to(annuals), label='assert annuals')
-  assert_that(actual2, equal_to(biennials), label='assert biennials')
-  assert_that(actual3, equal_to(perennials), label='assert perennials')
-
-
-def check_split_datasets(actual1, actual2):
-  # [START train_test]
-  train_dataset = [
-      {'icon': '🍓', 'name': 'Strawberry', 'duration': 'perennial'},
-      {'icon': '🥕', 'name': 'Carrot', 'duration': 'biennial'},
-      {'icon': '🥔', 'name': 'Potato', 'duration': 'perennial'},
-  ]
-  test_dataset = [
-      {'icon': '🍆', 'name': 'Eggplant', 'duration': 'perennial'},
-      {'icon': '🍅', 'name': 'Tomato', 'duration': 'annual'},
-  ]
-  # [END train_test]
-  assert_that(actual1, equal_to(train_dataset), label='assert train')
-  assert_that(actual2, equal_to(test_dataset), label='assert test')
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.partition.print', lambda elem: elem)
-# pylint: enable=line-too-long
-class PartitionTest(unittest.TestCase):
-  def test_partition_function(self):
-    partition.partition_function(check_partitions)
-
-  def test_partition_lambda(self):
-    partition.partition_lambda(check_partitions)
-
-  def test_partition_multiple_arguments(self):
-    partition.partition_multiple_arguments(check_split_datasets)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/regex.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/regex.py
deleted file mode 100644
index b39b534..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/regex.py
+++ /dev/null
@@ -1,236 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-
-def regex_matches(test=None):
-  # [START regex_matches]
-  import apache_beam as beam
-
-  # Matches a named group 'icon', and then two comma-separated groups.
-  regex = r'(?P<icon>[^\s,]+), *(\w+), *(\w+)'
-  with beam.Pipeline() as pipeline:
-    plants_matches = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '🍓, Strawberry, perennial',
-            '🥕, Carrot, biennial ignoring trailing words',
-            '🍆, Eggplant, perennial',
-            '🍅, Tomato, annual',
-            '🥔, Potato, perennial',
-            '# 🍌, invalid, format',
-            'invalid, 🍉, format',
-        ])
-        | 'Parse plants' >> beam.Regex.matches(regex)
-        | beam.Map(print)
-    )
-    # [END regex_matches]
-    if test:
-      test(plants_matches)
-
-
-def regex_all_matches(test=None):
-  # [START regex_all_matches]
-  import apache_beam as beam
-
-  # Matches a named group 'icon', and then two comma-separated groups.
-  regex = r'(?P<icon>[^\s,]+), *(\w+), *(\w+)'
-  with beam.Pipeline() as pipeline:
-    plants_all_matches = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '🍓, Strawberry, perennial',
-            '🥕, Carrot, biennial ignoring trailing words',
-            '🍆, Eggplant, perennial',
-            '🍅, Tomato, annual',
-            '🥔, Potato, perennial',
-            '# 🍌, invalid, format',
-            'invalid, 🍉, format',
-        ])
-        | 'Parse plants' >> beam.Regex.all_matches(regex)
-        | beam.Map(print)
-    )
-    # [END regex_all_matches]
-    if test:
-      test(plants_all_matches)
-
-
-def regex_matches_kv(test=None):
-  # [START regex_matches_kv]
-  import apache_beam as beam
-
-  # Matches a named group 'icon', and then two comma-separated groups.
-  regex = r'(?P<icon>[^\s,]+), *(\w+), *(\w+)'
-  with beam.Pipeline() as pipeline:
-    plants_matches_kv = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '🍓, Strawberry, perennial',
-            '🥕, Carrot, biennial ignoring trailing words',
-            '🍆, Eggplant, perennial',
-            '🍅, Tomato, annual',
-            '🥔, Potato, perennial',
-            '# 🍌, invalid, format',
-            'invalid, 🍉, format',
-        ])
-        | 'Parse plants' >> beam.Regex.matches_kv(regex, keyGroup='icon')
-        | beam.Map(print)
-    )
-    # [END regex_matches_kv]
-    if test:
-      test(plants_matches_kv)
-
-
-def regex_find(test=None):
-  # [START regex_find]
-  import apache_beam as beam
-
-  # Matches a named group 'icon', and then two comma-separated groups.
-  regex = r'(?P<icon>[^\s,]+), *(\w+), *(\w+)'
-  with beam.Pipeline() as pipeline:
-    plants_matches = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '# 🍓, Strawberry, perennial',
-            '# 🥕, Carrot, biennial ignoring trailing words',
-            '# 🍆, Eggplant, perennial - 🍌, Banana, perennial',
-            '# 🍅, Tomato, annual - 🍉, Watermelon, annual',
-            '# 🥔, Potato, perennial',
-        ])
-        | 'Parse plants' >> beam.Regex.find(regex)
-        | beam.Map(print)
-    )
-    # [END regex_find]
-    if test:
-      test(plants_matches)
-
-
-def regex_find_all(test=None):
-  # [START regex_find_all]
-  import apache_beam as beam
-
-  # Matches a named group 'icon', and then two comma-separated groups.
-  regex = r'(?P<icon>[^\s,]+), *(\w+), *(\w+)'
-  with beam.Pipeline() as pipeline:
-    plants_find_all = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '# 🍓, Strawberry, perennial',
-            '# 🥕, Carrot, biennial ignoring trailing words',
-            '# 🍆, Eggplant, perennial - 🍌, Banana, perennial',
-            '# 🍅, Tomato, annual - 🍉, Watermelon, annual',
-            '# 🥔, Potato, perennial',
-        ])
-        | 'Parse plants' >> beam.Regex.find_all(regex)
-        | beam.Map(print)
-    )
-    # [END regex_find_all]
-    if test:
-      test(plants_find_all)
-
-
-def regex_find_kv(test=None):
-  # [START regex_find_kv]
-  import apache_beam as beam
-
-  # Matches a named group 'icon', and then two comma-separated groups.
-  regex = r'(?P<icon>[^\s,]+), *(\w+), *(\w+)'
-  with beam.Pipeline() as pipeline:
-    plants_matches_kv = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '# 🍓, Strawberry, perennial',
-            '# 🥕, Carrot, biennial ignoring trailing words',
-            '# 🍆, Eggplant, perennial - 🍌, Banana, perennial',
-            '# 🍅, Tomato, annual - 🍉, Watermelon, annual',
-            '# 🥔, Potato, perennial',
-        ])
-        | 'Parse plants' >> beam.Regex.find_kv(regex, keyGroup='icon')
-        | beam.Map(print)
-    )
-    # [END regex_find_kv]
-    if test:
-      test(plants_matches_kv)
-
-
-def regex_replace_all(test=None):
-  # [START regex_replace_all]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants_replace_all = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '🍓 : Strawberry : perennial',
-            '🥕 : Carrot : biennial',
-            '🍆\t:\tEggplant\t:\tperennial',
-            '🍅 : Tomato : annual',
-            '🥔 : Potato : perennial',
-        ])
-        | 'To CSV' >> beam.Regex.replace_all(r'\s*:\s*', ',')
-        | beam.Map(print)
-    )
-    # [END regex_replace_all]
-    if test:
-      test(plants_replace_all)
-
-
-def regex_replace_first(test=None):
-  # [START regex_replace_first]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants_replace_first = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '🍓, Strawberry, perennial',
-            '🥕, Carrot, biennial',
-            '🍆,\tEggplant, perennial',
-            '🍅, Tomato, annual',
-            '🥔, Potato, perennial',
-        ])
-        | 'As dictionary' >> beam.Regex.replace_first(r'\s*,\s*', ': ')
-        | beam.Map(print)
-    )
-    # [END regex_replace_first]
-    if test:
-      test(plants_replace_first)
-
-
-def regex_split(test=None):
-  # [START regex_split]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants_split = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            '🍓 : Strawberry : perennial',
-            '🥕 : Carrot : biennial',
-            '🍆\t:\tEggplant : perennial',
-            '🍅 : Tomato : annual',
-            '🥔 : Potato : perennial',
-        ])
-        | 'Parse plants' >> beam.Regex.split(r'\s*:\s*')
-        | beam.Map(print)
-    )
-    # [END regex_split]
-    if test:
-      test(plants_split)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/regex_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/regex_test.py
deleted file mode 100644
index 7e2bf78..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/regex_test.py
+++ /dev/null
@@ -1,173 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-import unittest
-
-import mock
-
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-from . import regex
-
-
-def check_matches(actual):
-  # [START plants_matches]
-  plants_matches = [
-      '🍓, Strawberry, perennial',
-      '🥕, Carrot, biennial',
-      '🍆, Eggplant, perennial',
-      '🍅, Tomato, annual',
-      '🥔, Potato, perennial',
-  ]
-  # [END plants_matches]
-  assert_that(actual, equal_to(plants_matches))
-
-
-def check_all_matches(actual):
-  # [START plants_all_matches]
-  plants_all_matches = [
-      ['🍓, Strawberry, perennial', '🍓', 'Strawberry', 'perennial'],
-      ['🥕, Carrot, biennial', '🥕', 'Carrot', 'biennial'],
-      ['🍆, Eggplant, perennial', '🍆', 'Eggplant', 'perennial'],
-      ['🍅, Tomato, annual', '🍅', 'Tomato', 'annual'],
-      ['🥔, Potato, perennial', '🥔', 'Potato', 'perennial'],
-  ]
-  # [END plants_all_matches]
-  assert_that(actual, equal_to(plants_all_matches))
-
-
-def check_matches_kv(actual):
-  # [START plants_matches_kv]
-  plants_matches_kv = [
-      ('🍓', '🍓, Strawberry, perennial'),
-      ('🥕', '🥕, Carrot, biennial'),
-      ('🍆', '🍆, Eggplant, perennial'),
-      ('🍅', '🍅, Tomato, annual'),
-      ('🥔', '🥔, Potato, perennial'),
-  ]
-  # [END plants_matches_kv]
-  assert_that(actual, equal_to(plants_matches_kv))
-
-
-def check_find_all(actual):
-  # [START plants_find_all]
-  plants_find_all = [
-      ['🍓, Strawberry, perennial'],
-      ['🥕, Carrot, biennial'],
-      ['🍆, Eggplant, perennial', '🍌, Banana, perennial'],
-      ['🍅, Tomato, annual', '🍉, Watermelon, annual'],
-      ['🥔, Potato, perennial'],
-  ]
-  # [END plants_find_all]
-  assert_that(actual, equal_to(plants_find_all))
-
-
-def check_find_kv(actual):
-  # [START plants_find_kv]
-  plants_find_all = [
-      ('🍓', '🍓, Strawberry, perennial'),
-      ('🥕', '🥕, Carrot, biennial'),
-      ('🍆', '🍆, Eggplant, perennial'),
-      ('🍌', '🍌, Banana, perennial'),
-      ('🍅', '🍅, Tomato, annual'),
-      ('🍉', '🍉, Watermelon, annual'),
-      ('🥔', '🥔, Potato, perennial'),
-  ]
-  # [END plants_find_kv]
-  assert_that(actual, equal_to(plants_find_all))
-
-
-def check_replace_all(actual):
-  # [START plants_replace_all]
-  plants_replace_all = [
-      '🍓,Strawberry,perennial',
-      '🥕,Carrot,biennial',
-      '🍆,Eggplant,perennial',
-      '🍅,Tomato,annual',
-      '🥔,Potato,perennial',
-  ]
-  # [END plants_replace_all]
-  assert_that(actual, equal_to(plants_replace_all))
-
-
-def check_replace_first(actual):
-  # [START plants_replace_first]
-  plants_replace_first = [
-      '🍓: Strawberry, perennial',
-      '🥕: Carrot, biennial',
-      '🍆: Eggplant, perennial',
-      '🍅: Tomato, annual',
-      '🥔: Potato, perennial',
-  ]
-  # [END plants_replace_first]
-  assert_that(actual, equal_to(plants_replace_first))
-
-
-def check_split(actual):
-  # [START plants_split]
-  plants_split = [
-      ['🍓', 'Strawberry', 'perennial'],
-      ['🥕', 'Carrot', 'biennial'],
-      ['🍆', 'Eggplant', 'perennial'],
-      ['🍅', 'Tomato', 'annual'],
-      ['🥔', 'Potato', 'perennial'],
-  ]
-  # [END plants_split]
-  assert_that(actual, equal_to(plants_split))
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.regex.print', lambda elem: elem)
-# pylint: enable=line-too-long
-class RegexTest(unittest.TestCase):
-  def test_matches(self):
-    regex.regex_matches(check_matches)
-
-  def test_all_matches(self):
-    regex.regex_all_matches(check_all_matches)
-
-  def test_matches_kv(self):
-    regex.regex_matches_kv(check_matches_kv)
-
-  def test_find(self):
-    regex.regex_find(check_matches)
-
-  def test_find_all(self):
-    regex.regex_find_all(check_find_all)
-
-  def test_find_kv(self):
-    regex.regex_find_kv(check_find_kv)
-
-  def test_replace_all(self):
-    regex.regex_replace_all(check_replace_all)
-
-  def test_replace_first(self):
-    regex.regex_replace_first(check_replace_first)
-
-  def test_split(self):
-    regex.regex_split(check_split)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/values.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/values.py
deleted file mode 100644
index 8504ff4..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/values.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-
-def values(test=None):
-  # [START values]
-  import apache_beam as beam
-
-  with beam.Pipeline() as pipeline:
-    plants = (
-        pipeline
-        | 'Garden plants' >> beam.Create([
-            ('🍓', 'Strawberry'),
-            ('🥕', 'Carrot'),
-            ('🍆', 'Eggplant'),
-            ('🍅', 'Tomato'),
-            ('🥔', 'Potato'),
-        ])
-        | 'Values' >> beam.Values()
-        | beam.Map(print)
-    )
-    # [END values]
-    if test:
-      test(plants)
diff --git a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/values_test.py b/sdks/python/apache_beam/examples/snippets/transforms/element_wise/values_test.py
deleted file mode 100644
index b43d911..0000000
--- a/sdks/python/apache_beam/examples/snippets/transforms/element_wise/values_test.py
+++ /dev/null
@@ -1,55 +0,0 @@
-# coding=utf-8
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import print_function
-
-import unittest
-
-import mock
-
-from apache_beam.examples.snippets.transforms.element_wise.values import *
-from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.util import assert_that
-from apache_beam.testing.util import equal_to
-
-
-@mock.patch('apache_beam.Pipeline', TestPipeline)
-# pylint: disable=line-too-long
-@mock.patch('apache_beam.examples.snippets.transforms.element_wise.values.print', lambda elem: elem)
-# pylint: enable=line-too-long
-class ValuesTest(unittest.TestCase):
-  def __init__(self, methodName):
-    super(ValuesTest, self).__init__(methodName)
-    # [START plants]
-    plants = [
-        'Strawberry',
-        'Carrot',
-        'Eggplant',
-        'Tomato',
-        'Potato',
-    ]
-    # [END plants]
-    self.plants_test = lambda actual: assert_that(actual, equal_to(plants))
-
-  def test_values(self):
-    values(self.plants_test)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/sdks/python/setup.py b/sdks/python/setup.py
index 3e42485..4548488 100644
--- a/sdks/python/setup.py
+++ b/sdks/python/setup.py
@@ -119,7 +119,10 @@
     'oauth2client>=2.0.1,<4',
     'protobuf>=3.5.0.post1,<4',
     # [BEAM-6287] pyarrow is not supported on Windows for Python 2
-    ('pyarrow>=0.11.1,<0.15.0; python_version >= "3.0" or '
+    # [BEAM-8392] pyarrow 0.14.0 and 0.15.0 triggers an exception when importing
+    # apache_beam on macOS 10.15. Update version bounds as soon as the fix is
+    # ready
+    ('pyarrow>=0.11.1,<0.14.0; python_version >= "3.0" or '
      'platform_system != "Windows"'),
     'pydot>=1.2.0,<2',
     'python-dateutil>=2.8.0,<3',