Using filter() for Selection

Concepts covered: Functional Transforms

filter(function, iterable) runs the function against every item and keeps those for which it returns True. A function used this way is called a predicate: it answers a yes-or-no question about 1 item. Like a map object, the result is a filter object that computes each item as you read it and can be read only once.

runs = [
    {"job": "load_orders", "status": "ok", "minutes": 42},
    {"job": "dedupe_users", "status": "failed", "minutes": 7},
    {"job": "build_daily", "status": "ok", "minutes": 19},
    {"job": "sync_crm", "status": "failed", "minutes": 7},
]


def is_failed(run):
    return run["status"] == "failed"


for run in filter(is_failed, runs):
    print(run["job"], run["minutes"])

Output:

dedupe_users 7
sync_crm 7

The 2 dicts printed are the same objects that sit in runs. Nothing in them changed, and they come out in their original order. filter() chooses which items survive and never edits them, which gives you 2 properties to check in any output: it holds at most as many items as the input, and every item in it came from the input.

map()

• Function returns a new value for each item
• Output has exactly as many items as the input
• Items come out changed
• Answers: what does each item become?

filter()

• Function returns True or False for each item
• Output has as many items as the input, or fewer
• Items come out unchanged
• Answers: which items stay?

Predicates that test several conditions

A predicate can combine conditions with and and or, as long as it ends by returning True or False. It receives exactly 1 argument, the item, so a threshold it needs has to be written inside the function or read from a constant defined at module level.

SLOW_MINUTES = 10


def is_slow_success(run):
    return run["status"] == "ok" and run["minutes"] > SLOW_MINUTES


runs = [
    {"job": "load_orders", "status": "ok", "minutes": 42},
    {"job": "dedupe_users", "status": "failed", "minutes": 7},
    {"job": "build_daily", "status": "ok", "minutes": 19},
    {"job": "sync_crm", "status": "failed", "minutes": 7},
]
slow = list(filter(is_slow_success, runs))
print(len(slow), slow[0]["job"])

Output:

2 load_orders

load_orders at 42 minutes and build_daily at 19 pass both tests, so slow holds 2 dicts. The 2 failed runs fall at the status test, whatever their minutes. A threshold kept in a named constant such as SLOW_MINUTES changes in one place when the rule changes.

Filtering a dict

Looping over a dict gives its keys, so a predicate on values needs .items(), where each item is a (key, value) pair. Pass the surviving pairs to dict() to rebuild the mapping. Here only tables that loaded rows stay:

row_counts = {"orders": 1200, "users": 0, "events": 98000, "refunds": 0}


def has_rows(pair):
    return pair[1] > 0


print(dict(filter(has_rows, row_counts.items())))

Output:

{'orders': 1200, 'events': 98000}

Dropping missing values with a predicate

Earlier, None in a latency column broke sorted(), and a key placed it last. When missing readings should not appear in the result at all, a predicate built on is not None removes them, and sorted() takes the filter object directly:

def is_present(value):
    return value is not None


latencies = [120, None, 85, None, 240]
present = sorted(filter(is_present, latencies))
print(present)
print(len(latencies) - len(present), "readings missing")

Output:

[85, 120, 240]
2 readings missing

Placing gaps last and dropping them are 2 different report decisions. A chart built from the filtered list looks complete. Print or log how many items the filter removed, as the last line does, so whoever reads the result knows 2 of the 5 readings never reached it.

Chaining filter, map and sorted

Each of these functions accepts the others' output, so the 3 calls can nest in a single expression. Read it from the inside out: the innermost call is the first to touch the raw list.

runs = [
    {"job": "load_orders", "status": "ok", "minutes": 42},
    {"job": "dedupe_users", "status": "failed", "minutes": 7},
    {"job": "build_daily", "status": "ok", "minutes": 19},
    {"job": "sync_crm", "status": "failed", "minutes": 7},
]


def is_failed(run):
    return run["status"] == "failed"


def job_name(run):
    return run["job"]


print(sorted(map(job_name, filter(is_failed, runs)), reverse=True))

Output:

['sync_crm', 'dedupe_users']

Put filter() inside map() when the order is yours to choose. The transform then runs on fewer items, and the predicate sees the full original record with every field it might test, before map() reduces each record to a name.

The arguments to filter() are swapped. What happens when this runs?

def is_failed(run):
    return run["status"] == "failed"

runs = [{"job": "sync_crm", "status": "failed"}]
kept = filter(runs, is_failed)
print("built")
print(list(kept))

• It prints built, then raises TypeError at list()
• It raises TypeError at the filter() line, before printing
• It prints built, then an empty list
• It prints built, then the sync_crm record

About this lesson

This section is part of the Collections: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.