Код модели: model.py¶

Трансформация данных в кодовой модели описывается в файле model.py, который располагается в папке модели рядом с метаданными model.yml.

Точка входа: функция main()¶

Единственное требование к коду модели — наличие функции main(), которая возвращает итоговый узел графа вычислений:

from aw_etl.models.code import apply_sql


def main():
    """ """
    return apply_sql(None, 'select 1 as id, "test" as name')

Функция main() может принимать параметры — набор определяется по сигнатуре функции:

Параметр	Тип	Описание
`spark`	`SparkSession`	Локальная сессия Spark. Нужна, если модель работает с датафреймами напрямую (например, в функциях `apply_func`)
`data_sources`	`DataSourceBindings`	Доступ к источникам данных, описанным в `model.yml`

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

Если параметр не объявлен в сигнатуре — он не будет передан. Модель без источников данных может обойтись вообще без параметров.

Граф вычислений¶

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

from aw_etl.models.code import use_data_source, apply_join, add_calc_field


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

    n = apply_join(orders, products, 'product_id', 'id', how='left')
    n = apply_join(n, customers, 'customer_id', 'id', how='left')

    n = add_calc_field(n, 'order_day', 'extract(day from created_at)')
    n = add_calc_field(n, 'summa_with_discount', 'summa - 0.3 * summa')

    return n

Источники данных из data_sources превращаются в стартовые узлы графа.
Операции apply_join соединяют узлы между собой.
add_calc_field добавляет вычисляемые поля.
Возвращаемый узел становится итоговым датафреймом модели — его структура должна соответствовать списку полей из model.yml.

Граф вычисляется лениво: каждая операция лишь описывает преобразование, а данные обрабатываются, когда итоговый датафрейм запрашивается (при запуске модели в AW или при локальном запуске).

Вспомогательные функции¶

Помимо main() в model.py можно определять вспомогательные функции и константы. Типичный приём — выделить логику для apply_func в отдельную функцию верхнего уровня:

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

Ограничения на функции apply_func

Функция, передаваемая в apply_func, должна быть функцией верхнего уровня (module-level). Лямбды и вложенные функции не поддерживаются — так функция однозначно находится в модуле модели при выполнении в AW.

Чего в коде модели нет¶

Кодовая модель — это чистое описание трансформации. В model.py не следует:

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

Проверка кода локально¶

Проверить, что модель собирается и возвращает корректные данные, можно локально, не публикуя код в репозиторий. Для этого используются тесты — python-функция run_code_model() запускает модель на локальном Spark с тестовыми данными.