Техническое руководство по отслеживанию происхождения данных (4.x)
Механизм отслеживания происхождения данных извлекает зависимости на уровне таблиц и столбцов из поддерживаемых операторов DML Doris и передаёт их во внешние системы управления данными через LineagePlugin. Doris выступает источником сведений о происхождении данных. Он не сохраняет события, не предоставляет API для запросов к этим сведениям и не обеспечивает их визуализацию.
Эта возможность доступна в линейке выпусков 4.x начиная с Apache Doris 4.0.6.
Возможности и ограничения
Механизм формирует событие только после успешного выполнения одного из следующих операторов:
| Поддерживаемый оператор | Формирование события |
|---|---|
INSERT INTO ... SELECT | Формирует одно событие после успешного выполнения Insert. |
INSERT OVERWRITE TABLE ... SELECT | Формирует одно событие после успешного выполнения Overwrite. |
CREATE TABLE AS SELECT | Формирует одно событие после успешного выполнения внутреннего Insert. |
Текущая реализация не формирует события для SELECT, UPDATE, DELETE, заданий загрузки данных, вставок только с VALUES и операций записи, целевым объектом которых является __internal_schema. Некоторые пути выполнения UPDATE и DELETE внутри используют команду Insert, однако проверка исходного типа команды исключает их из отправки событий происхождения данных.
События происхождения данных доставляются асинхронно, по принципу best effort. Успешно выполненный оператор DML не откатывается при сбое сбора события или его доставки плагином. При заполнении очереди события могут отбрасываться; Doris не повторяет попытки доставки и не сохраняет события.
Техническая реализация
Процесс сбора
Для поддерживаемого оператора Doris сохраняет проанализированный логический план Nereids в хуке планировщика afterAnalyze. После успешного выполнения DML Doris извлекает сведения о происхождении данных из этого плана и помещает событие в локальную очередь FE. Один фоновый поток-демон проверяет каждый загруженный плагин и отправляет событие каждому плагину, у которого eventFilter() возвращает true.
Перед извлечением Doris вызывает eventFilter() у загруженных плагинов. Если ни один плагин не готов принимать события, извлечение пропускается. Перед отправкой фоновый поток повторно проверяет eventFilter().
Содержимое событий происхождения данных
Каждое событие LineageInfo содержит связи происхождения данных на уровне таблиц, прямые связи на уровне столбцов, два вида косвенных связей и контекст запроса. Полный пример SQL ниже создаёт исходные таблицы, целевую таблицу и результат, которые используются в следующих разделах.
Пример SQL и результата
У текущего пользователя должны быть права на создание базы данных и таблиц, запись и чтение данных, а на FE должен быть загружен плагин отслеживания происхождения данных. Пример использует две исходные таблицы и формирует сводку по клиентам с помощью CTE, Join, фильтров, агрегации, оконной функции, условного выражения и сортировки.
CREATE DATABASE IF NOT EXISTS lineage_demo;
USE lineage_demo;
CREATE TABLE lineage_orders (
order_id BIGINT,
customer_id BIGINT,
region VARCHAR(16),
amount DECIMAL(18, 2),
status VARCHAR(16)
)
DUPLICATE KEY(order_id)
DISTRIBUTED BY HASH(order_id) BUCKETS 1
PROPERTIES ("replication_num" = "1");
CREATE TABLE lineage_customers (
customer_id BIGINT,
customer_name VARCHAR(64),
customer_level VARCHAR(16)
)
DUPLICATE KEY(customer_id)
DISTRIBUTED BY HASH(customer_id) BUCKETS 1
PROPERTIES ("replication_num" = "1");
CREATE TABLE lineage_customer_summary (
customer_id BIGINT,
region_label VARCHAR(16),
total_amount DECIMAL(18, 2),
customer_seq BIGINT,
region_group VARCHAR(16)
)
DUPLICATE KEY(customer_id)
DISTRIBUTED BY HASH(customer_id) BUCKETS 1
PROPERTIES ("replication_num" = "1");
INSERT INTO lineage_customers VALUES
(1, 'Alice', 'VIP'),
(2, 'Bob', 'VIP'),
(3, 'Carol', 'REGULAR');
INSERT INTO lineage_orders VALUES
(101, 1, 'east', 100.00, 'PAID'),
(102, 1, 'east', 50.00, 'PAID'),
(103, 2, 'west', 80.00, 'PAID'),
(104, 2, 'west', 30.00, 'CANCELLED'),
(105, 3, 'north', 200.00, 'PAID');
INSERT INTO lineage_customer_summary
WITH customer_totals AS (
SELECT
o.customer_id,
UPPER(o.region) AS region_label,
SUM(o.amount) AS total_amount
FROM lineage_orders o
JOIN lineage_customers c
ON o.customer_id = c.customer_id
WHERE o.status = 'PAID'
AND c.customer_level = 'VIP'
GROUP BY o.customer_id, UPPER(o.region)
HAVING SUM(o.amount) >= 50
)
SELECT
customer_id,
region_label,
total_amount,
ROW_NUMBER() OVER (ORDER BY customer_id) AS customer_seq,
CASE
WHEN region_label = 'EAST' THEN 'CORE'
ELSE 'OTHER'
END AS region_group
FROM customer_totals
ORDER BY total_amount DESC;
SELECT customer_id, region_label, total_amount, customer_seq, region_group
FROM lineage_customer_summary
ORDER BY customer_id;
Запрос возвращает:
+-------------+--------------+--------------+--------------+--------------+
| customer_id | region_label | total_amount | customer_seq | region_group |
+-------------+--------------+--------------+--------------+--------------+
| 1 | EAST | 150.00 | 1 | CORE |
| 2 | WEST | 80.00 | 2 | OTHER |
+-------------+--------------+--------------+--------------+--------------+
Две вставки только с VALUES не формируют события происхождения данных. Оператор INSERT INTO lineage_customer_summary ... SELECT формирует одно событие LineageInfo, содержимое которого описано ниже.
Происхождение данных на уровне таблиц
Связи происхождения данных на уровне таблиц фиксируют одну целевую таблицу и все исходные таблицы, сканируемые проанализированным логическим планом. Запрос, записывающий данные в lineage_customer_summary, сканирует и lineage_orders, и lineage_customers, поэтому обе исходные таблицы связаны с целевой на уровне таблиц. Для CTE определяются лежащие в его основе таблицы, а каждая ветвь UNION вносит свой вклад в набор исходных таблиц.
В этом примере формируются следующие связи происхождения данных на уровне таблиц:
| Исходная таблица | Целевая таблица |
|---|---|
lineage_orders | lineage_customer_summary |
lineage_customers | lineage_customer_summary |
Прямые связи происхождения данных на уровне столбцов
Прямые связи происхождения данных на уровне столбцов показывают, какое исходное выражение формирует каждое значение целевого столбца. После раскрытия псевдонимов и CTE Doris классифицирует выражения в следующем порядке: выражение, содержащее агрегатную функцию, получает тип AGGREGATION; простая ссылка на исходный столбец получает тип IDENTITY; любое другое выражение получает тип TRANSFORMATION.
| Тип | Значение | Пример |
|---|---|---|
IDENTITY | Целевое значение берётся непосредственно из исходного столбца без функции или операции. | lineage_customer_summary.customer_id берётся непосредственно из lineage_orders.customer_id. |
TRANSFORMATION | Неагрегатное выражение преобразует целевое значение, например с помощью арифметической операции, строковой функции, оконной функции или условного выражения. | region_label получается из UPPER(lineage_orders.region); customer_seq получается из ROW_NUMBER(). |
AGGREGATION | Целевое значение формируется выражением, содержащим агрегатную функцию. Этот тип сохраняется, даже если агрегатная функция обёрнута другим выражением. | total_amount получается из SUM(lineage_orders.amount). |
В этом примере формируются следующие прямые связи происхождения данных на уровне столбцов:
| Целевой столбец | Исходное выражение | Тип |
|---|---|---|
lineage_customer_summary.customer_id | lineage_orders.customer_id | IDENTITY |
lineage_customer_summary.region_label | UPPER(lineage_orders.region) | TRANSFORMATION |
lineage_customer_summary.total_amount | SUM(lineage_orders.amount) | AGGREGATION |
lineage_customer_summary.customer_seq | ROW_NUMBER() OVER (ORDER BY lineage_orders.customer_id) | TRANSFORMATION |
lineage_customer_summary.region_group | CASE WHEN UPPER(lineage_orders.region) = 'EAST' ... END | TRANSFORMATION |
Косвенные связи происхождения данных на уровне набора данных
Косвенные связи происхождения данных на уровне набора данных фиксируют выражения, которые влияют на весь результирующий набор, но не формируют значение целевого столбца напрямую. Эти зависимости хранятся в отображении на уровне события и при необходимости могут применяться к каждому целевому выходному столбцу.
| Тип | Значение | Пример |
|---|---|---|
JOIN | Условия Join определяют, какие исходные строки можно объединить. | lineage_orders.customer_id = lineage_customers.customer_id. |
FILTER | Условия WHERE или HAVING определяют, какие строки или группы попадают в результат. | status = 'PAID', customer_level = 'VIP' и HAVING SUM(amount) >= 50. |
GROUP_BY | Выражения группировки определяют детализацию результатов агрегации. | Группировка по customer_id и UPPER(region). |
SORT | Выражения ORDER BY или TopN определяют порядок результата. | Сортировка по total_amount DESC, которое раскрывается в SUM(amount). |
В этом примере формируются следующие косвенные связи происхождения данных на уровне набора данных:
| Тип | Пример исходного выражения в событии | Влияние |
|---|---|---|
JOIN | lineage_orders.customer_id = lineage_customers.customer_id | Определяет, как заказы сопоставляются с клиентами. |
FILTER | status = 'PAID', customer_level = 'VIP' и SUM(amount) >= 50 | Предикаты WHERE и HAVING совместно определяют, какие строки и группы попадают в результат. |
GROUP_BY | lineage_orders.customer_id и UPPER(lineage_orders.region) | Определяют детализацию агрегации для каждого клиента и региона. |
SORT | total_amount DESC, раскрываемое в SUM(lineage_orders.amount) | Определяет порядок строк в результате запроса Insert. |
Косвенные связи происхождения данных на уровне выходных столбцов
Косвенные связи происхождения данных на уровне выходных столбцов привязываются только к затронутому целевому столбцу и не распространяются на остальные выходные столбцы.
| Тип | Значение | Пример |
|---|---|---|
WINDOW | Входные выражения PARTITION BY и ORDER BY оконной функции влияют на её выходной столбец. | customer_seq формируется выражением ROW_NUMBER() OVER (ORDER BY customer_id), у которого зависимость типа WINDOW задаётся выражением customer_id. |
CONDITIONAL | Условие CASE или IF, а также выражения-кандидаты COALESCE влияют на выходной столбец. | region_group определяется выражением CASE WHEN region_label = 'EAST' ..., у которого зависимостью типа CONDITIONAL является этот предикат. |
Само оконное или условное выражение по-прежнему отражается в прямых связях происхождения данных на уровне столбцов. Например, тип прямой связи для customer_seq имеет значение TRANSFORMATION, и у этого столбца также есть косвенная связь типа WINDOW. Тип прямой связи для region_group тоже имеет значение TRANSFORMATION, и у этого столбца также есть косвенная связь типа CONDITIONAL.
В этом примере формируются следующие косвенные связи происхождения данных на уровне выходных столбцов:
| Целевой столбец | Тип | Исходное выражение в событии | Влияние |
|---|---|---|---|
customer_seq | WINDOW | lineage_orders.customer_id | Используется как входное выражение сортировки окна для ROW_NUMBER() и влияет только на customer_seq. |
region_group | CONDITIONAL | UPPER(lineage_orders.region) = 'EAST' | Используется как условие CASE WHEN и влияет только на region_group. |
У остальных трёх целевых столбцов нет косвенных связей на уровне выходных столбцов. Зависимости уровня набора данных JOIN, FILTER, GROUP_BY и SORT по-прежнему влияют на весь результирующий набор.
Контекст запроса
Контекст запроса указывает, кто сформировал событие происхождения данных, в какой сессии и с помощью какого оператора DML.
| Поле | Пример или значение |
|---|---|
| Исходная команда | InsertIntoTableCommand, указывающая, что событие сформировано командой Insert. |
| Идентификатор запроса и исходный SQL | Идентификатор запроса и полный текст INSERT INTO lineage_customer_summary .... |
| Пользователь и IP-адрес клиента | Например, lineage_user и 192.0.2.10. |
| База данных и Catalog сессии | В этом примере: lineage_demo и internal. Они описывают контекст сессии и не обязательно указывают расположение целевой таблицы. |
| Состояние выполнения | Для успешно выполненного оператора DML обычно записывается OK. |
| Временная метка и длительность | Время создания события и число миллисекунд от начала запроса до создания события. Если значение недоступно, записывается -1. |
| Свойства внешних Catalog | Неконфиденциальные свойства внешних Catalog, к которым обращается запрос. Пароли, ключи и скрытые свойства удаляются. Если используется только Catalog internal, отображение пусто. |
В этом примере исходный SQL представляет собой приведённый выше сложный Insert, исходная команда имеет значение InsertIntoTableCommand, база данных сессии имеет имя lineage_demo, Catalog сессии имеет имя internal, а состояние успешного выполнения имеет значение OK. Идентификатор запроса, пользователь, IP-адрес клиента, временная метка и длительность берутся из фактического контекста выполнения. Поскольку в примере используется только Internal Catalog, отображение свойств внешних Catalog пусто.
Убедитесь в принимающей системе плагина, что событие получено. При первоначальном развёртывании также проверьте fe.log на наличие Loaded lineage plugin, ошибок конкретного плагина и предупреждений о заполнении очереди.
Механизм извлечения раскрывает выражения производителей CTE и разворачивает ветви UNION до доставки события. Плагин, передающий событие во внешнюю систему, должен преобразовать Java-объекты Doris, такие как TableIf, SlotReference и Expression, в стабильные идентификаторы или собственную схему события.
Поведение при обработке событий
lineage_event_queue_size задаёт максимальное количество событий происхождения данных, ожидающих обработки в каждой локальной очереди FE. Параметр измеряется в событиях, а не в байтах. Когда очередь заполнена, новое событие отбрасывается, а DML продолжает выполняться в обычном режиме. Фоновый поток вызывает плагины последовательно, поэтому медленный плагин задерживает последующие события и при длительной нагрузке может приводить к их потере. Исключения, выбрасываемые плагином, записываются в журнал и не останавливают фоновый поток и не влияют на DML.
Обнаружение и инициализация плагинов выполняются только при запуске FE. Динамическая перезагрузка и выгрузка не поддерживаются.
Настройка и использование
Предварительные требования
- Подготовьте внешний плагин отслеживания происхождения данных. В Doris нет встроенного приёмника событий. Описание контракта SPI, упаковки JAR и полную минимальную реализацию плагина см. в разделе Разработка плагина отслеживания происхождения данных.
- Скопируйте JAR-файл плагина и все необходимые JAR-файлы сторонних зависимостей на каждый FE.
- Если плагин отправляет события во внешний сервис управления данными, убедитесь, что этот сервис доступен с каждого FE, на котором могут выполняться операторы DML.
Развёртывание плагина
FE_HOME обозначает каталог одной установки FE, содержащий bin/, conf/ и lib/. Используйте следующую структуру на каждом FE. Загрузчик просматривает только непосредственные подкаталоги lineage/, а также JAR-файлы в каждом каталоге плагина и его подкаталоге lib/.
$FE_HOME/plugins/
└── lineage/
└── example-lineage/
├── example-lineage.jar
└── lib/
└── downstream-client.jar
Настройте fe.conf на каждом FE. Замените example-lineage значением, которое возвращает метод name() фабрики плагина:
plugin_dir = /opt/apache-doris/fe/plugins
activate_lineage_plugin = example-lineage
lineage_event_queue_size = 50000
| Параметр | Тип и значение по умолчанию | Обязателен | Описание |
|---|---|---|---|
plugin_dir | Строка; $FE_HOME/plugins | Нет | Корневой каталог плагинов. Загрузчик плагинов происхождения данных просматривает его непосредственные подкаталоги внутри lineage/. При использовании каталога по умолчанию этот параметр можно не указывать. |
activate_lineage_plugin | Массив строк; пустой массив | Рекомендуется явная настройка | Имена Factory, перечисленные через запятую, для создания экземпляров. Пустое значение отключает фильтрацию по именам и приводит к созданию экземпляров всех обнаруженных фабрик. |
lineage_event_queue_size | Положительное целое число; 50000 | Нет | Максимальное количество ожидающих обработки событий происхождения данных на текущем FE. При заполнении очереди новые события отбрасываются. |
Все три параметра задаются в $FE_HOME/conf/fe.conf на каждом FE и действуют только для соответствующего процесса FE. Они считываются только при запуске FE, поэтому после их изменения перезапустите соответствующий FE.
Активация плагинов
Настройте activate_lineage_plugin в $FE_HOME/conf/fe.conf на каждом FE. Этот параметр определяет, для каких обнаруженных фабрик создаются экземпляры. Он не отвечает за обнаружение JAR-файлов и не управляет вместимостью очереди. Каждое имя чувствительно к регистру и должно точно совпадать со значением LineagePluginFactory.name(). Реализация LineagePlugin.name() должна возвращать то же имя. Имя каталога плагина при сопоставлении не используется. Несколько имён плагинов разделяйте запятыми. Парсер конфигурации FE удаляет пробелы по краям каждого элемента в списке, разделённом запятыми, однако сопоставление имён остаётся чувствительным к регистру:
activate_lineage_plugin = example-lineage,governance-lineage
При запуске FE обнаруживает встроенные и внешние фабрики, а затем создаёт экземпляры только для настроенных имён. Каждый FE считывает собственную конфигурацию, поэтому на всех FE, которые могут выполнять DML, должен использоваться одинаковый набор плагинов. После изменения перезапустите FE. Этот параметр нельзя изменить динамически через ADMIN SET FRONTEND CONFIG.
В текущей реализации пустое значение activate_lineage_plugin не отключает плагины отслеживания происхождения данных. Оно пропускает фильтрацию по именам и приводит к созданию экземпляров всех обнаруженных фабрик. Явно перечисляйте плагины, которые нужно включить, и не используйте пустое значение для отключения. После загрузки плагина eventFilter() по-прежнему определяет, получит ли он событие; проверка выполняется перед извлечением и перед отправкой.
Настройка очереди событий
Настройте lineage_event_queue_size в $FE_HOME/conf/fe.conf на каждом FE. Значение должно быть положительным целым числом. У каждого FE, который может выполнять DML, есть независимая очередь, требующая отдельной настройки.
Этот параметр считывается при запуске FE и не может быть изменён динамически через ADMIN SET FRONTEND CONFIG. После его изменения перезапустите соответствующий FE. Увеличение значения повышает вместимость очереди ожидающих событий и потребление памяти FE. Оно не добавляет потоки обработки и не увеличивает пропускную способность плагинов. При потере событий сначала найдите в fe.log сообщения the lineage event queue is full и сократите время выполнения метода exec() плагина. Затем подберите размер очереди с учётом пикового числа ожидающих событий и доступной памяти FE.
Перезапускайте каждый FE после изменения других настроек плагина или замены его JAR-файлов. Конфигурация и каталоги плагинов считываются только при запуске FE.
Эксплуатация и устранение неполадок
Нет события после записи
Частые причины: неподдерживаемый оператор, ошибка выполнения DML, отсутствие загруженного плагина на текущем FE или возврат eventFilter() значения false.
Выполните проверки в следующем порядке:
- Убедитесь, что успешно выполнен оператор
INSERT INTO ... SELECT,INSERT OVERWRITE TABLE ... SELECTилиCREATE TABLE AS SELECT. Insert только сVALUESне формирует событие. - Убедитесь, что на FE, выполнившем DML, плагин настроен и загружен. Найдите в
fe.logсообщениеLoaded lineage plugin. - Убедитесь, что метод
eventFilter()плагина возвращаетtrueкак в потоке запроса, так и в фоновом потоке обработки. - Проверьте журналы плагина и принимающий сервис, чтобы убедиться, что обработка не завершилась ошибкой за пределами FE.
Плагин не загружен
Убедитесь, что каталог плагина содержит JAR-файл, а JAR-файл содержит META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory. Каждый каталог плагина может предоставлять только одну Factory. Имена Factory должны быть глобально уникальными и соответствовать значению activate_lineage_plugin. После исправления каталога, JAR-файла или конфигурации перезапустите соответствующий FE.
Потеря событий при высокой нагрузке
Найдите в fe.log сообщения the lineage event queue is full и исключения плагина. Если очередь заполнена, сначала сократите синхронные блокирующие операции в exec() и обеспечьте идемпотентную доставку во внешнюю систему с ограниченным числом повторных попыток. Затем оцените, следует ли увеличить lineage_event_queue_size. Перед увеличением учтите доступную память FE, поскольку размер событий зависит от оператора SQL.
Также убедитесь, что на всех FE, которые могут выполнять DML, используются одинаковые плагин и конфигурация. Каждый FE использует независимую локальную очередь; состояние плагинов и очередей между FE не синхронизируется.
Обновление плагина не вступило в силу
Каталоги плагинов просматриваются только при запуске FE. Динамическая перезагрузка и выгрузка не поддерживаются. После замены JAR-файлов плагина, зависимостей или конфигурации перезапустите все соответствующие FE и снова проверьте наличие Loaded lineage plugin.
Используйте идентификатор запроса и целевую таблицу для формирования идентификаторов событий во внешней системе. Это позволяет системе управления данными устранять дубликаты событий, даже если плагин реализует повторные попытки доставки вне данного механизма.