Макросы dbt для ClickHouse

{% macro format_camp(t) %}
{% set t = 
    {"creo":"_multi_", "creo_name":"мультиформат"}, 
    {"creo":"_preroll_", "creo_name":"преролл"},
    {"creo":"_post_", "creo_name":"пост "},
    {"creo":"_clip_", "creo_name":"клипы"},
    {"creo":"_tgb_", "creo_name":"ТГБ"},
    {"creo":"_lead_", "creo_name":"лид формы"}, 
    {"creo":"_dzen_", "creo_name":"дзен "}, 
    {"creo":"_vid_", "creo_name":"видео"},
    {"creo":"_carousel_", "creo_name":"карусель"}, 
    {"creo":"_story_", "creo_name":"истории"},
    {"creo":"_prospecting_", "creo_name":"проспектинг"},
    {"creo":"_dinrem_", "creo_name":"динамический ретаргетинг"}

 %}
{% for value in t %}
      WHEN match(lower(campaign), lower('{{ value.creo }}')) THEN '{{ value.creo_name }}'
{% endfor %}
{% endmacro %}

{% macro incremental_filter_ch(column) %}
    {% if is_incremental() %}
        where {{ column }} > (select max({{ column }}) from {{ this }})
    {% endif %}
{% endmacro %}

select * from events
{{ incremental_filter_ch('event_time') }}

{% macro var_or_daysago(key, n=1) -%}
  {% set _ndaysago = (modules.datetime.datetime.now() - modules.datetime.timedelta(days=n)).strftime('%Y-%m-%d') -%}
  {% set result = var(key, default=_ndaysago) -%}
  toDate(parseDateTimeBestEffort('{{ result }}'))
{%- endmacro %}

where
  date between {{ var_or_daysago('start_date', 30) }}
  and {{ var_or_daysago('end_date', 1) }}

Если мне нужно дедуплицировать, предварительно удалив старые записи, то тот же макрос используется в pre_hook. Сперва проверяем, что запуск инкрементальный, иначе удаление может выдать ошибку, а потом yeahbuddylightweightbaby с параметрами. Главное - не налажать с кавычками.

{{ config(
  pre_hook = "{% if is_incremental() %}delеte from {{this}} on cluster  {{target.cluster}} where date between {{ var_or_daysago('start_date', 30) }} and {{ var_or_daysago('end_date', 1) }} {% endif %}"
) }} 

Я не рекомендую данную стратегию дедупликации, но, когда есть выбор, это хорошо.

Партиции

ClickHouse очень не любит мутаций. Они про это целую статью написали и вообще рекомендуют производить мутации только администратору БД 😅 Одна из очевидных стратегий дедупликаций - замена партиций.

Нашёл вот такой пример генерации запроса на удаление. Его предназначение - сформировать массив названий недельных партиций за период, а потом перечислить их в дропе.

{% macro prehook_drop_partition_yyyyww(this, startDate, endDate) -%}

    {% set start_week_date = modules.datetime.datetime.strptime(startDate, '%Y-%m-%d') - modules.datetime.timedelta(days=modules.datetime.datetime.strptime(startDate, '%Y-%m-%d').weekday()) %}
    {% set end_week_date = modules.datetime.datetime.strptime(endDate, '%Y-%m-%d') - modules.datetime.timedelta(days=modules.datetime.datetime.strptime(endDate, '%Y-%m-%d').weekday()) %}

    {% set diff_weeks = ((end_week_date - start_week_date).days // 7) + 1 %}

    {% set week_list = [] %}
    {% for i in range(diff_weeks) %}
      {% set week_date = start_week_date + modules.datetime.timedelta(weeks=i) %}
      {% set week_str = week_date.strftime('%G%V') %}
      {% do week_list.append(week_str) %}
    {% endfor %}

    alter table {{ this }} on cluster {{target.cluster}}
    {% for week in week_list %}
    drop partition {{ week }}
		{%- if not loop.last %},{% endif %}
    {%- endfor %}
{%- endmacro %}

Пример использования

pre_hook ="{{ prehook_drop_partition_yyyyww(this, var_or_daysago_raw('start_date', 65), var_or_daysago_raw('end_date', 1)) }}"

Проверка дублей в источнике

{% macro check_unique_key(source_ref, columns) %}
    {% set columns_str = columns | join(', ') %}
    with duplicates AS (
        select count(*) AS count
        from {{ source_ref }}
        group by {{ columns_str }}
        having count(*) > 1
    )
    select 
        count(*) AS duplicate_count 
    from duplicates
{% endmacro %}

Пример использования в другом 🤯 макросе

{% macro validate_duplicate_in_source_table(source_ref, columns) %}
    {% set result = run_query(check_unique_key(source_ref, columns)) %}
    {% set duplicate_count = result.columns[0].values()[0] if result else 0 %}
    
    {% if execute %}
        {% if duplicate_count > 0 %}
            {{ log("Дубликаты найдены в " ~ source_ref ~ ". Таблица "  ~ source_ref ~ " будет проигнорирована.", info=True) }}
            {{ return("(select * from " ~ source_ref ~ " where date = today() limit 1)") }}
        {% else %}
            {{ log("Дубликаты не найдены в " ~ source_ref, info=True) }}
            {{ return(source_ref) }}
        {% endif %}
    {% endif %}
{% endmacro %}

Что тут происходит? Проверяем наличие дублей по ключу. Если нашлись, выдаём предупреждением и скипаем источник болванкой. Если дублей нет - всё ок, просто логируем факт проверки и разрешаем обратиться к источнику. Соответственно, в моделях в него оборачивается ref.

from {{ validate_duplicate_in_source_table(ref('campaign_data'), ['date', 'campaign_id', 'placement_id']) }}

Проверка репликации

{% macro validate_replication(model, cluster_name='default') %}
    {%- set query %}
        select
            fqdn(), table
        from clusterAllReplicas({{cluster_name}}, system.tables)
        where database = '{{ model.schema }}' and table = '{{ model.name }}'
        and total_rows = 0
    {%- endset %}

    {%- set result = run_query(query) %}

    {% if execute %}
        {% if result|length > 0 %}
            {{ exceptions.raise_compiler_error("🚨 Data is NOT synchronised on all replicas in the table " ~ model) }}
        {% else %}
            {{ log("✅ Data is synchronised on all replicas in the table " ~ model, info=True) }}
            {{ return(model) }}
        {% endif %}
    {% endif %}
{% endmacro %}

По разным причинам репликации данных между нодами ClickHouse может задерживаться. В этом случае мы не хотим продолжать отношения с дагом и бросаем его. Если не хотите сами сообщать родственникам, они могут подписаться на уведомления об ошибках в запусках через планировщик. Мы пришлём письмо с причиной 😘