Использование шаблона строителя для конфигурируемых трубопроводов данных в области проектирования данных

class PipelineBuilder:
 def __init__(self):
 self._source = None
 self._transformations = []
 self._sinks = []
 self._retry_policy = None

 def with_source(self, source):
 self._source = source
 return self

 def add_transform(self, transform):
 self._transformations.append(transform)
 return self

 def add_sink(self, sink):
 self._sinks.append(sink)
 return self

 def with_retry(self, retry_policy):
 self._retry_policy = retry_policy
 return self

 def build(self):
 if not self._source or not self._sinks:
 raise ValueError("Source and at least one sink are required")
 return Pipeline(self._source, self._transformations, self._sinks, self._retry_policy)

С помощью строителя создание трубопровода становится декларативным:

pipeline = (PipelineBuilder()
 .with_source(S3CsvSource(bucket="data-landing", prefix="orders/"))
 .add_transform(FilterTransform(condition="status == 'active'"))
 .add_transform(AggregateTransform(group_by="customer_id", metrics=["sum(amount)"]))
 .add_sink(DatabaseSink(connection="prod_db", table="customer_orders"))
 .add_sink(ParquetSink(path="s3://analytics/orders/"))
 .with_retry(RetryPolicy(max_attempts=3, backoff_seconds=5))
 .build())

Такой подход централизует конфигурацию, что позволяет легко повторно использовать один и тот же конструктор с различными параметрами для постановки и производственной среды.

Реальное применение: построение гибкого трубопровода ETL

Рассмотрим компанию электронной коммерции, которая должна потреблять данные о ежедневных заказах из нескольких регионов, очищать и стандартизировать их, вычислять ежедневный доход по категориям и загружать результаты в базу данных отчетности и озеро данных. Используя шаблон сборщика, они создают многоразовый OrderETLBuilder .

1. Определение исходных конфигураций: Заказы каждого региона поступают из разных баз данных (PostgreSQL, MySQL), но экспортируются в общий формат CSV.
2. Добавить стандартные преобразования: Очистка данных (удалить идентификаторы нулевого заказа, проверить коды валют) и обогащение (объединиться с каталогом продуктов, чтобы получить категорию).
3. Настройка агрегации: .
4. Путь к нескольким раковинам: и .
5. Строить и выполнить: Тот же строитель может сначала построить трубопровод, который считывает только регион ЕС для тестирования, а затем поменять на все регионы для производства.

Эта модель значительно снижает дублирование кода: компания теперь поддерживает один класс строителей вместо нескольких специальных скриптов в каждом регионе или среде.

Преимущества Recap

• Гибкость: Изменить поведение конвейера, не касаясь логики исполнения. Нужно добавить новую трансформацию? Просто позвоните с новым шагом.
• Устойчивость: Определения трубопровода читаются как рецепт высокого уровня. Конфигурация каждого компонента изолирована, что делает отладку и обзор кода простыми.
• Многоразовые: Строители могут быть упакованы в библиотеки. Команды повторно используют один и тот же конструктор в разных проектах, корректируя только входные параметры.
• Масштабируемость: Добавление нового типа компонента (например, потоковой мойки) требует только расширения сборщика, а не переписывания всей сборки трубопровода.
• Проверяемость: Строители могут создавать тестовые трубопроводы с искусственными источниками и поглотителями, что позволяет проводить изолированные единичные испытания для самой логики сборки трубопровода.

Лучшие практики использования шаблона строителя в области обработки данных

Держите конфигурацию строителя чистой

Строитель должен собирать и проверять только конфигурацию. Фактический процесс выполнения трубопровода должен быть ответственностью объекта Pipeline , построенного . Это разделение делает конструктор простым и проверяемым.

Проверить рано, провалить быстро

В методе убедитесь, что все необходимые компоненты присутствуют и что конфигурации являются согласованными (например, этапы преобразования ссылаются на существующие столбцы источника).

Неизменные конструкции Immutable Builds

После того, как назван, конструктор может быть сброшен или повторно использован для создания другого трубопровода с различными настройками. Избегайте состояния хранения, которое сохраняется на сборках, если это не преднамеренно.

Обеспечить разумные недостатки

Для дополнительных компонентов, таких как политика повторного использования или регистрация, установите разумные по умолчанию в конструкторе строителя. Это минимизирует шаблон, все еще позволяя переопределять.

Версия: Ваш строитель рядом с вашими трубопроводами

По мере развития вашей инфраструктуры данных API строителя также будет. Tag builder выпускает в управлении версиями, поэтому определения конвейера могут прикрепляться к конкретной версии строителя, предотвращая неожиданное распространение изменений.

Использование внешних ссылок для сложных компонентов

Для компонентов со многими внутренними деталями (например, конфигурация сеанса Spark или пользовательский UDF) рассматривайте их как предварительно построенные объекты, а не как построенные внутри строителя трубопровода. Рефакторинг.