Filter

Pydoc Pydoc




Given a predicate, filter out all elements that don’t satisfy that predicate. May also be used to filter based on an inequality with a given value based on the comparison ordering of the element.

Examples

In the following examples, we create a pipeline with a PCollection of produce with their icon, name, and duration. Then, we apply Filter in multiple ways to filter out produce by their duration value.

Filter accepts a function that keeps elements that return True, and filters out the remaining elements.

Example 1: Filtering with a function

We define a function is_perennial which returns True if the element’s duration equals 'perennial', and False otherwise.

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)
  )

Output PCollection after Filter:

perennials = [
    {'icon': 'πŸ“', 'name': 'Strawberry', 'duration': 'perennial'},
    {'icon': 'πŸ†', 'name': 'Eggplant', 'duration': 'perennial'},
    {'icon': 'πŸ₯”', 'name': 'Potato', 'duration': 'perennial'},
]
Run in Colab Run code now
View source code View source code




Example 2: Filtering with a lambda function

We can also use lambda functions to simplify Example 1.

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)
  )

Output PCollection after Filter:

perennials = [
    {'icon': 'πŸ“', 'name': 'Strawberry', 'duration': 'perennial'},
    {'icon': 'πŸ†', 'name': 'Eggplant', 'duration': 'perennial'},
    {'icon': 'πŸ₯”', 'name': 'Potato', 'duration': 'perennial'},
]
Run code now Run code now
View source code View source code




Example 3: Filtering with multiple arguments

You can pass functions with multiple arguments to Filter. They are passed as additional positional arguments or keyword arguments to the function.

In this example, has_duration takes plant and duration as 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)
  )

Output PCollection after Filter:

perennials = [
    {'icon': 'πŸ“', 'name': 'Strawberry', 'duration': 'perennial'},
    {'icon': 'πŸ†', 'name': 'Eggplant', 'duration': 'perennial'},
    {'icon': 'πŸ₯”', 'name': 'Potato', 'duration': 'perennial'},
]
Run in Colab Run code now
View source code View source code




Example 4: Filtering with side inputs as singletons

If the PCollection has a single value, such as the average from another computation, passing the PCollection as a singleton accesses that value.

In this example, we pass a PCollection the value 'perennial' as a singleton. We then use that value to filter out perennials.

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)
  )

Output PCollection after Filter:

perennials = [
    {'icon': 'πŸ“', 'name': 'Strawberry', 'duration': 'perennial'},
    {'icon': 'πŸ†', 'name': 'Eggplant', 'duration': 'perennial'},
    {'icon': 'πŸ₯”', 'name': 'Potato', 'duration': 'perennial'},
]
Run in Colab Run code now
View source code View source code




Example 5: Filtering with side inputs as iterators

If the PCollection has multiple values, pass the PCollection as an iterator. This accesses elements lazily as they are needed, so it is possible to iterate over large PCollections that won’t fit into memory.

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)
  )

Output PCollection after Filter:

valid_plants = [
    {'icon': 'πŸ“', 'name': 'Strawberry', 'duration': 'perennial'},
    {'icon': 'πŸ₯•', 'name': 'Carrot', 'duration': 'biennial'},
    {'icon': 'πŸ†', 'name': 'Eggplant', 'duration': 'perennial'},
    {'icon': 'πŸ…', 'name': 'Tomato', 'duration': 'annual'},
]
Run in Colab Run code now
View source code View source code




Note: You can pass the PCollection as a list with beam.pvalue.AsList(pcollection), but this requires that all the elements fit into memory.

Example 6: Filtering with side inputs as dictionaries

If a PCollection is small enough to fit into memory, then that PCollection can be passed as a dictionary. Each element must be a (key, value) pair. Note that all the elements of the PCollection must fit into memory for this. If the PCollection won’t fit into memory, use beam.pvalue.AsIter(pcollection) instead.

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)
  )

Output PCollection after Filter:

perennials = [
    {'icon': 'πŸ“', 'name': 'Strawberry', 'duration': 'perennial'},
    {'icon': 'πŸ†', 'name': 'Eggplant', 'duration': 'perennial'},
    {'icon': 'πŸ₯”', 'name': 'Potato', 'duration': 'perennial'},
]
Run in Colab Run code now
View source code View source code




Pydoc Pydoc