Операции кодовых моделей¶

Все операции для построения кодовых моделей импортируются из модуля aw_etl.models.code:

from aw_etl.models.code import (
    use_data_source,
    apply_sql,
    apply_join,
    apply_union,
    apply_func,
    add_calc_field,
)

Каждая операция возвращает узел графа вычислений (CodeModelNode). Из узлов складывается модель: узлы-источники преобразуются операциями, а итоговый узел возвращается из main().

Все операции вычисляются лениво: они описывают преобразование, но не выполняют его. Данные обрабатываются, когда итоговый датафрейм запрашивается.

use_data_source¶

Получает данные из источника данных, описанного в model.yml.

Параметр	Тип	Описание
`data_source`	биндинг источника	Элемент `data_sources.<name>` из параметра `main()`. Обязательно
`sql`	`str`	SQL-запрос к источнику. Необязательно — без него берётся вся таблица

from aw_etl.models.code import use_data_source, apply_join


def main(data_sources):
    """ """
    orders = use_data_source(data_sources.orders, sql="select * from orders")
    products = use_data_source(data_sources.products)

    return apply_join(
        left=orders, right=products, left_on="product_id", right_on="id", how="left"
    )

При указании sql источник обрабатывается как запрос — тип источника в model.yml должен быть sql.

apply_sql¶

Выполняет SQL-запрос над одним или несколькими узлами графа. Узлы регистрируются как временные представления (temp view) с указанными именами, и к ним можно обращаться в запросе по имени.

Параметр	Тип	Описание
`nodes`	`CodeModelNode` или `list[tuple[str, CodeModelNode]]` или `None`	Узлы, доступные в запросе. Один узел, список пар `(имя, узел)` или `None`, если запрос ни от чего не зависит
`sql`	`str`	SQL-запрос (диалект Spark SQL)

from aw_etl.models.code import apply_sql, use_data_source


def main(data_sources):
    """ """
    orders = use_data_source(data_sources.orders)

    # один узел доступен под именем child
    big = apply_sql(orders, 'select * from child where summa > 1000')

    # несколько узлов — под своими именами
    summary = apply_sql(
        nodes=[('orders', orders), ('big', big)],
        sql='''
            select o.customer_id, count(*) as orders_cnt, sum(o.summa) as total
            from orders o join big b on o.id = b.id
            group by o.customer_id
        ''',
    )
    return summary

apply_join¶

Соединяет два узла графа.

Параметр	Тип	Описание
`left`	`CodeModelNode`	Левый узел
`right`	`CodeModelNode`	Правый узел
`left_on`	`str` или `list[str]`	Ключи соединения слева
`right_on`	`str` или `list[str]`	Ключи соединения справа
`how`	`str`	Тип соединения: `inner`, `left`, `right`, `outer` и др. По умолчанию `inner`
`rename`	`str`	Суффикс для одноимённых колонок правого узла: `left` или `right`. По умолчанию `right`
`keep_key`	`str` или `None`	Какую версию ключа оставить в результате. По умолчанию `None`

from aw_etl.models.code import use_data_source, apply_join


def main(data_sources):
    """ """
    orders = use_data_source(data_sources.orders)
    products = use_data_source(data_sources.products)

    return apply_join(orders, products, 'product_id', 'id', how='left')

Если ключи называются одинаково (id в обоих узлах), колонка правого узла переименовывается согласно rename (например, id_right), либо указывается keep_key, чтобы оставить только одну версию ключа.

apply_union¶

Объединяет несколько узлов с одинаковой структурой (union).

Параметр	Тип	Описание
`nodes`	`list[CodeModelNode]`	Список объединяемых узлов
`union_all`	`bool`	`True` (по умолчанию) — union all (без дедупликации), `False` — с дедупликацией
`by_name`	`bool`	`False` (по умолчанию) — объединение по позициям колонок; `True` — по именам колонок
`allow_missing_columns`	`bool`	При `by_name=True` разрешает отсутствие колонок в некоторых узлах (заполняются `null`)

from aw_etl.models.code import apply_union, apply_sql


def main():
    """ """
    orders_2024 = apply_sql(None, 'select id, summa from orders_2024')
    orders_2025 = apply_sql(None, 'select id, summa from orders_2025')

    return apply_union([orders_2024, orders_2025])

apply_func¶

Применяет произвольную python-функцию к узлам графа. Функция получает датафреймы узлов и возвращает новый датафрейм — так в модель встраивается любая логика, которую неудобно выразить SQL.

Параметр	Тип	Описание
`nodes`	`CodeModelNode` или `list[tuple[str, CodeModelNode]]` или `None`	Узлы, передаваемые в функцию
`func`	`Callable`	Функция, возвращающая `DataFrame`

Функция func получает параметры в зависимости от собственной сигнатуры:

Имя параметра	Что получает
`spark`	Сессия Spark
`df`	Датафрейм первого узла (если `nodes` — один узел или список)
`dfs`	Словарь `{имя: датафрейм}` всех узлов

from aw_etl.models.code import apply_sql, apply_func


def main():
    """ """
    s1 = apply_sql(None, 'select 1 as id1, "a1" as name1')
    s2 = apply_sql(None, 'select 1 as id2, "b1" as name2')

    return apply_func(nodes=[('s1', s1), ('s2', s2)], func=cross_join)


def cross_join(dfs):
    return dfs['s1'].crossJoin(dfs['s2'])

Ограничения на func

Функция должна быть объявлена на верхнем уровне модуля model.py: лямбды и вложенные функции не поддерживаются — при выполнении в AW функция ищется в модуле модели по имени. Обязательные параметры функции могут быть только spark, df, dfs — остальные параметры должны иметь значения по умолчанию.

Работа с датафреймами Spark описана в разделе Spark.

add_calc_field¶

Добавляет в граф узел с вычисляемым полем. Значение поля считается по SQL-выражению от колонок узла.

Параметр	Тип	Описание
`node`	`CodeModelNode`	Узел, к которому добавляется поле
`field_name`	`str`	Название вычисляемого поля
`expression`	`str`	SQL-выражение значения поля (диалект Spark SQL)

from aw_etl.models.code import use_data_source, add_calc_field


def main(data_sources):
    """ """
    orders = use_data_source(data_sources.orders)

    n = add_calc_field(orders, 'order_day', 'extract(day from created_at)')
    n = add_calc_field(n, 'order_month', 'extract(month from created_at)')
    n = add_calc_field(n, 'order_year', 'extract(year from created_at)')
    n = add_calc_field(n, 'summa_with_discount', 'summa - 0.3 * summa')

    return n

Поля, добавляемые через add_calc_field, удобно декларировать в model.yml и включать в иерархии — например, order_year → order_month → order_day.