Data Apps Design

Data Apps Design pinned «✅ Data Engineering in the AI Era: Building a Robust Foundation for AI Applications Слайды доклада: https://drive.google.com/file/d/1U3vm2jZlz9GHlM0dsRhmUwZQzM9rgS8k/view?usp=sharing Запись на YouTube скоро будет 🔸 Основные секции доклада: 🟡 BUSINESS DELIVERABLES…»

🟡 BUSINESS DELIVERABLES AND OUTCOMES

When it comes to AI, the most important things are:

— 📈 Opportunities you can identify (competitive advantages)
— 🛠 The way you can do it (technical capabilities)
— ⏱️ How fast you can ship it (time to market)

Start with business deliverables and outcomes:

— 📊 Summarizing data on dashboards to report key metrics and their dynamics.
— 🤖 Leveraging AI to query DWH data through BI tools using natural language.
— 🔍 Identifying potential PII and notifying for action to ensure compliance.
— 📧 Tailoring communications via CRM platforms based on customer attributes.
— 🎯 Next best offer or recommendation systems for customers.
— 🕒 Creating smart queues and fair order distribution to improve utilization.
— 🚨 Detecting fraudulent orders and transactions at an early stage (near real-time).

🌐 @data_apps | Навигация по каналу

✅ Data Engineering in the AI Era: Building a Robust Foundation for AI Applications

Слайды доклада: https://drive.google.com/file/d/1U3vm2jZlz9GHlM0dsRhmUwZQzM9rgS8k/view?usp=sharing

Запись на YouTube скоро будет 🔸

Основные секции доклада:

🟡 BUSINESS DELIVERABLES AND OUTCOMES

🔵 DATA INGESTION

🟢 DATA TRANSFORMATION

🟤 DATA OPERATIONS


Следующими сообщениями опубликую ключевые идеи доклада


🌐 @data_apps | Навигация по каналу

⛈ Осенний бигдатник от SQL Ninja 30 Ноября (совсем скоро)

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

▶ SQL + dbt = God Mode Data Modeling / Подходы к созданию витрины корпоративных метрик

Много работал за последние месяцы с Kafka, Debezium, real-time streaming, поэтому меня в особенности заинтересовали доклады:

— Введение в потоковую обработку данных с примерами на Apache Flink
— Streamhouse Architecture - единообразный способ работы с пакетными и потоковыми данными
— Опыт эксплуатации Apache Iceberg на петабайтных таблицах. Проблемы и решения
— Debezium — инструмент получения данных с источников, используя паттерн CDC

В этот раз подключусь зрителем онлайн.

Но рекомендую посетить меропритие offline и завести новые знакомства и контакты.

🌐 @data_apps | Навигация по каналу

✅ Data Engineering in the AI Era: Building a Robust Foundation for AI Applications

Готовлю доклад на Data Science Conference Europe.

Поехать лично, к сожалению, не получилось. Но нашелся слот для онлайн-выступления.

Data is the foundational layer for any AI application. In this talk, we will explore ways of building robust data pipelines and employing advanced data modeling techniques:

— Enabling Data Streaming: Transitioning to near real-time and event-driven applications using Debezium and Snowplow.
— Reducing third-party vendor dependencies by leveraging open-source software, resulting in annual savings of $20K.
— Achieving full control over data flow and transformations using dbt and Kafka SMT.
— Optimizing compute and storage costs with Snowflake, achieving a 30% reduction compared to Amazon Redshift.
— Ensuring complete data ownership with zero processing outside the company perimeter.
— Securing data access through RBAC, data masking policies, and PII handling.

This talk distills key knowledge, revealing multiple challenges and pitfalls, and provides practical insights for building a solid data foundation for AI applications.

❤ Дайте реакций, если заинтересованы, и я попробую сделать запись и выложить на YouTube.

🌐 @data_apps | Навигация по каналу

🆕 dbt Power User VS Code Extension + devcontainer

Среди читателей блога очень много пользователей dbt и тех, кто решает задачи моделирования. На днях столкнулся с проблемой и решил поделиться опытом с вами.


🟡 Контекст:

— Я использую devcontainer для каждого отдельного проекта
— В сборку такого контейнера включены необходимые утилиты определенных версий (dbt, adapter for clickhouse/snowflake/redshift, aws cli, obsutils, jq, kcat и т.д.) и их конфигурация (initialization, secrets, etc.)
— Также в контейнер можно включить VS Code Extensions + Common utilities
— Такой контейнер можно запустить локально на своем компьютере и в облаке (это называется Codespaces)
— По сути это похоже на Python virtual envs, только мы не ограничиваемся сугубо Python

Как результат - у всех членов команды одна и та же полноценная среда разработки со всем необходимым инструментарием буквально по одному щелчку мышки. Читайте: никаких проблем с установкой пакетов на разных OS, версиями Python, dependency hell и т.д.

На практике это означает, что, например, Product Analyst или Data Scientist, которого вы наняли вчера, уже сегодня откроет полноценную среду разработки и начнет приносить пользу и создавать ценность.


🩷 Проблема:

— У одного из членов команды проблема с работой dbt Power User Extension
— An error occured while initializing the dbt project: No module named 'agate'
— Суть проблемы сводится к тому, что Extension считает, что dbt не установлен (не найден)


🔵 Решение (скрины будут в комментариях):

— Запуск диагностики: Run diagnostics
— Внимательно изучаю вывод в своем окружении и у человека, который испытывает затруднения
— Суть в том, что у нас отличается Python paths
— Setup Extension: Select Python Interpreter
— Вуаля! Проблема решена. Extension работает корректно


🟤 А зачем всё это?

Теперь несколько слов о самом dbt Power User Extension. Самые востребованные лично для меня возможности:

— Ориентация по DAG: parent / children nodes / model tests and docs
— Горячие клавиши почти на все действия dbt. В основном я пользуюсь для просмотра compiled / run кода и отладки
— Автогенерация .yml-файлов (sources, models, columns, tests)
— Просмотр документации проекта + Lineage

О других возможностях узнать подробнее можно в оф. документации: https://docs.myaltimate.com/setup/installation/


💬 Задавайте вопросы, если есть.

Happy modeling ☺️

🌐 @data_apps | Навигация по каналу

✅ Data Streaming Troubleshooting: Kafka Connect / Debezium / Array Encoding / GeoJSON / Dead Letter Queue / Incremental snapshots

Приготовьтесь, потому что это будет сложно, но мне необходимо поделиться с вами этим 🙂


🟢 Контекст (дано):

— MongoDB как источник (полуструктурированные данные с гибкой схемой)

— Debezium as Source Connector для репликации лога в Kafka

— MongoDB New Document State Extraction SMT для преобразования сообщений Debezium и загрузки их в DWH

— Конфигурация array.encoding=document для репликации массивов

— Dead Letter Queue (DLQ) для сбора сообщений с ошибками репликации

— Geo Data: коллекция в MongoDB содежащая георгафические зоны в формате GeoJSON


🔴 Проблематика:

— Специфика Array encoding для MongoDB

By default, the event flattening SMT converts MongoDB arrays into arrays that are compatible with Apache Kafka Connect, or Apache Avro schemas. While MongoDB arrays can contain multiple types of elements, all elements in a Kafka array must be of the same type.

— Но без конфигурации array.encoding=document было бы пропущено по оценкам 50-60% всех реплицируемых данных (недопустимо!)

— Чтобы прочесть валидный GeoJSON в СУБД TRY_TO_GEOGRAPHY(geojson) (в тип GEOGRPAHY) мне нужно реплицировать именно массивы!

— Однако и в этом случае теряется несколько записей. Из DLQ я вижу, что это за строки и описание причины:

Field coordinates of schema kafka_mongodbt_connector__geo_zones.geometry.features.geometry is not a homogenous array.
Check option 'struct' of parameter 'array.encoding'

🔵 Решение:

— С помощью Kafka Connect predicates устанавливаем array.encoding=array для избранных коллекций; Для всех остальных коллекций - array.encoding=document

— Анализируем проблемные записи в Dead Letter Queue. Выясняется, что GeoJSON содержит дополнительный ключ features, в котором дублируются координаты из 3-х чисел, 2 из которых float, а третье всегда 0. И с вероятностью 99% это вызывает ошибку. Необходимо либо проставить 0.0, либо вообще убрать ключ geometry.features из репликации.

— Исключаем nested field из репликации на уровне Source Connector: "field.exclude.list": "db.geo.geometry.features"

— Обновляем конфиг работающего коннектора через REST API: jq .config ${SOURCE__MONGODB} | http PUT ${SOURCE__MONGODB}/config

— Инициируем Incremental Snapshot для коллекции с geo через signalling kafka topic

✅ Вуаля! Все записи доступны. Все GeoJson атрибуты корректны.


🟤 Ключевые выводы:

— Допустимо использовать различные варианты конфига array.encoding (document / array) для разных таблиц (коллекций) БД с помощью Kafka Connect predicates

— Без топика Dead Letter Queue (доступен только для Sink Connectors!) не узнал бы об ошибках репликации и их причинах

— Dead Letter Queue удобно просматривать там же в DWH (SELECT * FROM DLQ), настраивать тесты и уведомления (поэтому для DLQ нужен отдельный Sink Connector!)

— Фича Incremental Snapshot помогает прочесть и реплицировать данные по запросу (не останавливая streaming - чтение лога БД / OpLog!)


💬 Задавайте вопросы / оставляйте комментарии.

❤ Кстати, есть тут те, кто тоже хочет real time data streaming? Для целей аналитики, ML, Anti-fraud или прочего оперативного реагирования?


🌐 @data_apps | Навигация по каналу

⚡ Data Engineering in the AI Era: Building a Robust Foundation for AI Applications

Хочу выступить с таким докладом в Сербии на datasciconference.com


Data is the foundational layer for any AI application. In this talk, we will explore Wheely's evolution in building robust data pipelines and employing advanced data modeling techniques:

— Enabling Data Streaming: Transitioning to near real-time and event-driven applications using Debezium and Snowplow.

— Reducing third-party vendor dependencies by leveraging open-source software, resulting in annual savings of $20K.

— Achieving full control over data flow and transformations using dbt and Kafka SMT.

— Optimizing compute and storage costs with Snowflake, achieving a 30% reduction compared to Amazon Redshift.

— Ensuring complete data ownership with zero processing outside the company perimeter.

— Securing data access through RBAC, data masking policies, and PII handling.

This talk distills key knowledge, revealing multiple challenges and pitfalls, and provides practical insights for building a solid data foundation for AI applications.

Заявку подал. До конференции остался месяц.

Если не там, то выступлю где-то еще 🙂

💬 Какие еще ИТ-конференции знаете и рекомендуете?

❄️ Snowflake will require Multi-Factor Authentication and Stronger Password Policies by default

— Наконец-то получил письмо с уведомлением от Snowflake
— До этого неоднократно писал в поддержку и Sales rep в призыве к этому
— Ранее нашел способ включить политику MFA принудительно (ниже)
— В качестве MFA используется только DUO APP (from Cisco), Google Auth нельзя

🩷 Мой подход:

---------------------------
-- AUTHENTICATION POLICY --
---------------------------

CREATE AUTHENTICATION POLICY IF NOT EXISTS AUTHENTICATION_POLICY_REQUIRE_2FA
  AUTHENTICATION_METHODS = ('ALL')
  MFA_AUTHENTICATION_METHODS = ('PASSWORD', 'SAML')
  MFA_ENROLLMENT = REQUIRED
  CLIENT_TYPES = ('ALL')
  SECURITY_INTEGRATIONS = ('ALL')
  COMMENT = 'Enforce 2FA with DUO APP'
;

CREATE AUTHENTICATION POLICY IF NOT EXISTS AUTHENTICATION_POLICY_NONUSER_LOGIN
  AUTHENTICATION_METHODS = ('PASSWORD', 'KEYPAIR')
  CLIENT_TYPES = ('DRIVERS')
  COMMENT = 'Non-user logins are only allowed to authenticate through drivers via password or keypair'
;

ALTER USER A_KOZYR SET AUTHENTICATION POLICY AUTHENTICATION_POLICY_REQUIRE_2FA ;

ALTER USER KAFKA_CONNECTOR SET AUTHENTICATION POLICY AUTHENTICATION_POLICY_NONUSER_LOGIN ;

🔵 Самые важные отрывки из письма:

What you need to know:

This new MFA policy only applies to Snowflake users with the TYPE=PERSON or unset type (i.e., human users) that use a Snowflake built-in password to login. Please see the documentation for more information on user types and see the Timeline and Implementation below for additional details on how these policies will be applied.

For users created after a new account is bootstrapped

Make sure to mark the TYPE for all users
For human users TYPE=PERSON, follow the DUO enrollment process and start using MFA
For service users TYPE=SERVICE, make sure to either: 1) use external OAuth (preferred), or 2) use key pair authentication combined with network policy.
If a service user cannot leverage a key pair, mark them as TYPE= LEGACY_SERVICE. This is a temporary solution and we highly recommend that you fix your deployment based on the above.

✅ Считаю, что MFA должен быть включен по дефолту при работе с а) любыми активами в виде данных и б) при работе с облаками.

🌐 @data_apps | Навигация по каналу

🔹 TEMPLATED QUERY TO DWH FROM BI или ужасный недостаток Superset

Чего хочу добиться:

— Есть витрина фактов: поездки / заказы, которые привязаны к карте (ГЕО)
— Хочу подсчитать метрику в привязке к точке (в радиусе) или зоне (GeoJSON, etc.) на карте
— Динамику метрики отследить за периоды времени
— Предоставить пользователям возможность интерактивно выбирать: точку на карте, радиус, временной период

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

Прототип запроса:

SELECT 
  order_id
  , create_ts
  , order_value
FROM analytics.f_orders AS o
WHERE TRUE
  AND create_ts >= {{ time_filter.from_expr }}
  AND create_ts < {{ time_filter.to_expr }}
  AND status IN {{ filter_values('status') | where_in }}
  AND greatCircleDistance(lon, lat, {{ longitude }}, {{ latitude }}) <= {{ radius }}
;

Пробую сделать это в Apache Superset

Прочел SQL Templating, просмотрел Preset Live Demo - Jinja Templating

— Вношу изменения в конфиг Superset: ENABLE_TEMPLATE_PROCESSING: true, перезапускаю контейнеры. Без этого вообще дальше что-то пробовать бессмысленно
— Пишу запрос в SQL Lab. Вписываю Jinja. Запрос отработает только если вписать дефолтные значения в меню Parameters
— Пытаюсь сохранить Virtual Dataset - получаю ошибку. Ок, иду на хитрость. Сохраняю без параметров. И далее редактирую в меню Edit Dataset
— И здесь опять вписываю дефолтные значения параметров, чтобы можно было создать Chart
— Пробую нанести Chart на Dashboard и использовать фильтры. Успеха нет. К сожалению, здесь мое терпение кончилось

☹️😑 Очень много шагов, много ошибок и действий к их устранению. Я оставляю эту идею.

Эта же задача решается крайне легко в Metabase - SQL parameters

Причем с продвинутым функционалом, документацией и примерами.

— Setting (Optional) SQL parameters
— Add a parameter to the URL and load the page.
— SQL variable types: Text, Number, Date, etc.
— Setting Default values

Вывод: В Superset это сделать Almost Impossible, в Metabase - крайне легко.

В тему сравнения плюсов и минусов и критериев выбора инструмента для BI-аналитики.

🌐 @data_apps | Навигация по каналу

🆕 ОБНОВЛЕНИЯ ЗА Q2-Q3 2024

ПЛАНЫ

— Kubernetes Orchestration для сервисов и приложений
— Snowflake Dynamic tables для ускорения поставки данных в витрины и BI
— dbt code base refactoring: Optimize performance, Incorporate latest features, Enhance CI checks
— Snowflake SSO (Google) + Network restrictions
— Prefect / Dagster для Data Integration Pipelines (API Yandex.Metrika / Yandex.Direct)
— Alternative BI tools : Sigma, Thoughtspot, Superset

Обучающие сессии и запуски курсов

— Designing Modern Data Apps
— Хардкорный курс по интеграции данных (фокус на Kafka, Debezium, real time streaming)

Consulting + решения для бизнесов

— Аналитическая инфраструктура: развертывание в облако, управление, мониторинг, обновления
— Интеграция данных
— Аналитическая СУБД
— Инструменты визуализации, экспериментов, ML
— Цифровизация бизнеса: ключевые метрики, паттерны и инициативы доступны владельцам и менеджменту

💬 Если есть подказки, вопросы или хочется узнать о чем-то подробнее - дайте знать в комментариях.

🌐 @data_apps | Навигация по каналу

🆕 ОБНОВЛЕНИЯ ЗА Q2-Q3 2024

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


SNOWFLAKE

— Data Masking Policies для маскирования определенных колонок (ГЕО, хеш-суммы, sensitive data) и даже ключей в JSON (VARIANT)
— Snowpipe - автоматическая загрузка новых файлов из S3 в таблицы Snowflake
— Snowflake 2FA authenticatication policy - установил как обязательную политику для всех пользователей
— Table clustering - включил кластеризацию по ключу там, где это дает наибольший буст производительности

В целом, переход на Snowflake - это одно из лучших решений 2024. Об этом я писал ранее: раз, два, три, и напишу еще.


DATA INTEGRATION PIPELINES

— Выбрал стек KAFKA + DEBEZIUM
— Source connectors: MongoDB, PostgreSQL, MySQL, JDBC
— Sink Connectors: Snowflake, Clickhouse
— Для удобства развернул и использую Kafka UI
— Operations: Ad-hoc Incremental snapshots, signalling - в любой момент я могу перезалить таблицу (целиком или с учетом фильтра)
— Monitoring metrics Prometheus + Grafana. Вижу коннекторы, что работает, сколько событий приходит, как долго работают snapshots
— В итоге у меня Near real time data streaming - все изменения почти сразу оказываются в Snowflake
— Schematization + schema evolution: таблицы в Snowflake создаются и меняются автоматически
— Относительно легкий процесс Connector versions upgrade (JAR-файлы + Docker)
— PII handling on EL step: Exclude columns или Java SMT для преобразований (хеш-суммы, фильтры, условные NULLIFY) на лету
— SaaS Hevo offboarding: данные не утекают к SaaS, экономим $20K / year на стоимости сервиса https://t.me/data_apps/394


dbt (T = data transformations)

— Изменения, связанные с переходом на Debezium: log deduplication (QUALIFY)
— Адаптация к работе с ARRAYS которые пишет Debezium (Array Encoding)
— Incremental builds + frequent updates - востребованные витрины обновляются каждый час
— 4 years time limit - в целом во всем DWH принято решение ограничиться 4-мя годами истории


SNOWPLOW: Support production-grade event tracking (App / Web)

— Развернул сервисы: Collector, Enrich, Iglu (Schemas), Snowflake Loader
— Промежуточное хранилище событий (persist with Kafka) - поддержка multiple producers/consumers, at least once semantics, events replay (change offsets)
— Only accept verified events - avoid invalid events / flood / spam / unauthorized producers reaching endpoints
— Schema validation - support schema evolution, handle different producers (versions and schemas)
— Data enrichments
— События приходят в Snowflake в режиме реального времени (отставание <1 минуты)


YANDEX METRIKA + YANDEX DIRECT

— Создал и зарегистрировал приложение для получения OAuth-токенов для работы с API
— Написал bash-скрипты для выгрузки данных
— Поддержка исторической (произвольный период) и регулярной выгрузки (3 последних дня)
— Укладка в S3 в виде TSV-файлов на каждый день (с возможностью перезаписи) - Datalake approach
— С последующим чтением из Clickhouse (S3 table engine / table function)
— Автоматизировал выгрузки в Github Actions

⛈ Дамы и господа, всем доброго вечера!


За последние несколько дней я сделал невероятное (для себя):

— Изучил различные подходы к трансформации данных в Kafka: SMT / ksql / kstreams / flink / spark streaming / apache beam / Faust

— Собрал Devcontainer для Java 21 + Maven

— Изучил десятки примеров кода создания SMT + Unit tests

— Научился работать с Maven: properties, dependencies, build

— Создал код для собственных Single Message Transformations + Unit tests

— Собрал JAR-файлы и отправлял их в Kafka Connect docker image

— Потратил уйму часов на Debug, изучение логов, поиск ошибок и исправления на реальных данных в Kafka Connect cluster

— И в конечном счете достиг искомого результата ✅


🐹 В процессе я прибегал к помощи Copilot, и это классный опыт! Заслуживает отдельного обзора.


В итоге, вырисовывается такая картинка общим планом:

— Все БД-источники данных подключены через CDC (Kafka + Debezium)
— Near real time sync (отставание до 1 минуты)
— PII чистится на лету, никогда не попадая в Snowflake 🫢
— 97% cost saved (20% overall spend) в сравнении с предыдущим EL tool (Hevo)
— Это смешно, но Snowpipe streaming 24 / 7 стоит копейки (10$ в неделю 🤯)
— Открываются невероятные возможности создания онлайн-приложений: fraud, scoring, marketplace, reporting

❤ Как только я успею к дедлайну и закончу основую разработку, я примусь за обещанные публикации, обзоры, мнения.

🌐 @data_apps | Навигация по каналу

✔️ Мой чеклист и рекомендации по документированию аналитических приложений с помощью dbt Docs


🟢 Источники данных - заполните максимум для себя и тех кто будет изучать документацию

Примерный список атрибутов:

— Source (Mobile Apps Events)
— Type (JSON files)
— Way of data integration (S3 table engine)
— Description (Events tracked in Mobile Apps, single file for each event)

🟡Структурирование проекта на слои преобразований

Мой пример:
— Sources
— Staging
— Intermediate
— Analytics (dims / facts)

🔵Цветовая схема на графе dbt Docs

Ноды dbt (включая модели) можно окрасить в разные цвета на графе:

    staging:
      +docs:
        node_color: "#219EBC"

🔴Скрыть нерелевантные модели из документации

models:
  <resource-path>:
    +docs:
      show: true | false

🟤Блоки {{ doc() }}: Переиспользование блоков документации

В целом, большие блоки документации стоит вынести в отдельные markdown-файлы. Возможно, с картинками и схеми.

Также имеет смысл один раз описать атрибут (колонку), который присутствует в 1+ моделях, и впоследствие ссылаться на его описание:

models:
  - name: events
    description: '{{ doc("table_events") }}'

🩷Custom overview page - главная страница сайта с документацией

Рекомендую грамотно использовать title page и разместить на нем самую важную информацию:

— Используемые инструменты и сервисы
— Архитектурная схема
— Описание источников данных
— Ссылки на JIRA (task trackers), Wiki pages, Operational systems, Contacts (@), etc.

{% docs __overview__ %}

## OVERVIEW
## INFRASTRUCTURE SCHEMA
## TOOLS USED
## DATA SOURCES
## USEFUL LINKS

{% enddocs %}

🟢Публикация вебсайта

Т.к. сайт статический, его достаточно разместить в Object Storage с поддержкой Static website hosting:

— Host on Amazon S3 (optionally with IP access restrictions)
— Publish with Netlify
— Use your own web server like Apache/Nginx

🟡Актуализация вебсайта с документацией - всегда Up to Date

Чтобы вебсайт всегда отражал актуальное состояние, необходимо после каждого изменения в код (Merge Request), генерировать и загружать новую версию.

Делать это можно с помощью Github Actions или аналогов, автоматизируя действия:

— dbt docs generate
— aws s3 sync ./target s3://my-website-bucket/

🌐 @data_apps | Навигация по каналу

👑 Серию замечательных статей об Аналитике и Данных с фэнтези-аллегориями подготовил Алексей Сидоров из myBI Connect

🟤 Данные - путешествие Туда и обратно
🟤 Данные - путешествие Туда и обратно. Часть вторая.

Рекомендую для структурирования и систематизации знаний:

— Источники данных
— Выгрузки и обращения по API
— Основы моделирования
— Витрины данных и визуализации
— Семантический слой
— There and back again 😏

Всё это с аллегориями и захватывающими иллюстрациями.

Всем, кто любит Средиземье и вселенную LoTR это особенно понравится.

🌐 @data_apps | Навигация по каналу

🔵 ОСТРЫЕ ТЕМЫ: SALARY, PERFORMANCE REVIEW, ДЕГРАДАЦИЯ КУЛЬТУРЫ РАЗРАБОТКИ, ПОЗИЦИОНИРОВАНИЕ В ОТНОШЕНИЯХ С EMPLOYER

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

Есть мнение, что существуют гораздо более актуальные темы, которые читатели хотели бы услышать и обсудить. Среди тех, на которые я хотел бы высказаться:


🔢 Почему компании со скрипом повышают SALARY даже на 10% даже в том случае, если вы экономите (ЗАРАБАТЫВАЕТЕ) для нее $50K+ в год

🔢 Мое мнение насчет проведения PERFORMANCE REVIEW в компании и почему вы с высокой вероятностью получите "Ты молодец! Продолжай в том же духе!"

🔢 Что дает человеку чувство собственной власти и независимости от EMPLOYER, а равно и спокойствие, уверенность, и высокую производительность

🔢 Как дать понять МЕНЕДЖЕРУ и всем СТЕЙКХОЛДЕРАМ, что вы занимаетесь ключевой и важной работой, без должного выполнения которой их деятельность будет затруднена (НЕВОЗМОЖНА)

🔢 Почему даже в СТАРТАПАХ деградирует культура разработки и поставки качественных решений (кода), а стиль управления оставляет желать лучшего


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

🌐 @data_apps | Навигация по каналу

⭐ Docker: перенести Volumes в другую Filesystem без потери данных

🔹 Ситуация

Вы создали VM, разместили на ней свои сервисы в Docker, запустили, сервисы (например, Kafka + Kafka Connect) начали писать данные.

Вдруг вы обнаруживаете, что disk volume заканчивается.

df -h

Что делать? Как перенести Docker Volumes и избежать ошибки Disk full?


🔺 Действия

— Добавить новый Filesystem, на котором будет достаточно места
— Остановить контейнеры
— Отключить Docker services
— Unmount docker overlay filesystems
— Перенести данные в новый раздел
— Внести изменения в конфигурационный файл Docker daemon.json
— Включить Docker services
— Запустить контейнеры


🟡 Скрипт

# Stop Docker containers
docker compose stop

# Stop Docker services
systemctl stop docker
systemctl stop docker.socket

# Unmount docker filesystems
umount /var/lib/docker/overlay2/*/*

# Move data
mv /var/lib/docker/ /mnt/${DESTINATION}/

# Edit Docker service configuration file:
nano /etc/docker/daemon.json

# Add or modify the data-root configuration option to point to the new directory:
{
  "data-root": "/mnt/path/docker"
}
# Save and close the file.


# Start the Docker service:
systemctl start docker
systemctl start docker.socket

# Start Docker containers
docker compose start

Также этот скрипт на Github Gist: https://gist.github.com/kzzzr/fef4c9c88f301c282275e6d451e080f5

Это сработало для меня, и должно сработать для вас.

В частности, это помогло мне расширить дисковое пространство на VM, и позволило Docker писать в новую Filesystem БЕЗ повторного запуска full sync истории по 100+ коллекциям MongoDB (Debezium Connector + Kafka topics).


🌐 @data_apps | Навигация по каналу

🔹 Хочу устроить Real Time Data Sync — дайте критику и комментарии

Привет! Последние пару недель плотно работаю над настройкой Data Pipelines через Debezium + Kafka Connect.

🔺 Общая схема будет выглядеть так:

— Sources: MongoDB, Postgres
— Source connectors: Kafka Connect + Debezium (CDC)
— Kafka as intermediate storage
— Sink connector: Snowflake connector for Kafka (Snowpipe streaming)
— Snowflake as destination

Упражнения, которые я проделываю:

🟡 Deploy services: Kafka + Kafka Connect (+ Zookeeper + Schema registry + REST Proxy)

Вообще, готовые образы есть от: Confluent / Debezium / Strimzi.

Я пока использую образы от Debezium + Strimzi. Модифицирую, добавляя нужные мне коннекторы указанных версий (JAR-файлы).

У Strimzi есть операторы для K8s, которые я планирую использовать.

А вот K8s deployment от Confluent входит в Enterprise plan и подлежит лицензированию.

🔵 Choose collections for replication

Выбрал несколько несложных коллекций для тестирования и дебага.

Несложные, потому что:

— Небольшое количество колонок
— Минимальное количество вложеных nested-структур
— При этом частые обновления - получаю лог изменений CDC сразу.
— Глазами можно отследить все применяемые изменения (либо то, что не получается)

🟢 Source connector (MongoDB)

На выбор у меня было 2 Connectors:

— MongoDB Kafka Connector
— Debezium connector for MongoDB

Несмотря на наличие отличной документации и даже поддержку snapshot с учетом фильтра (например, sync истории только за 1 год), выбор сделан в пользу Debezium Connector. Позже подробно напишу доступные возможности и пример конфигурации.

🩷 Metadata fields for events

В каждое событие дополнительно пишу ряд метаданных:

— ts_ms = metadata timestamp
— op = c (create), u (update), d (delete), r (snapshot)

🔴 Message transformations: Topic route, New Document State Extraction

— Topic route: Мессаджи направляю сразу в топики с целевыми названиями таблиц в Snowflake.

— New Document State Extraction: этат трансформация позволяет перейти от сложного и детального формата события Debezium (schema, before, after) к упрощенному виду (только after), который мне и нужен в Snowflake.

🟤 Debug events

Развернул devcontainer (docker-compose based), установил kcat + jq.

Использую kcat для просмотра топиков, key-value значений, оффсетов.

jq использую для форматирования ответов от REST API (Kafka Connect).

Очень удобно проконтролировать изменения, которые вношу пунктом ранее.


🟡В следующем посте:

— Externalizing secrets
— Create a sink connector (Snowflake)
— Set up monitoring (Prometheus + Grafana)
— Use AVRO data format (io.confluent.connect.avro.AvroConverter)
— Configure topics (compact + delete, etc.)
— Configure incrmental snapshots loads via Debezium signals
— Ensure PII data masking (message transformations, connector configuration)
— Add HTTP bridge / REST proxy

🔻 Могли бы покритиковать / поделиться опытом / посоветовать что-либо?

🌐 @data_apps | Навигация по каналу

🔰 Быстро и просто устроить ENVs для Data: DEV, TEST, PROD (part 2)

✅ Как это сделать

— Проще всего использовать единую СУБД и разделить ее на набор databases / schemas
— Среда - это лишь совокупность connection details + role + target schema
— В dbt это называется targets и задается в файле profiles.yml
— В зависимости от используемой среды вы хотите модифицировать тот или иной код

✅ Давай примеры

🩷 profiles.yml для dbt с 3-мя средами.

Вместо деталей для подключения используются переменные среды, которые задаются:

— Локально на компьютере пользователя в DEV
— В CI-инструменте (Github Actions) для TEST
— В Orchestration tool (dbtCloud) для PROD

Полностью по ссылке: https://gist.github.com/kzzzr/02d4f1f86d19df9c90210b177a561406

🩷 Limit rows for DEV and TEST with Jinja SQL:

    {#- prepare expression to limit rows depending on target.name and flags provided -#}
    {%- if target.name in ['prod'] and history_depth_days > 0 -%}
        {%- set watermark = 'DATEADD(days, ' ~ -1 * history_depth_days ~ ', CURRENT_TIMESTAMP::DATE)' -%}
        {%- set history_depth_expression = timestamp_column ~ ' >= ' ~ watermark -%}
    {%- elif target.name not in ['prod'] and limit_data_for_dev == true -%}
        {%- set watermark = 'DATEADD(days, ' ~ -1 * var('data_tests_depth_days') ~ ', CURRENT_TIMESTAMP::DATE)' -%}
        {%- set history_depth_expression = timestamp_column ~ ' >= ' ~ watermark -%}
    {%- else -%}
        {%- set history_depth_expression = '1 = 1' -%}
    {%- endif -%}

    {#- prepare final filter expression -#}
    where 1 = 1
        and {{ deleted_rows_expression }}
        and {{ history_depth_expression }}

🩷 Уровень серьезности (Severity) теста: ERROR для PROD, WARN для DEV:

        - name: timezone
          tests:
              - dbt_utils.not_null_proportion:
                  at_least: 0.95
                  severity: "{{ 'error' if target.name in ('prod') else 'warn' }}"

🩷 Используемый COMPUTE в зависимости от среды:

{{
    config(
        materialized='table',
        snowflake_warehouse='WH_LARGE' if target.name == 'prod' else target.warehouse
    )
}}

🩷 Запись артефактов dbt run в метаданные (только для PROD):

on-run-end:
    - "{% if target.name == 'prod' %}{{ dbt_artifacts.upload_results(results) }}{% endif %}"

✅ Какие преимущества

— Все пользуются одними и теми же исходными наборами данных (data sources) - исключены расхождения и несогласованность

— Zero effort - не нужно создавать среды как отдельные инстансы, разворачивать ПО, копировать датасеты и следить за актуальностью

— Один проект = один репозиторий с кодом ✔

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

— Это экономически выгодно и целесообразно. Используя Snowflake вообще perfect, т.к. у вас decoupled storage & compute, при этом compute unlimited

— Это быстро и безболезненно с точки зрения поставки кода и Time2Market

❤ Расскажите, как эти процессы реализуете вы? В чем недостатки и преимущества подхода?

🌐 @data_apps | Навигация по каналу