Поделиться через

Определение готовности информации об изменениях данных

  • Статья

Область применения: среда выполнения интеграции SSIS SQL Server в Фабрика данных Azure

В потоке управления пакета служб Integration Services, выполняющего добавочную нагрузку данных об изменениях, во-вторых, необходимо убедиться, что измененные данные для выбранного интервала готовы. Этот шаг необходим потому, что процесс асинхронной записи, возможно, еще не дошел до выбранной конечной точки.

Примечание.

Первая задача потока управления — вычислить конечные точки интервала изменений. Дополнительные сведения об этой задаче см. в разделе Задание интервала для информации об изменениях данных Описание общего процесса по проектированию потока управления см. в разделе Система отслеживания измененных данных (SSIS).

Основные сведения о компонентах решения

В решении, описанном в этом разделе, используются 4 компонента Служб Integration Services:

  • Контейнер «цикл по элементам», который регулярно оценивает выходные данные задачи «Выполнение SQL».

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

  • Компонент, обеспечивающий задержку в процессе обработки в случаях, когда данные еще не готовы. Это может быть либо задача «Скрипт», либо задача «Выполнение SQL».

  • При необходимости используется компонент, извещающий об ошибке или о времени ожидания, если задача «Выполнение SQL» возвращает значение, указывающее на ошибку либо на время ожидания.

Эти компоненты задают или считывают значения нескольких переменных пакета, чтобы управлять потоком выполнения внутри цикла и позднее в пакете.

Настройка переменных пакета

  1. В sql Server Data Tools (SSDT) в окне переменных создайте следующие переменные:

    1. Создайте переменную с типом данных integer для хранения значения состояния, возвращаемого задачей «Выполнение SQL».

      В этом примере используется переменная с именем DataReady с исходным значением 0.

    2. Создайте переменную для хранения периода времени задержки в случае, если данные не готовы. Если планируется реализовать задержку с помощью задачи «Скрипт», тип данных переменной должен быть integer. Если будет использоваться задача «Выполнение SQL» с инструкцией WAITFOR, переменная должна иметь строковый тип данных, чтобы переменная могла принимать значения типа «00:00:10».

      В этом примере используется переменная с именем DelaySeconds с исходным значением 10.

    3. Создайте переменную с типом данных integer для хранения текущей итерации цикла.

      В этом примере используется переменная с именем TimeoutCount с исходным значением 0.

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

      В этом примере используется переменная с именем TimeoutCeiling с исходным значением 20.

    5. Создайте переменную с типом данных integer, которую можно использовать для указания первой загрузки измененных данных (необязательно).

      В этом примере используется переменная с именем IntervalID и проверяется только значение 0, указывающее на первоначальную загрузку.

Настройка контейнера «цикл по элементам»

Вместе с набором переменных первым добавляемым компонентом является контейнер «цикл по элементам».

Настройка контейнера «цикл по элементам» для ожидания готовности измененных данных

  1. На вкладке "Поток управления" конструктора служб SSIS добавьте контейнер for Loop в поток управления.

  2. Соедините задачу «Выполнение SQL», вычисляющую конечные точки интервала, с контейнером «цикл по элементам».

  3. В окне Редактор циклов по элементамвыберите следующие параметры.

    1. В поле InitExpressionвведите выражение @DataReady = 0.

      Это выражение задает исходное значение для переменной цикла.

    2. В поле EvalExpressionвведите @DataReady == 0.

      Когда это выражение принимает значение False, выполнение цикла прекращается и начинается добавочная загрузка.

Настройка задачи «Выполнение SQL» для запроса об измененных данных

Внутри контейнера «цикл по элементам» добавьте задачу «Выполнение SQL». Эта задача направляет запросы к таблицам, которые обслуживаются системой отслеживания измененных данных в базах данных. Результат этого запроса — значение состояния, указывающее на готовность измененных данных.

В первом столбце следующей таблицы показаны значения, возвращаемые из задачи «Выполнение SQL» образцом запроса на языке Transact-SQL. Второй столбец показывает, как реагируют на эти значения остальные компоненты.

Возвращаемое значение Значение Response
0 Показывает, что измененные данные не готовы.

Отсутствуют записи системы отслеживания измененных данных за период после конечной точки выбранного интервала.		 Выполнение продолжает компонент, реализующий задержку. Затем управление возвращается контейнеру «цикл по элементам», который продолжает проверку задачи «Выполнение SQL» до тех пор, пока возвращается значение 0.
1 Может означать, что измененные данные не были отслежены на протяжении всего интервала, либо они были удалены. Это считается ошибкой.

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

Имеются записи системы отслеживания измененных данных как за период до начальной точки выбранного интервала, так и за период после конечной точки.		 Вычисление выходит за пределы контейнера «цикл по элементам», и начинается добавочная загрузка.
3 Указывает на первоначальную загрузку всех доступных измененных данных.

Условная логика получает это значение от специальной переменной пакета, которая используется только для этой цели.		 Вычисление выходит за пределы контейнера «цикл по элементам», и начинается добавочная загрузка.
5 Указывает на то, что достигнуто значение TimeoutCeiling.

Цикл произвел проверку на наличие данных указанное число раз, а данные все еще недоступны. Если не выполнять эту или аналогичную ей проверку, пакет может выполняться неограниченно долго.		 Выполнение продолжает дополнительный компонент, который регистрирует истечение времени ожидания.

Настройка задачи «Выполнение SQL» для запроса готовности измененных данных

  1. В пределах контейнера «цикл по элементам» добавьте задачу «Выполнение SQL».

  2. В окне Редактор задачи «Выполнение SQL»на странице Общие выберите следующие параметры:

    1. Для параметра ResultSetвыберите значение Одна строка.

    2. Настройте допустимое соединение с базой данных-источником.

    3. Для параметра SQLSourceTypeвыберите значение Прямой ввод.

    4. В поле SQLStatementвведите приведенную ниже инструкцию SQL:

      declare @DataReady int, @TimeoutCount int  

if not exists (select tran_end_time from cdc.lsn_time_mapping  
        where tran_end_time > ?  )  
    select @DataReady = 0  
else  
    if ? = 0  
        select @DataReady = 3   
else  
    if not exists (select tran_end_time from cdc.lsn_time_mapping  
            where tran_end_time <= ? )  
        select @DataReady = 1   
else  
    select @DataReady = 2  

select @TimeoutCount = ?  
if (@DataReady = 0)  
    select @TimeoutCount = @TimeoutCount + 1  
else  
    select @TimeoutCount = 0  

if (@TimeoutCount > ?)  
    select @DataReady = 5  

select @DataReady as DataReady, @TimeoutCount as TimeoutCount

  3. На странице Сопоставление параметров средства Редактор задачи «Выполнение SQL»произведите следующие сопоставления.

    1. Сопоставьте переменную ExtractEndTime с параметром 0.

    2. Сопоставьте переменную IntervalID с параметром 1.

    3. Сопоставьте переменную ExtractStartTime с параметром 2.

    4. Сопоставьте переменную TimeoutCount с параметром 3.

    5. Сопоставьте переменную TimeoutCeiling с параметром 4.

  4. На странице Результирующий набор средства Редактор задачи «Выполнение SQL»сопоставьте результат DataReady с переменной DataReady и результат TimeoutCount с переменной TimeoutCount.

Ожидание готовности измененных данных

Если измененные данные не готовы, можно использовать один из нескольких методов реализации задержки. Две следующие процедуры показывают, как реализуется задержка с помощью задачи «Скрипт» или задачи «Выполнение SQL».

Примечание.

Выполнение предварительно скомпилированных скриптов требует меньше ресурсов, чем использование задачи «Выполнение SQL».

Реализация задержки с помощью задачи «Скрипт»

  1. В пределах контейнера «цикл по элементам» добавьте задачу «Скрипт».

  2. Соедините задачу «Выполнение SQL», которая запрашивает готовность системы отслеживания измененных данных, с новой задачей «Скрипт».

  3. Для управления очередностью, соединяющего задачу «Выполнение SQL» с задачей «Скрипт», откройте Редактор управления очередностью и выберите следующие параметры:

    1. В списке Вычислительная операциявыберите пункт Выражение и ограничение.

    2. Для параметра Значениевыберите значение Успех.

      Ограничение по значению Успех относится к успешности выполнения предыдущей задачи. В этом случае это означает успешное выполнение задачи «Выполнение SQL».

    3. В поле Выражениевведите @DataReady == 0 && @TimeoutCount <= @TimeoutCeiling.

    4. Установите флажок Логическое И. Все аргументы должны иметь значение True, если он еще не выбран.

  4. В средстве Редактор задачи «Скрипт»на странице Скрипт для свойства ReadOnlyVariablesвыберите из списка целочисленную переменную User::DelaySeconds .

  5. В окне Редактор задачи «Скрипт»на странице Скрипт нажмите кнопку Изменить скрипт , чтобы открыть среду разработки скриптов.

  6. В главной процедуре введите одну из следующих строк кода.

    • При программировании на языке C# введите следующую строку кода:

      System.Threading.Thread.Sleep((int)Dts.Variables["DelaySeconds"].Value * 1000);

      - или -

    • Если вы программируются в Visual Basic, введите следующую строку кода:

      System.Threading.Thread.Sleep(Ctype(Dts.Variables("DelaySeconds").Value, Integer) * 1000)

      Примечание.

      Метод Thread.Sleep ожидает аргумент, указанный в миллисекундах.

  7. Оставьте без изменений создаваемую по умолчанию строку кода, которая возвращает DtsExecResult.Success в результате выполнения скрипта.

  8. Закройте среду разработки скриптов и Редактор задачи «Скрипт».

Реализация задержки с помощью задачи «Выполнение SQL»

  1. В пределах контейнера «цикл по элементам» добавьте задачу «Выполнение SQL».

  2. Соедините задачу «Выполнение SQL», которая направляет запросы для выявления готовности измененных данных, с новой задачей «Выполнение SQL».

  3. Для управления очередностью, которое соединяет две задачи «Выполнение SQL», откройте средство Редактор управления очередностью и выберите следующие параметры.

    1. В списке Вычислительная операциявыберите пункт Выражение и ограничение.

    2. Для параметра Значениевыберите значение Успех.

      Ограничение по значению Успешно указывает на успешное завершение предшествующей задачи «Выполнение SQL».

    3. В поле Выражениевведите @DataReady == 0.

    4. Установите флажок Логическое И. Все аргументы должны иметь значение True, если он еще не выбран.

      При выборе этого варианта оба условия (ограничение и выражение) должны иметь значение true.

  4. В окне Редактор задачи «Выполнение SQL»на странице Общие выберите следующие параметры:

    1. Для параметра ResultSetвыберите значение Одна строка.

    2. Настройте допустимое соединение с базой данных-источником.

    3. Для параметра SQLSourceTypeвыберите значение Прямой ввод.

    4. В поле SQLStatementвведите приведенную ниже инструкцию SQL:

      WAITFOR DELAY ?

  5. На странице Сопоставление параметров редактора сопоставьте строковую переменную DelaySeconds с параметром 0.

Обработка ошибки

При необходимости можно настроить внутри цикла дополнительный компонент для регистрации ошибок или превышении времени ожидания.

  • Этот компонент может регистрировать ошибку, если значение переменной DataReady = 1. Это значение указывает на то, что измененных данных за период до начала выбранного интервала нет.

  • Данный компонент может также регистрировать превышение времени ожидания по достижении указанного значения переменной TimeoutCeiling. Это значение указывает на то, что цикл выполнил проверку на наличие данных указанное число раз, но данные все еще недоступны. Если не выполнять эту или аналогичную ей проверку, пакет может выполняться неограниченно долго.

Настройка дополнительной задачи «Скрипт» для регистрации ошибки

  1. Если необходимо создать отчет об ошибке или превышении времени ожидания, записав сообщение в журнал, настройте ведение журнала для пакета. Дополнительные сведения см. в разделе Включение средств ведения журналов в SQL Server Data Tools.

  2. В пределах контейнера «цикл по элементам» добавьте задачу «Скрипт».

  3. Соедините задачу «Выполнение SQL», которая запрашивает готовность системы отслеживания измененных данных, с новой задачей «Скрипт».

  4. Для управления очередностью, соединяющего задачу «Выполнение SQL» с задачей «Скрипт», откройте Редактор управления очередностью и выберите следующие параметры:

    1. В списке Вычислительная операциявыберите пункт Выражение и ограничение.

    2. Для параметра Значениевыберите значение Успех.

      Ограничение по значению Успех относится к успешности выполнения предыдущей задачи. В этом случае это означает успешное выполнение задачи «Выполнение SQL».

    3. В поле Выражениевведите @DataReady == 1 || @DataReady == 5.

    4. Установите флажок Логическое И. Все аргументы должны иметь значение True, если он еще не выбран.

      При выборе этого варианта оба условия (ограничение и выражение) должны иметь значение true.

  5. В средстве Редактор задачи «Скрипт»на странице Скрипт для свойства ReadOnlyVariablesвыберите из списка параметры User::DataReady и User::ExtractStartTime , чтобы их значения были доступны для скрипта.

    Если необходимо включить в записываемые в журнал сведения данные из определенных системных переменных (например, System::PackageName), выберите и эти переменные.

  6. В окне Редактор задачи «Скрипт»на странице Скрипт нажмите кнопку Изменить скрипт , чтобы открыть среду разработки скриптов.

  7. В главную процедуру добавьте код для регистрации ошибки путем вызова метода Dts.Log или создайте событие, путем вызова одного из методов интерфейса Dts.Events . Информируйте пакет об ошибке, вернув Dts.TaskResult = Dts.Results.Failure.

    Следующий образец показывает, как записать сообщение в журнал. Дополнительные сведения см. в разделах Logging in the Script Task, Raising Events in the Script Taskи Returning Results from the Script Task.

    ' User variables.  
Dim dataReady As Integer = _  
  CType(Dts.Variables("DataReady").Value, Integer)  
Dim extractStartTime As Date = _  
  CType(Dts.Variables("ExtractStartTime").Value, DateTime)  

' System variables.  
Dim packageName As String = _  
  Dts.Variables("PackageName").Value.ToString()  
Dim executionStartTime As Date = _  
  CType(Dts.Variables("StartTime").Value, DateTime)  

Dim eventMessage As New System.Text.StringBuilder()  

If dataReady = 1 OrElse dataReady = 5 Then  

  If dataReady = 1 Then  
    eventMessage.AppendLine("Start Time Error")  
  Else  
    eventMessage.AppendLine("Timeout Error")  
  End If  

  With eventMessage  
    .Append("The package ")  
    .Append(packageName)  
    .Append(" started at ")  
    .Append(executionStartTime.ToString())  
    .Append(" and ended at ")  
    .AppendLine(DateTime.Now().ToString())  
    If dataReady = 1 Then  
      .Append("The specified ExtractStartTime was ")  
      .AppendLine(extractStartTime.ToString())  
    End If  
  End With  

  System.Windows.Forms.MessageBox.Show(eventMessage.ToString())  

  Dts.Log(eventMessage.ToString(), 0, Nothing)  

  Dts.TaskResult = Dts.Results.Failure  

Else  

  Dts.TaskResult = Dts.Results.Success  

End If

  8. Закройте среду разработки скриптов и Редактор задачи «Скрипт».

Следующий шаг

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

Следующий раздел: подготовка к запросу к измененным данным