Как контролировать сбои данных в конвейерах фабрики данных Azure?

Question

Время от времени я получаю сообщение об ошибке из-за несовместимых данных в моем исходном наборе данных по сравнению с моим целевым набором данных. Я хотел бы контролировать действие, которое конвейер определяет на основе типов ошибок, может быть, выводить или отбрасывать эти строки частиц, но завершать все остальное. Это возможно? Кроме того, возможно ли получить фактическую ошибочную строку (и) из фабрики данных, не обращаясь и не ища в фактическом наборе данных источника каким-либо простым способом?

При копировании обнаружена ошибка пользователя на стороне приемника: ErrorCode=UserErrorInvalidDataValue,'Type=Microsoft.DataTransfer.Common.Shared.HybridDeliveryException,Message= столбец'Timestamp'содержит недопустимое значение'11667'. Невозможно преобразовать '11667' в тип 'DateTimeOffset'.,Source=Microsoft.DataTransfer.Common,''Type=System.FormatException,Message=String не был распознан как допустимый DateTime.,Source=mscorlib,'.

Спасибо

Answer 1

Я думаю, что вы столкнулись с довольно распространенной проблемой и ограничениями в ADF. Хотя наборы данных, которые вы определяете с помощью JSON, позволяют ADF понимать структуру данных, то есть все, только структура, инструмент оркестровки не может ничего сделать для преобразования или манипулирования данными как части обработки действия.

Чтобы ответить на ваш вопрос напрямую, это, безусловно, возможно. Но вам нужно взломать C# и использовать функциональность расширяемости ADF, чтобы справиться с плохими строками, прежде чем передавать их в конечный пункт назначения.

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

Этот подход мы часто используем, так как не все данные идеальны (я желаю), а ETL или ELT не работают. Я предпочитаю акроним ECLT. Где "С" означает "чистый". Или очистить, подготовить и т. Д. Это, безусловно, относится к ADF, потому что у этой службы нет собственного механизма вычислений или потока данных в стиле SSIS.

Так...

С точки зрения того, как это сделать. Сначала я рекомендую вам прочитать этот пост в блоге о создании пользовательских действий ADF. Ссылка на сайт:

https://www.purplefrogsystems.com/paul/2016/11/creating-azure-data-factory-custom-activities/

Тогда в вашем классе C# унаследовано от IDotNetActivity сделать что-то вроде ниже.

    public IDictionary<string, string> Execute(
        IEnumerable<LinkedService> linkedServices,
        IEnumerable<Dataset> datasets,
        Activity activity,
        IActivityLogger logger)
    {

    //etc

    using (StreamReader vReader = new StreamReader(YourSource))
        {
            using (StreamWriter vWriter = new StreamWriter(YourDestination))
            {
                while (!vReader.EndOfStream)
                {
                //data transform logic, if bad row etc
                }
            }
        }
  }

Вы поняли идею. Создайте свой собственный поток данных SSIS!

Затем запишите чистую строку в качестве выходного набора данных, который может быть входом для вашего следующего действия ADF. Либо с несколькими конвейерами, либо в виде цепочки действий внутри одного конвейера.

Это единственный способ, с помощью которого ADF сможет справиться с вашими неверными данными в текущих предложениях услуг.

Надеюсь это поможет