Change Data Capture

Change Data Capture

Приложение

Приложение

и много пользователей

Приложение

и много пользователей
и много данных в большой БД

Приложение

и много пользователей
и много данных в большой БД
и сложная, запутанная бизнес логика

            один
       пользовательский
           запрос
             |
             ▼
         Приложение 
             |
             |-> 1-10 подзапросов к БД
             |-> 2-3-5 подзапросов к внешнему API
             |-> 1-2 подзапроса на обновление поискового индекса
             |-> тысячи их
        
        

Так ли плохо?

Проблемы

• Подзапросы затягивают время ответа пользователю -> пользователь ждёт -> пользователь грустит
• Подзапросы засоряют код -> программисту дольше и тяжелее читать код -> программист грустит

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    return JsonResponse(article.as_json()) 
        
        

А что там с поиском?

А что там с поиском?

ElasticSearch, Solr, Sphinx, ...?

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    search_engine.update(article)
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    search_engine.update(article)
    mail_sender.send(to=article.author,
                     message=__congrats(article))
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    search_engine.update(article)
    mail_sender.send(to=article.author,
                     message=__congrats(article))
    push_notifications.send(to=Users.object.all(),
                     message=__promote(article))
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    search_engine.update(article)
    mail_sender.send(to=article.author,
                     message=__congrats(article))
    push_notifications.send(to=Users.object.all(),
                     message=__promote(article))
    payment_gateway.charge(article.author,
                           __details(article))
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    __do_some_ancient_magick()
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    __do_some_ancient_magick(
        object=article,
        update_index=YES,
        send_email_notification=YES,
        send_push_notification=YES,
    )
    return JsonResponse(article.as_json()) 
        
        

@api_view(["POST"])
def mark_as_featured(request, id):
    article = Articles.objects.get(id=id)
    article.featured_start_date=now()
    article.featured_end_date=__caclulate(article)
    article.save()
    # the magick is happening somewhere under the hood
    return JsonResponse(article.as_json()) 
        
        

            
            
            
            

         ⇄  Захватчик изменений для поискового индекса
         ⇄  Захватчик изменений для пушей
         ⇄  Захватчик изменений для рассылки писем
        
        

Способы детектирования изменений в базе

• Метаданные (поля last_update/change_date)
• Триггеры на базе
• Транзакционный лог/Журнал изменений

Детектирование изменений через метаданные

Детектирование изменений через метаданные

Детектирование изменений через метаданные

Захват изменений через триггеры

• BEFORE UPDATE
• BEFORE INSERT
• BEFORE DELETE
• AFTER UPDATE
• AFTER INSERT
• AFTER DELETE

Захват изменений через триггеры

            
CREATE TRIGGER log_article_changes_trigger
    BEFORE UPDATE OR INSERT OR DELETE
    ON articles
    FOR EACH ROW
    EXECUTE PROCEDURE log_article_changes();
        

Захват изменений через триггеры

            
CREATE OR REPLACE FUNCTION log_article_changes()
  RETURNS TRIGGER 
  LANGUAGE PLPGSQL
  AS
$$
BEGIN
    IF NEW.featured_start_date <> OLD.featured_start_date THEN
        INSERT INTO articles_featured_log
        (
            article_id, featured_start_date, changed_on
        )
        VALUES
        (
            OLD.id, NEW.featured_start_date, now()
            /* TODO: fix for "AFTER INSERT" cases */
        );
	END IF;
	RETURN NEW;
END;
$$
        

Транзакционный лог
Журнал изменений

Транзакционный лог в случае PostgreSQL

WAL (Write Ahead Log)

WAL PostgreSQL как механизм репликации

Схема использования WAL для репликации PostgreSQL

Схема использования WAL для репликации PostgreSQL

Debezium как вариант реализации CDC для PostgreSQL

Debezium как вариант реализации CDC

Debezium без Kafka (экспериментальный режим!)

Встраивание Debezium в ваше Java приложение

Встраивание Debezium в ваше Java приложение

PgSync: синхронизация PostgreSQL и Elasticsearch

• https://pgsync.com/
• https://github.com/toluaina/pgsync

https://github.com/topics/change-data-capture

Supabase Realtime: изменения из PostgreSQL в вебсокеты

Огромное количество возможностей

Минусы реализации CDC через WAL механизмы

• WAL не бесплатен
• Под CDC резервируются replication slots
• Потенциальное переполнение диска базы (-> падение базы)

Debezium как вариант реализации CDC

Плюсы и минусы CDC

• Инфраструктура и архитектура усложняются
• Тяжело дебажить

Плюсы и минусы CDC

• Непросто "перенакатить" действия в случае выявленных ошибок
• Нужен мониторинг

Плюсы и минусы CDC

• Упрощается кодовая база основного приложения
• Легко добавлять новые обработчики

Плюсы и минусы CDC

• Большая свобода выбора подходящий инструментов под задачу
• Снижается нагрузка на основную БД

Плюсы и минусы CDC

В каких случаях выбирать CDC подход

• Проект Data-центричен
• Приложение уже сильно завязано на базу данных

В каких случаях выбирать CDC подход

• Проект Data-центричен
• Приложение уже сильно завязано на базу данных
• Выполняемые задачи можно сделать идемпотентными

В каких случаях выбирать CDC подход

• Проект Data-центричен
• Приложение уже сильно завязано на базу данных
• Выполняемые задачи можно сделать идемпотентными
• В проекте уже есть ETL/ELT пайплайны и есть нужда ускорить их

В каких случаях выбирать CDC подход

• Проект Data-центричен
• Приложение уже сильно завязано на базу данных
• Выполняемые задачи можно сделать идемпотентными
• В проекте уже есть ETL/ELT пайплайны и есть нужда ускорить их
• В проект внедряется Event-Driven архитектура (см. паттерн Transactional outbox)

Итого

Вопросы?