{
    "version": "https:\/\/jsonfeed.org\/version\/1.1",
    "title": "Yuriy Gavrilov: posts tagged Lakehouse",
    "_rss_description": "Welcome to my personal place for love, peace and happiness 🤖 Yuiry Gavrilov",
    "_rss_language": "en",
    "_itunes_email": "yvgavrilov@gmail.com",
    "_itunes_categories_xml": "",
    "_itunes_image": "https:\/\/gavrilov.info\/pictures\/userpic\/userpic-square@2x.jpg?1643451008",
    "_itunes_explicit": "no",
    "home_page_url": "https:\/\/gavrilov.info\/tags\/lakehouse\/",
    "feed_url": "https:\/\/gavrilov.info\/tags\/lakehouse\/json\/",
    "icon": "https:\/\/gavrilov.info\/pictures\/userpic\/userpic@2x.jpg?1643451008",
    "authors": [
        {
            "name": "Yuriy Gavrilov - B[u]g - for charity.gavrilov.eth",
            "url": "https:\/\/gavrilov.info\/",
            "avatar": "https:\/\/gavrilov.info\/pictures\/userpic\/userpic@2x.jpg?1643451008"
        }
    ],
    "items": [
        {
            "id": "278",
            "url": "https:\/\/gavrilov.info\/all\/sozdaem-streaming-lakehouse-za-chas-rukovodstvo-po-risingwave-la\/",
            "title": "Создаем Streaming Lakehouse за час: руководство по RisingWave, Lakekeeper и Trino",
            "content_html": "<p>Вы когда-нибудь мечтали о платформе, где данные, отправленные через простой API-вызов, через секунды становятся доступны для аналитических запросов в вашем озере данных? Мечты сбываются. Эта статья — подробное, основанное на реальном опыте руководство, которое покажет, как построить современный Streaming Lakehouse с нуля.<\/p>\n<p>Доки, которые пригодились:<\/p>\n<p><a href=\"https:\/\/github.com\/risingwavelabs\/risingwave\/blob\/main\/docker\/docker-compose.yml\">https:\/\/github.com\/risingwavelabs\/risingwave\/blob\/main\/docker\/docker-compose.yml<\/a><br \/>\n<a href=\"https:\/\/github.com\/lakekeeper\/lakekeeper\/blob\/main\/examples\/minimal\/docker-compose.yaml\">https:\/\/github.com\/lakekeeper\/lakekeeper\/blob\/main\/examples\/minimal\/docker-compose.yaml<\/a><br \/>\n<a href=\"https:\/\/docs.risingwave.com\/iceberg\/deliver-to-iceberg#rest-catalog\">https:\/\/docs.risingwave.com\/iceberg\/deliver-to-iceberg#rest-catalog<\/a><\/p>\n<p>Наши главные герои:<\/p>\n<ul>\n<li><b>RisingWave<\/b>: Потоковая база данных, “сердце” нашего пайплайна. Она будет принимать, преобразовывать и материализовывать данные на лету.<\/li>\n<\/ul>\n<div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/image-212.png-1.jpg\" width=\"2560\" height=\"868\" alt=\"\" \/>\n<\/div>\n<ul>\n<li><b>Lakekeeper<\/b>: Современный REST-каталог для Apache Iceberg. Наш “библиотекарь”, который знает все о структуре данных в озере.<\/li>\n<\/ul>\n<div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.37.28.png\" width=\"1046\" height=\"270\" alt=\"\" \/>\n<\/div>\n<ul>\n<li><b>Trino<\/b>: Мощный движок для федеративных запросов. Наше “окно” в озеро данных для выполнения ad-hoc аналитики.<\/li>\n<\/ul>\n<div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.39.34.png\" width=\"958\" height=\"446\" alt=\"\" \/>\n<\/div>\n<p>Мы пройдем весь путь: от сравнения технологий и настройки окружения до отправки данных и любования результатами на дашбордах Grafana. И самое главное — мы поделимся всеми “граблями”, на которые наступили, чтобы вы могли их обойти.<\/p>\n<h3>Глава 1: Почему RisingWave? Взгляд на альтернативы<\/h3>\n<p>На рынке потоковой обработки есть много инструментов, но все они предлагают разные подходы. Почему для нашей задачи мы выбрали именно RisingWave?<\/p>\n<p><b>RisingWave<\/b> — это распределенная потоковая база данных, созданная для упрощения обработки данных в реальном времени. Ее ключевая особенность — использование материализованных представлений поверх потоков данных. Вы пишете знакомый SQL, а RisingWave берет на себя всю сложную работу по инкрементальному обновлению результатов с минимальной задержкой.<\/p>\n<p>Давайте сравним его с популярными альтернативами.<\/p>\n<h4>Сравнительная таблица<\/h4>\n<table cellpadding=\"0\" cellspacing=\"0\" border=\"0\" class=\"e2-text-table\">\n<tr>\n<td style=\"text-align: center\">Критерий<\/td>\n<td style=\"text-align: center\">RisingWave<\/td>\n<td style=\"text-align: center\">Связка Debezium + Flink<\/td>\n<td style=\"text-align: center\">Apache SeaTunnel<\/td>\n<\/tr>\n<tr>\n<td style=\"text-align: center\"><b>Архитектура<\/b><\/td>\n<td style=\"text-align: center\">Единая система: хранение состояния (state) и вычисления в одном продукте.<\/td>\n<td style=\"text-align: center\">Компонентная: Debezium (CDC), Kafka (очередь), Flink (обработка), отдельное хранилище состояния.<\/td>\n<td style=\"text-align: center\">Инструмент для перемещения данных (data mover) с коннекторами.<\/td>\n<\/tr>\n<tr>\n<td style=\"text-align: right\"><b>Основная задача<\/b><\/td>\n<td style=\"text-align: center\">Создание и поддержка инкрементально обновляемых материализованных представлений.<\/td>\n<td style=\"text-align: center\">Гибкая, низкоуровневая обработка потоков общего назначения.<\/td>\n<td style=\"text-align: center\">Пакетная и потоковая синхронизация данных между разнородными источниками и приемниками.<\/td>\n<\/tr>\n<tr>\n<td style=\"text-align: center\"><b>Простота использования<\/b><\/td>\n<td style=\"text-align: center\"><b>Очень высокая.<\/b> Знание SQL — это 90% успеха. Скрывает сложность управления состоянием.<\/td>\n<td style=\"text-align: center\">Низкая. Требует экспертизы в каждом компоненте, написания кода на Java\/Scala, управления состоянием.<\/td>\n<td style=\"text-align: center\">Средняя. Конфигурация через файлы, но требует понимания особенностей каждого коннектора.<\/td>\n<\/tr>\n<tr>\n<td style=\"text-align: right\"><b>Обработка данных<\/b><\/td>\n<td style=\"text-align: center\">SQL-ориентированная. `CREATE MATERIALIZED VIEW ... AS SELECT ...`.<\/td>\n<td style=\"text-align: center\">Программная. DataStream API, Table API\/SQL. Позволяет писать сложную бизнес-логику.<\/td>\n<td style=\"text-align: center\">Декларативная. Определяет `source`, `transform`, `sink`. Менее гибкая для сложных трансформаций.<\/td>\n<\/tr>\n<tr>\n<td style=\"text-align: center\"><b>Поддержка SQL<\/b><\/td>\n<td style=\"text-align: center\"><b>Первоклассная.<\/b> Совместимость с PostgreSQL на уровне синтаксиса и протокола.<\/td>\n<td style=\"text-align: center\">Хорошая (Flink SQL), но не является основным интерфейсом.<\/td>\n<td style=\"text-align: center\">Ограниченная. Используется для простых трансформаций, а не для определения логики потока.<\/td>\n<\/tr>\n<tr>\n<td style=\"text-align: center\"><b>Управление состоянием<\/b><\/td>\n<td style=\"text-align: center\"><b>Встроенное и автоматическое.<\/b> Использует облачное хранилище (S3) как персистентный слой.<\/td>\n<td style=\"text-align: center\"><b>Ручное.<\/b> Требуется настраивать и управлять чекпоинтами и состоянием (например, RocksDB).<\/td>\n<td style=\"text-align: center\">Зависит от движка (Flink\/Spark). Не является основной функцией самого SeaTunnel.<\/td>\n<\/tr>\n<\/table>\n<p><b>Выводы:<\/b><\/p>\n<ul>\n<li><b>Связка Debezium + Flink<\/b> — это невероятно мощный, но сложный “конструктор”. Он идеален для компаний с большими командами инженеров данных, которым нужна максимальная гибкость для создания кастомной логики.<\/li>\n<li><b>Apache SeaTunnel<\/b> — это отличный “швейцарский нож” для перемещения данных. Его сила — в огромном количестве коннекторов. Он идеален для задач ETL\/ELT, когда нужно перелить данные из точки А в точку Б с минимальными трансформациями.<\/li>\n<li><b>RisingWave<\/b> занимает золотую середину для аналитических задач в реальном времени. Он предлагает простоту и элегантность SQL, скрывая под капотом всю сложность потоковой обработки. Если ваша цель — быстро получить свежие аналитические витрины из потоков данных, RisingWave — ваш выбор.<\/li>\n<\/ul>\n<h3>Глава 2: “Кексы” — фишки RisingWave, которые упрощают жизнь 🍰<\/h3>\n<p>Что делает RisingWave таким привлекательным на практике?<\/p>\n<ol start=\"1\">\n<li><b>PostgreSQL-совместимость:<\/b> Вы можете подключиться к RisingWave любым клиентом, который “говорит” на протоколе Postgres (например, DBeaver, psql). Синтаксис SQL для создания представлений и запросов вам уже знаком.<\/li>\n<li><b>Все-в-одном для стриминга:<\/b> RisingWave объединяет в себе прием данных (коннекторы), их обработку (инкрементальные вычисления) и хранение состояния. Вам не нужно разворачивать и связывать вместе Kafka, Zookeeper, Flink и RocksDB.<\/li>\n<li><b>Нативные Sink’и и Source’ы:<\/b> В нашем примере мы использовали встроенный `webhook` коннектор — не нужно писать отдельный сервис для приема данных! RisingWave нативно умеет работать с Kafka\/Redpanda, Kinesis, Pulsar, а также писать данные напрямую в Iceberg, Delta Lake и другие системы.<\/li>\n<li><b>Инкрементальные вычисления “под капотом”:<\/b> Когда вы создаете материализованное представление, RisingWave строит план потоковой обработки. При поступлении новых данных он не пересчитывает все заново, а инкрементально обновляет результат. Это обеспечивает сверхнизкую задержку.<\/li>\n<\/ol>\n<h3>Глава 3: Практика: Строим наш Streaming Lakehouse шаг за шагом<\/h3>\n<p>Теперь перейдем к самому интересному — воссозданию нашего успешного проекта.<\/p>\n<h4>Этап 1: Архитектура и подготовка окружения (00:00 – 00:15)<\/h4>\n<p>Наша архитектура выглядит так:<br \/>\n`Webhook` → `RisingWave (Source → MView → Sink)` → `Lakekeeper (Catalog) + MinIO (Storage)` ← `Trino (Query)`<\/p>\n<p>Мы используем два `docker-compose` файла:<\/p>\n<ol start=\"1\">\n<li><b>Для Lakekeeper и его экосистемы<\/b> (Postgres, MinIO, Trino): <b>lakekeeper\/examples\/minimal<\/b> <a href=\"https:\/\/github.com\/lakekeeper\/lakekeeper\/tree\/main\/examples\/minimal\">https:\/\/github.com\/lakekeeper\/lakekeeper\/tree\/main\/examples\/minimal<\/a> .<\/li>\n<li><b>Для RisingWave и его окружения<\/b>  Postgres для метаданных, MinIO для состояния, Grafana):      <b>risingwave\/docker\/docker-compose.yml<\/b> <a href=\"https:\/\/github.com\/risingwavelabs\/risingwave\/blob\/main\/docker\/docker-compose.yml\">https:\/\/github.com\/risingwavelabs\/risingwave\/blob\/main\/docker\/docker-compose.yml<\/a> .<\/li>\n<\/ol>\n<p><b>Ключевое действие:<\/b> Мы запускаем оба стека, но для RisingWave вносим изменения, чтобы он мог взаимодействовать с Lakekeeper и Trino. Мы объединяем их в одну сеть, добавив в `docker-compose.yml` от RisingWave следующие строки:<\/p>\n<pre class=\"e2-text-code\"><code class=\"\"># risingwave\/docker\/docker-compose.yml\n\nservices:\n  risingwave-standalone:\n    # ...\n    # Открываем порт для вебхука, по умолчанию он не открыт наружу\n           .....\n        --webhook-listen-addr 0.0.0.0:4567 \\ \n           .....\n    ports:\n      - &quot;4566:4566&quot;. \n      # ... другие порты\n      - &quot;4567:4567&quot;   # &lt;--- Это важно для рабочего webhook \n    networks:\n      - trino_network\n# ... и для других сервисов, которые должны общаться с внешним стеком ...\n\nnetworks:\n  trino_network:\n    name: minimal_iceberg_net # Имя сети из docker-compose Lakekeeper\n    external: true<\/code><\/pre><p><b>Важный момент:<\/b> По умолчанию RisingWave не выставляет порт `4567` для вебхуков наружу. Мы добавили его в секцию `ports`, чтобы иметь возможность отправлять `curl` запросы с хост-машины.<\/p>\n<h4>Этап 2: Настройка каталогов (00:15 – 00:25)<\/h4>\n<p>“Озеро” без каталога — это просто “болото”. Lakekeeper будет нашим каталогом, а Trino — первым, кто научится им пользоваться.<\/p>\n<ol start=\"1\">\n<li><b>Создаем динамический каталог в Trino:<\/b><\/li>\n<\/ol>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE CATALOG risingwave USING iceberg\n    WITH (\n        &quot;iceberg.catalog.type&quot; = 'rest',\n        &quot;iceberg.rest-catalog.uri&quot; = 'http:\/\/lakekeeper:8181\/catalog',\n        &quot;iceberg.rest-catalog.warehouse&quot; = 'demo',\n        &quot;s3.region&quot;= 'dummy',\n        &quot;s3.path-style-access&quot; = 'true',\n        &quot;s3.endpoint&quot; = 'http:\/\/minio:9000',\n        &quot;fs.native-s3.enabled&quot; = 'true'\n    );<\/code><\/pre><ol start=\"2\">\n<li><b>Создаем “пустую” таблицу в Trino:<\/b> Этот шаг создает метаданные в Lakekeeper. RisingWave будет находить эту таблицу и наполнять ее данными.<\/li>\n<\/ol>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE TABLE risingwave.trino_namespace.product_view_events (\n       event_id varchar,\n       user_id varchar,\n       event_name varchar,\n       product_id varchar,\n       category varchar,\n       price double,\n       event_timestamp timestamp(6) with time zone,\n       raw_data varchar\n    );<\/code><\/pre><h4>Этап 3: Магия RisingWave (00:25 – 00:45) 🚀<\/h4>\n<p>Подключаемся к RisingWave через DBeaver (используя порт `4566` и стандартный драйвер PostgreSQL) и начинаем творить магию.<\/p>\n<ol start=\"1\">\n<li><b>Создаем источник-вебхук:<\/b><\/li>\n<\/ol>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE TABLE wbhtable1 (\n      data JSONB\n    ) WITH (\n      connector = 'webhook'\n    ) VALIDATE AS secure_compare(\n      headers-&gt;&gt;'authorization',\n      'TEST_WEBHOOK'\n    );<\/code><\/pre><p>Эта команда создает эндпоинт, который принимает JSON и кладет его в таблицу `wbhtable1`. `VALIDATE AS` обеспечивает простую, но эффективную аутентификацию.<\/p>\n<ol start=\"2\">\n<li><b>Создаем материализованное представление:<\/b><\/li>\n<\/ol>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE MATERIALIZED VIEW product_view_events AS\n    SELECT\n      (data-&gt;&gt;'event_id')::VARCHAR AS event_id,\n      (data-&gt;&gt;'user_id')::VARCHAR AS user_id,\n      (data-&gt;&gt;'event_name')::VARCHAR AS event_name,\n      (data-&gt;'properties'-&gt;&gt;'product_id')::VARCHAR AS product_id,\n      (data-&gt;'properties'-&gt;&gt;'category')::VARCHAR AS category,\n      (data-&gt;'properties'-&gt;&gt;'price')::DOUBLE PRECISION AS price,\n      (data-&gt;&gt;'timestamp')::TIMESTAMP WITH TIME ZONE AS event_timestamp,\n      data::VARCHAR AS raw_data\n    FROM wbhtable1;<\/code><\/pre><p>Это ядро нашей логики. Мы на лету парсим входящий `JSONB`, приводим типы и создаем структурированное представление `product_view_events`, которое обновляется автоматически.<\/p>\n<ol start=\"3\">\n<li><b>Создаем синк (Sink) в Iceberg:<\/b><\/li>\n<\/ol>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE SINK rest_sink FROM product_view_events\n    WITH (\n        connector = 'iceberg',\n        type = 'upsert',\n        primary_key = 'event_id',\n        catalog.type = 'rest',\n        catalog.uri = 'http:\/\/lakekeeper:8181\/catalog',\n        warehouse.path = 'demo',\n        database.name = 'trino_namespace',\n        table.name = 'product_view_events',\n        s3.endpoint = 'http:\/\/minio:9000',\n        s3.path.style.access = 'true',\n        s3.access.key = 'minio-root-user',\n        s3.secret.key = 'minio-root-password',\n        s3.region = 'dummy'\n    );<\/code><\/pre><p><b>“Грабли”, которые мы собрали:<\/b> На пути к этому финальному запросу мы столкнулись с несколькими ошибками, которые стоили нам времени. Вот они, чтобы вы не повторяли наших ошибок:<\/p>\n<ul>\n<li>`catalog.uri`: Должен указывать на полный путь к REST API каталогу, в случае Lakekeeper это `<a href=\"http:\/\/lakekeeper:8181\/catalog\">http:\/\/lakekeeper:8181\/catalog<\/a>`.<\/li>\n<li>`warehouse.path`: Должен содержать <b>логическое имя<\/b> хранилища (`demo`), а не его физический путь в S3.<\/li>\n<li>`s3.region`: <b>Критически важный параметр!<\/b> S3-клиент внутри RisingWave требует его обязательного указания, даже для MinIO. Хотя само значение (`us-east-1` или любое другое) для MinIO не принципиально, его отсутствие приводит к ошибке `region is missing` и сбою записи данных.<\/li>\n<\/ul>\n<h4>Этап 4: Запуск и проверка (00:45 – 01:00)<\/h4>\n<p>Время накормить нашу систему данными! Запускаем в терминале скрипт для генерации и отправки 100 событий, а можно и тысячу. Этот скрипт полностью рабочий и готов к копированию:<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">seq 1 100 | xargs -I {} -P 10 bash -c '\n  EVENT_ID=$(uuidgen)\n  USER_ID=&quot;usr_$(uuidgen | head -c 8)&quot;\n  PRODUCT_ID=&quot;prod_$(uuidgen | head -c 8)&quot;\n  TIMESTAMP=$(date -u +&quot;%Y-%m-%dT%H:%M:%SZ&quot;)\n\n  curl -s -o \/dev\/null -X POST \\\n    http:\/\/localhost:4567\/webhook\/dev\/public\/wbhtable1 \\\n    -H &quot;Content-Type: application\/json&quot; \\\n    -H &quot;Authorization: TEST_WEBHOOK&quot; \\\n    -d &quot;{\n          \\&quot;event_id\\&quot;: \\&quot;$EVENT_ID\\&quot;,\n          \\&quot;user_id\\&quot;: \\&quot;$USER_ID\\&quot;,\n          \\&quot;event_name\\&quot;: \\&quot;product_viewed\\&quot;,\n          \\&quot;properties\\&quot;: {\n            \\&quot;product_id\\&quot;: \\&quot;$PRODUCT_ID\\&quot;,\n            \\&quot;category\\&quot;: \\&quot;electronics\\&quot;,\n            \\&quot;price\\&quot;: 9199.99\n          },\n          \\&quot;timestamp\\&quot;: \\&quot;$TIMESTAMP\\&quot;\n        }&quot;\n'<\/code><\/pre><p>И вот он, момент истины. Идем в DBeaver, открываем подключение к Trino и выполняем:<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">select * from risingwave.trino_namespace.product_view_events;<\/code><\/pre><p>Результат перед вами:<\/p>\n<div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.22.20.png.jpg\" width=\"2560\" height=\"643\" alt=\"\" \/>\n<\/div>\n<p>Данные, только что сгенерированные и отправленные по HTTP, уже лежат в озере данных в формате Parquet и доступны для анализа. Ура!<\/p>\n<h3>Глава 4: Наблюдаемость: Смотрим на систему под нагрузкой<\/h3>\n<p>RisingWave поставляется с готовыми дашбордами для Grafana. Взглянем на них после нашей нагрузки.<\/p>\n<div class=\"e2-text-picture\">\n<div class=\"fotorama\" data-width=\"986\" data-ratio=\"2.293023255814\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.18.25.png\" width=\"986\" height=\"430\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.50.png\" width=\"986\" height=\"454\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.44.png\" width=\"976\" height=\"454\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.35.png\" width=\"974\" height=\"452\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.17.png\" width=\"980\" height=\"446\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.19.57.png\" width=\"978\" height=\"446\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.19.06.png\" width=\"978\" height=\"450\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.18.58.png\" width=\"974\" height=\"456\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.18.53.png\" width=\"978\" height=\"448\" alt=\"\" \/>\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.54.png\" width=\"980\" height=\"462\" alt=\"\" \/>\n<\/div>\n<div class=\"e2-text-caption\">Можно листать стрелками -->><\/div>\n<\/div>\n<ul>\n<li><b>Пропускная способность (Throughput):<\/b> Мы видим, как данные проходят через материализованное представление и записываются синком. Пики на графике соответствуют нашей нагрузке.<\/li>\n<\/ul>\n<ul>\n<li><b>Задержка барьеров (Barrier Latency):<\/b> Это ключевой показатель здоровья потоковой системы. Он показывает время, необходимое для создания контрольной точки (чекпоинта). Значения в десятки миллисекунд говорят о том, что система абсолютно здорова и справляется с нагрузкой без задержек.<\/li>\n<\/ul>\n<ul>\n<li><b>Ресурсы (CPU\/Memory):<\/b> Графики показывают стабильное и предсказуемое потребление ресурсов.<\/li>\n<\/ul>\n<p>Эти метрики доказывают, что система не просто работает, а работает стабильно и эффективно.<\/p>\n<h3>Заключение<\/h3>\n<p>Мы сделали это! Меньше чем за час мы развернули и настроили полноценный Streaming Lakehouse. Мы доказали, что современные инструменты, такие как RisingWave, могут кардинально упростить создание сложных систем обработки данных в реальном времени.<\/p>\n<p>Путь от ошибки `Table does not exist` до работающего пайплайна был непростым, но каждая решенная проблема углубляла мое понимание системы. Теперь есть не просто набор инструкций, а проверенный в бою рецепт, учитывающий все “подводные камни”.<\/p>\n<p>Путь к аналитике в реальном времени открыт. Хорошего стриминга и бурного потока с домом у озера, главное что бы избушку не смыло :)<\/p>\n<p>UPD: Проверил еще пару штук<\/p>\n<p>Создаем сурс из Кафки<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE SOURCE kafka_src (\n  action VARCHAR\n) WITH (\n  connector = 'kafka',\n  topic = 'query_complete',\n  properties.bootstrap.server = 'broker1:29092'\n);<\/code><\/pre><p>создаем синк в другую кафку<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE SINK kafka_sink from kafka_src WITH (\n  connector = 'kafka',\n  topic = 'query_complete',\n  properties.bootstrap.server = 'broker2:29092'\n) FORMAT PLAIN ENCODE JSON<\/code><\/pre><p>Еще вебхук<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE TABLE wbhtable2 (\n      data JSONB\n    ) WITH (\n      connector = 'webhook'\n    ) VALIDATE AS secure_compare(\n      headers-&gt;&gt;'authorization',\n      'TEST_WEBHOOK'\n    );<\/code><\/pre><p>Делаем материализацию<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE MATERIALIZED VIEW events AS\n    SELECT\n      (data-&gt;&gt;'action')::VARCHAR AS action\n    FROM wbhtable2;<\/code><\/pre><p>делаем синк из материализации в кафку<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">CREATE SINK kafka_sink2 FROM events WITH (\n  connector = 'kafka',\n  topic = 'query_complete',\n  properties.bootstrap.server = 'broker2:29092'\n) FORMAT PLAIN ENCODE JSON (force_append_only='true');<\/code><\/pre><p>Без материализации сообщения прилетают так: {“data”:“{\\”action\\“: \\”55555\\“}”}<br \/>\nА с материализацией: {“action”:“99999”}<\/p>\n<p>Пример запроса<\/p>\n<pre class=\"e2-text-code\"><code class=\"\">curl -s -o \/dev\/null -X POST \\\n    http:\/\/localhost:4567\/webhook\/dev\/public\/wbhtable2 \\\n    -H &quot;Content-Type: application\/json&quot; \\\n    -H &quot;Authorization: TEST_WEBHOOK&quot; \\\n    -d &quot;{\\&quot;action\\&quot;: \\&quot;11111\\&quot;}&quot;<\/code><\/pre><p>Еще про s3 подобные архитектуры: <a href=\"https:\/\/gavrilov.info\/all\/bitva-novyh-arhitektur-sravnivaem-arc-gigapi-i-ducklake\/\">https:\/\/gavrilov.info\/all\/bitva-novyh-arhitektur-sravnivaem-arc-gigapi-i-ducklake\/<\/a><\/p>\n<h2><b>Ошибки в версии 2.4 – какая то пакость была, но поставил 2.6.1 и все заработало<\/b><\/h2>\n<pre class=\"e2-text-code\"><code class=\"\">-- 0. устанавливаем последнюю версию risingwave ( 2.6.1 )\n\n\n-- 1 Создаем вебхук \n\nCREATE TABLE wbhtable5 (\n      data JSONB\n    ) WITH (\n      connector = 'webhook'\n    ) VALIDATE AS secure_compare(\n      headers-&gt;&gt;'authorization',\n      'TEST_WEBHOOK'\n    );\n\n-- 2 Создаем материализацию \n\nCREATE MATERIALIZED VIEW product_view_events5 AS\n    SELECT\n      (data-&gt;&gt;'event_id')::VARCHAR AS event_id,\n      (data-&gt;&gt;'user_id')::VARCHAR AS user_id,\n      (data-&gt;&gt;'event_name')::VARCHAR AS event_name,\n      (data-&gt;'properties'-&gt;&gt;'product_id')::VARCHAR AS product_id,\n      (data-&gt;'properties'-&gt;&gt;'category')::VARCHAR AS category,\n      (data-&gt;'properties'-&gt;&gt;'price')::DOUBLE PRECISION AS price,\n      (data-&gt;&gt;'timestamp')::TIMESTAMP WITH TIME ZONE AS event_timestamp,\n      data::VARCHAR AS raw_data\n    FROM wbhtable5;\n\n-- 3 создаем подключение к iceberg \n\nCREATE CONNECTION my_iceberg_conn5 WITH (\n    type = 'iceberg',\n    warehouse.path = 'risi',  -- s3:\/\/my-bucket\/warehouse\/\n   -- database.name = 'risi_space',  оказалось не нужна \n    s3.region = 'dummy',\n    s3.access.key = 'ЧЧЧ', -- Ваши ключи\n    s3.secret.key = 'ЧЧЧ',   -- Ваши ключи\n    catalog.type = 'rest',\n    s3.endpoint = 'https:\/\/gateway.storjshare.io',\n    s3.path.style.access = 'true',\n    \n    -- ИСПОЛЬЗУЕМ ИМЯ СЕРВИСА И ЕГО ВНУТРЕННИЙ ПОРТ!\n    catalog.uri = 'http:\/\/lakekeeper:8181\/catalog'\n)\n\n-- 4 Устанавливаем его по умолчанию \n\nSET iceberg_engine_connection = 'public.my_iceberg_conn5';\n\n\n-- Создаем таблицу ( обязательно с ключами )\n\nCREATE TABLE public.my_iceberg_table5 (\n       event_id VARCHAR PRIMARY KEY,\n       user_id varchar,\n       event_name varchar,\n       product_id varchar,\n       category varchar,\n       price double,\n       event_timestamp Timestamptz,\n       raw_data varchar\n) ENGINE = iceberg;\n\n \n-- 5 создаем синк \n\nCREATE SINK to_sales_events5 INTO my_iceberg_table5 AS\nSELECT * FROM product_view_events5;\n\n--- Тут можно curl запустить \n\nseq 1 10 | xargs -I {} -P 10 bash -c '\n  EVENT_ID=$(uuidgen)\n  USER_ID=&quot;usr_$(uuidgen | head -c 8)&quot;\n  PRODUCT_ID=&quot;prod_$(uuidgen | head -c 8)&quot;\n  TIMESTAMP=$(date -u +&quot;%Y-%m-%dT%H:%M:%SZ&quot;)\n\n  curl -s -o \/dev\/null -X POST \\\n    http:\/\/localhost:4567\/webhook\/dev\/public\/wbhtable5 \\\n    -H &quot;Content-Type: application\/json&quot; \\\n    -H &quot;Authorization: TEST_WEBHOOK&quot; \\\n    -d &quot;{\n          \\&quot;event_id\\&quot;: \\&quot;$EVENT_ID\\&quot;,\n          \\&quot;user_id\\&quot;: \\&quot;$USER_ID\\&quot;,\n          \\&quot;event_name\\&quot;: \\&quot;product_viewed\\&quot;,\n          \\&quot;properties\\&quot;: {\n            \\&quot;product_id\\&quot;: \\&quot;$PRODUCT_ID\\&quot;,\n            \\&quot;category\\&quot;: \\&quot;electronics\\&quot;,\n            \\&quot;price\\&quot;: 9199.99\n          },\n          \\&quot;timestamp\\&quot;: \\&quot;$TIMESTAMP\\&quot;\n        }&quot;\n'\n\n\n-- 6 проверяем \n\nSELECT * FROM product_view_events5;\n\n-- 7 проверяем ( появляются не сразу ) \n\nselect * from my_iceberg_table5<\/code><\/pre><div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.31.56.png\" width=\"2218\" height=\"530\" alt=\"\" \/>\n<\/div>\n<p>И в keeper она есть<\/p>\n<div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.33.31.png\" width=\"1420\" height=\"1166\" alt=\"\" \/>\n<\/div>\n<div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.34.29.png\" width=\"2428\" height=\"456\" alt=\"\" \/>\n<\/div>\n<p>И на S3<\/p>\n<div class=\"e2-text-picture\">\n<img src=\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.35.37.png\" width=\"1874\" height=\"126\" alt=\"\" \/>\n<\/div>\n",
            "date_published": "2025-09-06T23:03:00+03:00",
            "date_modified": "2025-11-09T17:38:42+03:00",
            "tags": [
                "Data",
                "Lakehouse",
                "Streaming"
            ],
            "image": "https:\/\/gavrilov.info\/pictures\/image-212.png-1.jpg",
            "_date_published_rfc2822": "Sat, 06 Sep 2025 23:03:00 +0300",
            "_rss_guid_is_permalink": "false",
            "_rss_guid": "278",
            "_rss_enclosures": [],
            "_e2_data": {
                "is_favourite": false,
                "links_required": [
                    "highlight\/highlight.js",
                    "highlight\/highlight.css",
                    "jquery\/jquery.js",
                    "fotorama\/fotorama.css",
                    "fotorama\/fotorama.js"
                ],
                "og_images": [
                    "https:\/\/gavrilov.info\/pictures\/image-212.png-1.jpg",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.37.28.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.39.34.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.22.20.png.jpg",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.18.25.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.50.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.44.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.35.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.17.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.19.57.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.19.06.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.18.58.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.18.53.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-06-v-22.20.54.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.31.56.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.33.31.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.34.29.png",
                    "https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-11-09-v-17.35.37.png"
                ]
            }
        }
    ],
    "_e2_version": 4171,
    "_e2_ua_string": "Aegea 11.4 (v4171e)"
}