Note

Because this blog post represents an actual project from my job, and this is not my employer’s official blog, I have obscured or changed certain key facts. Specifically, astute readers will notice I am vague and shifty about where datasets originated from — where they were stored and how they were retrieved, that is. That’s deliberate. Those changes do not effect the substance of this article, however.

Aho-Corasick is a classic multi-pattern string matching algorithm — like regex, but significantly faster. Unlike regex, which (depending on the implementation) may scale exponentially with input size, Aho-Corasick scales linearly. That property — linear time complexity — makes it ridiculously fast, and ideal for large-scale data remediation.

I first learned about Aho-Corasick back in 2018 but filed it away under “interesting but unnecessary for me right now.” Years later, I found myself responsible for remediating records across datasets at petabyte scale.

At the time, remediations were being performed using SQL queries embedded with regular expressions, running against an already-overloaded relational database. The approach was inefficient: most client datasets took between 30 minutes to an hour to process, and in some cases stretched as long as 55 hours. As you can probably imagine, those datasets with excessive runtimes were enormous.

Recognizing an opportunity to improve things, I began re-architecting the remediation code. Given the scale of the data and the need for runtime and memory efficiency, it quickly became clear that the shared RDBMS was a major bottleneck. Profiling revealed that regular expressions — not just SQL overhead and database performance — were the primary cause of long runtimes. As the number and length of substrings grew, performance degraded exponentially. This also proved true when using Python’s built-in re module.

In modern data engineering, large datasets are often processed using distributed systems like Apache Spark, with custom user-defined functions to apply transformation logic. Another common tool is DuckDB, which also supports UDFs and is popular for fast local analytics and ease of use.

However, DuckDB began leaking memory at scale — confirming concerns previously raised by the community. And while Spark offered better scale, efficiently sharing a large Aho-Corasick automaton between executors introduced more complexity than it solved. User-defined functions also tend to be inefficient.

To avoid shared-memory challenges in Spark and instability in DuckDB, I opted to use Python’s built-in multiprocessing library in tandem with Aho-Corasick.

Since Python multiprocessing typically involves deep memory copying between processes, which would have blown up RAM usage at scale, I used the fork start method on Linux to ensure memory sharing via copy-on-write semantics. This allowed large, immutable data structures (like the target dataset and automatons) to be reused across processes without duplication, keeping memory usage comfy.

Additionally, I took care to scope all shared data within a global cache, which avoids passing large payloads via Pool.apply_async(...) calls and minimizes serialization overhead.

The results were staggering: 55 hours → 7 seconds.

A key reason for this performance wasn’t simply the use of Aho-Corasick but how the data was shared. By loading large datasets and compiled automatons once, and forking worker processes afterward, I avoided memory duplication entirely. If I had passed these structures via arguments or used spawn, it would have resulted in prohibitive memory usage and slower compute due to unnecessary serialization and GC pressure.

Critical Concepts ¶

Before you rush to implement Aho-Corasick expecting miracles, a word of caution: Aho-Corasick + parallelization won’t yield superb performance unless your code is optimized.

To borrow from the pseudo-code further below, you’ll need to understand:

• The difference between “spawn” and “fork” memory allocation modes in Python’s multiprocessing module and how fork enables memory-efficient parallelism via copy-on-write. This matters because using spawn (the default on macOS and Windows) will fully copy large objects, causing massive memory spikes if you’re not careful. On Linux, fork allows those objects to be shared so long as they’re never mutated.
• The importance of the “tidy data” principle and general data layout — so your data can be scanned efficiently.¹
• The necessity of profiling your code, early and often.²

Additionally, realize you may not need to use the multiprocessing library after all. You might be able to write a user-defined function that’s implemented in DuckDB or Spark. That decision depends primarily on the scale of your data³ and-or comfort with digging deep into Spark. To be honest, I actually recommend that you use a user-defined function in DuckDB — that is, if the scale of your data isn’t enormous. It will be less efficient than using Aho-Corasick + multiprocessing but certainly simpler.

The pseudo-code that follows accepts two pandas DataFrame objects: target and sensitive_values.

• target contains the data that must be scanned for sensitive content and remediated sensitive_values contains the values to search for and obfuscate
• The Aho-Corasick automaton does not care how sensitive values appear in target—whether as substrings or exact matches. All matches are remediated the same way.

For example:

• If "1234" is a sensitive value, then "1234_5678" becomes "xXxX_5678"
• If "1234" appears on its own, it becomes "xXxX".

Multiple matches per record are handled without issue. Referring back to the tidy data concept: sensitive_values is represented in what I will call a “long format”.

1
2
3
4
5

[
  {"element_name": "x", "sensitive_value": "1234"},
  {"element_name": "x", "sensitive_value": "5678"},
  {"element_name": "y", "sensitive_value": "abc"}
]

Why does this matter so much . . . ? Because this long format allows the data to be grouped by element_name, deduplicated, and rapidly loaded into per-column automatons. Each parallel process then scans each record in each column for matches in its associated automaton.

Dependencies ¶

You’ll need to download pyahocorasick and pandas. You can use polars instead of pandas if you prefer. polars may actually make the following pseudo-code even faster, albeit marginally⁴. There are also Rust-based implementations available.

Pseudo-Code ¶

Tip

You can copy lines of code or permalinks by selecting the line numbers below and clicking the ellipses button. To select multiple lines, press and hold Shift (⇧) as you select the desired line range.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

import gc
from collections import defaultdict
from multiprocessing import Pool, get_context

import pandas as pd
from ahocorasick import Automaton


# Singleton cache used to hold large, immutable shared data 
# (e.g. automatons, target df).
# Under 'fork' mode, this data is shared across processes 
# via copy-on-write, avoiding memory duplication.
class ObfuscatorCache:
    def __init__(self):
        self.automatons = None
        self.target = None
        self.executor = None
        self.metrics = 0


# initializing global cache object for sharing immutable
# data between processes
cache = ObfuscatorCache()


class Obfuscator:
    def __init__(
        self,
        sensitive_values: pd.DataFrame,
        target: pd.DataFrame,
        automatons: dict[str, Automaton] | None = None,
        max_workers: int | None = None,
    ):
        self._sensitive_values = sensitive_values
        self._max_workers = 5 if max_workers is None else max_workers
        
        # Enables copy-on-write memory sharing on Linux/Unix
        self._ctx = get_context("fork")
        cache.target = target
        cache.automatons = (
            self.make_automatons(sensitive_values)
            if automatons is None
            else automatons
        )

    @property
    def target(self) -> pd.DataFrame:
        return cache.target

    @target.setter
    def target(self, value: pd.DataFrame):
        cache.target = value
        if hasattr(self, "_ctx"):
            self._ctx.target = value

    @property
    def metrics(self) -> int:
        return cache.metrics

    @metrics.setter
    def metrics(self, value: int):
        cache.metrics = value

    @property
    def automatons(self) -> dict[str, Automaton]:
        return cache.automatons

    @property
    def executor(self) -> Pool:
        if cache.executor is None:
            cache.executor = self._ctx.Pool(processes=self._max_workers)
        return cache.executor

    def shutdown_executor(self):
        if cache.executor is not None:
            cache.executor.terminate()
            cache.executor.join()
            cache.executor = None

    @staticmethod
    def make_automaton(patterns: set) -> Automaton:
        A = Automaton()
        for pattern in patterns:
            A.add_word(pattern, pattern)
        A.make_automaton()
        return A
    
    def make_automatons(self, sensitive_values: pd.DataFrame) -> (
        dict[str, Automaton]
    ):
        findings = defaultdict(set)
        for element, group in sensitive_values.groupby("element_name"):
            findings[element].update(
                str(finding) for finding in group["sensitive_value"]
            )
        del self._sensitive_values
        return {
            element: self.make_automaton(patterns)
            for element, patterns in findings.items()
        }
    
    def obfuscate(self, inplace: bool = True) -> pd.DataFrame:
        if not inplace:
            obfuscated_target = self.target.copy()
            futures = [
                self.executor.apply_async(self._obfuscate_column, args=(column,))
                for column in self.automatons.keys()
            ]
            for future in futures:
                column, obfuscated_series, local_counts = future.get()
                obfuscated_target[column] = obfuscated_series
                self.metrics += local_counts[column]
            return obfuscated_target
        
        else:
            futures = [
                self.executor.apply_async(self._obfuscate_column, args=(column,))
                for column in self.automatons.keys()
            ]
            for future in futures:
                column, obfuscated_series, local_counts = future.get()
                self.target[column] = obfuscated_series
                self.metrics += local_counts[column]
                del column, obfuscated_series, local_counts, future
                gc.collect()
            return self.target
    
    def _obfuscate_column(self, column: str) -> (
        tuple[str, pd.Series, dict[str, int]]
    ):
        local_metrics = defaultdict(int)
        series = self.target[column]
        obfuscated_series = series.map(
            lambda v: self._obfuscate_record(
                v, self.automatons, column, local_metrics
            )
        )
        return column, obfuscated_series, local_metrics
    
    @staticmethod
    def _obfuscate_record(
        value: str,
        automatons: dict[str, Automaton],
        column: str,
        counter: dict,
        obfuscate_char: str = "x",
    ) -> str:
        if not (
            matches := [
                (end - len(match) + 1, end)
                for end, match in automatons[column].iter(value)
            ]
        ):
            return value
        
        merged = []
        for start, end in sorted(matches):
            if not merged or start > merged[-1][1]:
                merged.append([start, end])
            else:
                merged[-1] = (merged[-1][0], max(merged[-1][1], end))
        
        chars = list(value)
        counter[column] += len(merged)
        
        for start, end in merged:
            length = end - start + 1
            pattern = (
                (obfuscate_char + obfuscate_char.upper()) * ((length + 1) // 2)
            )
            chars[start : end + 1] = list(pattern[:length])
        return "".join(chars)