Azure Data Factory надає вбудовану функцію моніторингу для відстеження статусів конвеєрів та активностей. Однак, якщо ви хочете створити власну систему журналювання, яка захоплює та записує деталі у SQL базу даних, цей блог допоможе вам налаштувати один з багатьох підходів для цього.
Цей блог розповість, як створити власну систему журналювання для відстеження активностей в Azure Data Factory, яка буде записувати деталі в SQL базу даних. Ми зосередимось на простому і практичному методі налаштування цього процесу.
Мета
Навчитися будувати таблицю журналу для відстеження активностей конвеєра в Azure Data Factory.
Що ви дізнаєтеся з цього
- Які різні об'єкти результату повертаються кожною активністю
- Як працює кожна активність
- Як налаштувати змінні
- Як додавати записи з конвеєра в SQL за допомогою збережених процедур
- Як працюють кожен умовний шлях (On Success, On Failure, On Complete, On Skip)
Архітектура
Резюме
Витяг даних
- Дані отримуються з HTTP джерела за допомогою інструменту Copy Data.
Потік конвеєра
- Інструмент Copy Data записує інформацію про завершення в Append Copy Data log.
Скопійовані дані проходять через кілька етапів:
- Notebook1 обробляє дані до рівня Bronze та записує його завершення в Notebook1 log.
- Notebook2 обробляє дані до рівня Silver після Notebook1 та записує його завершення в Notebook2 log.
- Notebook3 обробляє дані до рівня Gold після Notebook2 та записує його завершення в Notebook3 log.
Архівування даних
- Оброблені дані надсилаються до Archive Data Flow, де архівуються в приймач.
- Після успішного архівування, дані видаляються з джерела.
- Інформація про архівування записується в Append Archive log.
Механізм журналювання
- Журнали додаються в Append_Variable (ARRAY), який зберігає деталі активностей (наприклад, Copy Data, Archive, Notebook1, Notebook2, Notebook3, Delete).
Налаштування змінних конвеєра
Налаштування значення змінної Append
#Для звичайних активностей
@concat(
'{"startTime":"', activity('').ExecutionStartTime,
'","endTime":"', activity('').ExecutionEndTime,
'","processId":"', activity('').ActivityRunId,
'","processName":"Process name eg: Copy Data',
'","status":"', activity('').Status,
'","errorMessage":"', activity('').Error?.message,
'","pipelineName":"',pipeline().Pipeline,
'"}'
)
#Для активностей Databricks Notebook
@concat(
'{"startTime":"', activity('').ExecutionStartTime,
'","endTime":"', activity('').ExecutionEndTime,
'","processId":"', activity('').ActivityRunId,
'","processName":"Process name eg: Copy Data',
'","status":"', activity('').Status,
'","errorMessage":"', #if the error exist but not thrown from notebook
if(
equals(activity('').output?.runError, null),
activity('').Error?.Message, #default error
activity('').output?.runError #Notebook thrown error
),
'","pipelineName":"', pipeline().Pipeline,
'"}'
)
Останнє журналювання до SQL
- Append_Variable (що містить всі журнали активностей) обробляється по черзі:
- Кожен елемент у масиві встановлюється як змінна.
- Збережена процедура записує кожен запис журналу з змінної в Log_Table в SQL базі даних.
Налаштування активності ForEach
-- Збережена процедура SQL (створіть SP у вашій SQL базі даних)
CREATE PROCEDURE LogActivityExecution
@PipelineName NVARCHAR(255),
@ActivityName NVARCHAR(255),
@StartTime DATETIME,
@EndTime DATETIME,
@Status NVARCHAR(50),
@ErrorMessage NVARCHAR(MAX)
AS
BEGIN
INSERT INTO ActivityLogs (PipelineName, ActivityName, StartTime, EndTime, Status, ErrorMessage)
VALUES (@PipelineName, @ActivityName, @StartTime, @EndTime, @Status, @ErrorMessage);
END
Налаштування активності збереженої процедури
Умовні шляхи
- On Success
Мета: Забезпечує виконання наступної активності лише в разі успішного виконання попередньої активності.
Чому використовується: Для підтримки логічного потоку конвеєра та уникнення непотрібного або некоректного виконання, якщо активність не вдалася.
Наприклад, дані повинні просуватися до наступного етапу в блокноті або архіву лише після успішної трансформації або копіювання.
- On Complete
Мета: Запускає підключену активність незалежно від того, чи була попередня активність успішною чи ні.
Чому використовується: Для того, щоб забезпечити завжди виконуване логування, незалежно від успіху чи невдачі. Це важливо для додавання деталей активностей (наприклад, журнал Copy Data, журнали Notebook) в масив Append Variable, щоб підтримувати повний запис виконання конвеєра.
- On Skip
Мета: Виконує підключену активність, якщо попередня активність була пропущена або не вдалося її виконати.
Чому використовується: Для обробки сценаріїв невдачі, коли конвеєр зупиняється передчасно. Це забезпечує, що журнали все одно додаються в таблицю журналів, фіксуючи деталі всіх виконаних активностей до моменту невдачі для кращого налагодження та відстеження.
[
Помилка конвеєра та повідомлення про помилку — Azure Data Factory
Дізнайтеся, як визначаються статус помилки конвеєра та повідомлення про помилку
learn.microsoft.com
](https://learn.microsoft.com/en-us/azure/data-factory/tutorial-pipeline-failure-error-handling?source=post_page-----0e08a0faecce--------------------------------)
Остаточний ADF конвеєр
Конвеєр ADF: Реалізація архітектури
Сподіваюся, що ця стаття дала вам чітке розуміння того, як створювати журнали для кількох активностей у конвеєрі.
Дякуємо за читання!
Перекладено з: Logging Activity Details into SQL within Azure Data Factory