Техническое руководство по отслеживанию происхождения данных

Механизм отслеживания происхождения данных извлекает зависимости на уровне таблиц и столбцов из поддерживаемых операторов DML в Doris и передаёт их во внешние системы управления данными через LineagePlugin. Doris выступает источником сведений о происхождении данных. Он не хранит события, не предоставляет API для запросов к этим сведениям и средств их визуализации.

Эта возможность доступна в версии Dev. Впервые она появилась в 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, однако проверка типа исходной команды исключает их из отправки событий происхождения данных.

Гарантия доставки

События происхождения данных доставляются асинхронно и без гарантии доставки. При сбое сбора событий или их доставки плагину успешно выполненный оператор 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	Условия соединения определяют, какие исходные строки могут быть объединены.	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.
Query ID и исходный SQL	Query ID и полный текст 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. Query ID, пользователь, IP-адрес клиента, метка времени и длительность берутся из фактического контекста выполнения. Поскольку в примере используется только Internal Catalog, словарь свойств внешних Catalog пуст.

Убедитесь в получении события в системе, куда плагин отправляет данные. При первоначальном развёртывании также проверьте fe.log на наличие Loaded lineage plugin, ошибок плагина и предупреждений о заполнении очереди.

Перед доставкой события механизм извлечения разрешает выражения, формирующие CTE, и раскрывает ветви UNION. Плагин, передающий данные во внешнюю систему, должен преобразовывать объекты Java в Doris, такие как TableIf, SlotReference и Expression, в стабильные идентификаторы или в собственную схему событий.

Поведение при обработке событий

lineage_event_queue_size задаёт максимальное количество событий происхождения данных, ожидающих обработки в каждой локальной очереди FE. Параметр учитывает количество событий, а не байтов. Когда очередь заполнена, новое событие отбрасывается, а DML продолжает выполняться в обычном режиме. Фоновый поток вызывает плагины последовательно, поэтому медленный плагин задерживает последующие события и при длительной нагрузке может приводить к их потере. Исключения, возникающие в плагине, записываются в журнал, не останавливают фоновый поток и не влияют на DML.

Обнаружение и инициализация плагинов выполняются только при запуске FE. Динамическая перезагрузка и выгрузка не поддерживаются.

Настройка и использование

Предварительные требования

1. Подготовьте внешний плагин отслеживания происхождения данных. В Doris нет встроенного компонента отправки таких событий. Описание контракта SPI, упаковки JAR и полный минимальный пример плагина см. в разделе Разработка плагинов для отслеживания происхождения данных.
2. Скопируйте JAR плагина и все необходимые JAR сторонних зависимостей на каждый FE.
3. Если плагин отправляет события во внешнюю службу управления данными, убедитесь, что эта служба доступна с каждого 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	String; $FE_HOME/plugins	Нет	Корневой каталог плагинов. Загрузчик плагинов происхождения данных просматривает непосредственные подкаталоги внутри lineage/. При использовании каталога по умолчанию этот параметр можно не указывать.
activate_lineage_plugin	Массив String; пустой	Рекомендуется указать явно	Имена 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.

Выполните проверки в следующем порядке:

1. Убедитесь, что успешно выполнен оператор INSERT INTO ... SELECT, INSERT OVERWRITE TABLE ... SELECT или CREATE TABLE AS SELECT. Insert только с VALUES не создаёт событие.
2. Убедитесь, что на FE, выполнившем DML, плагин настроен и загружен. Найдите в fe.log сообщение Loaded lineage plugin.
3. Убедитесь, что eventFilter() плагина возвращает true как в потоке запроса, так и в фоновом потоке обработки.
4. Проверьте журналы плагина и принимающую службу, чтобы убедиться, что при обработке за пределами 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.

Используйте Query ID и целевую таблицу для формирования идентификаторов событий в принимающей системе. Это позволит системе управления данными устранять дубликаты событий, даже если плагин реализует повторные попытки доставки за пределами механизма Doris.

Дополнительные материалы

• Разработка плагинов для отслеживания происхождения данных