Skapa funktionen för att hämta ändringsdata

Gäller för:SQL Server SSIS Integration Runtime i Azure Data Factory

När du har slutfört kontrollflödet för ett Integration Services-paket som utför en inkrementell belastning av ändringsdata är nästa uppgift att skapa en tabellvärdesfunktion (TVF) som hämtar ändringsdata. Du behöver bara skapa den här funktionen en gång före den första inkrementella inläsningen.

Designöverväganden för cdc-funktioner (change data capture)

För att hämta ändringsdata anropar en källkomponent i paketets dataflöde någon av följande frågefunktioner för ändringsdatainsamling:

• <cdc.fn_cdc_get_net_changes_capture_instance> För den här frågan innehåller den enda rad som returneras för varje uppdatering det slutliga tillståndet för varje ändrad rad. I de flesta fall behöver du bara de data som returneras av en fråga för nettoändringar. Mer information finns i cdc.fn_cdc_get_net_changes_<capture_instance> (Transact-SQL).

• <cdc.fn_cdc_get_all_changes_capture_instance> Den här frågan returnerar alla ändringar som har inträffat på varje rad under insamlingsintervallet. Mer information finns i cdc.fn_cdc_get_all_changes_<capture_instance> (Transact-SQL).

Källkomponenten tar sedan de resultat som returneras av funktionen och skickar dem till nedströmsomvandlingar och mål, som tillämpar ändringsdata på slutmålet.

En Integration Services-källkomponent kan dock inte anropa dessa funktioner för insamling av ändringsdata direkt. En Integration Services-källkomponent kräver metadata om de kolumner som frågan returnerar. Funktionerna för att hämta ändringsdata definierar inte kolumnerna i utdatatabellen. Dessa funktioner returnerar därför inte tillräckligt med metadata för en Integration Services-källkomponent.

I stället använder du en tabellvärdeshanteringsfunktion eftersom den här typen av funktion uttryckligen definierar kolumnerna i utdatatabellen i dess RETURNS-sats. Den här explicita definitionen av kolumner innehåller de metadata som en Integration Services-källkomponent behöver. Du måste skapa den här funktionen för varje tabell som du vill hämta ändringsdata för.

Du har två alternativ för att skapa den tabellvärdeshanteringsfunktion som anropar frågefunktionen för ändringsdatainsamling:

• Du kan anropa den sys.sp_cdc_generate_wrapper_function system lagrade proceduren för att skapa tabellvärdesfunktioner åt dig.

• Du kan skriva en egen tabellvärdesfunktion med hjälp av riktlinjerna och exemplet i det här avsnittet.

Anropa en lagrad procedur för att skapa funktionen table-valued

Det snabbaste och enklaste sättet att skapa de tabellvärdesfunktioner som du behöver är att anropa den sys.sp_cdc_generate_wrapper_function system lagrade proceduren. Den här lagrade proceduren genererar skript för att skapa omslutningsfunktioner som är särskilt utformade för att uppfylla behoven hos en Integration Services-källkomponent.

För att förstå hur du använder den här system lagrade proceduren bör du förstå vad proceduren gör, vilka skript proceduren genererar och vilka omslutningsfunktioner skripten skapar.

Förstå och använda den lagrade proceduren

Den sys.sp_cdc_generate_wrapper_function systemlagrade proceduren genererar skript för att skapa omslutningsfunktioner för användning av Integration Services-paket.

Här är de första raderna i definitionen av den lagrade proceduren:

CREATE PROCEDURE sys.sp_cdc_generate_wrapper_function
(
@capture_instance sysname = null
@closed_high_end_point bit = 1,
@column_list = null,
@update_flag_list = null
)

Alla parametrar för den lagrade proceduren är valfria. Om du anropar den lagrade proceduren utan att ange värden för någon av parametrarna skapar den lagrade proceduren omslutningsfunktioner för alla avbildningsinstanser som du har åtkomst till.

Den lagrade proceduren genererar alltid en omslutningsfunktion för att returnera alla ändringar från varje insamlingsinstans. Om parametern @supports_net_changes angavs när avbildningsinstansen skapades genererar den lagrade proceduren även en omslutningsfunktion för att returnera nettoändringar från varje tillämplig avbildningsinstans.

Den lagrade proceduren returnerar en resultatuppsättning med två kolumner:

• Namnet på den omslutningsfunktion som den lagrade proceduren har genererat. Den här lagrade proceduren härleder funktionsnamnet från namnet på avbildningsinstansen. (Funktionsnamnet är "fn_all_changes_" följt av namnet på capture-instansen. Prefixet som används för funktionen för nettoändringar, om den skapas, är "fn_net_changes_".)

• CREATE-instruktionen för omslutningsfunktionen.

Förstå och använda skripten som skapats av den lagrade proceduren

Vanligtvis skulle en utvecklare använda en INSERT... EXEC-instruktion för att anropa den sys.sp_cdc_generate_wrapper_function lagrade proceduren och spara skripten som den lagrade proceduren skapar i en tillfällig tabell. Varje skript kan sedan väljas individuellt och köras för att skapa motsvarande omslutningsfunktion. En utvecklare kan dock också använda en uppsättning SQL-kommandon för att köra alla CREATE-skript, som du ser i följande exempelkod:

create table #wrapper_functions  
      (function_name sysname, create_stmt nvarchar(max))  
insert into #wrapper_functions  
exec sys.sp_cdc_generate_wrapper_function  
  
declare @stmt nvarchar(max)  
declare #hfunctions cursor local fast_forward for   
      select create_stmt from #wrapper_functions  
open #hfunctions  
fetch #hfunctions into @stmt  
while (@@fetch_status <> -1)  
begin  
      exec sp_executesql @stmt  
      fetch #hfunctions into @stmt  
end  
close #hfunctions  
deallocate #hfunctions  

Förstå och använda de funktioner som skapats av den lagrade proceduren

För att systematiskt gå på tidslinjen för insamlade ändringsdata förväntar sig de genererade omslutningsfunktionerna att @end_time-parametern för ett intervall blir den @start_time parametern för det efterföljande intervallet. När den här konventionen följs kan de genererade omslutningsfunktionerna utföra följande uppgifter:

• Mappa datum-/tidsvärdena till de LSN-värden som används internt.

• Kontrollera att inga data går förlorade eller upprepas.

För att göra det enklare att fråga efter alla rader i en ändringstabell stöder de genererade omslutningsfunktionerna även följande konventioner:

• Om parametern @start_time är null använder omslutningsfunktionerna det lägsta LSN-värdet i avbildningsinstansen som frågans nedre gräns.

• Om parametern @end_time är null använder omslutningsfunktionerna det högsta LSN-värdet i avbildningsinstansen som den övre gränsen för frågan.

• Om värdet för endera parametern @start_time eller @end_time ligger utanför intervallet för tidpunkten för det lägsta respektive högsta LSN-värdet, returnerar körning av de genererade wrapperfunktionerna felkod 313: Msg 313, Level 16, State 3, Line 1 An insufficient number of arguments were supplied for the procedure or function. Det här felet bör hanteras av utvecklaren.

De flesta användare bör kunna använda de omslutningsfunktioner som den sys.sp_cdc_generate_wrapper_function system lagrade proceduren skapar utan ändringar. Men för att anpassa omslutningsfunktionerna måste du anpassa CREATE-skripten innan du kör skripten.

När paketet anropar omslutningsfunktionerna måste paketet ange värden för tre parametrar. De här tre parametrarna liknar de tre parametrar som funktionerna för att hämta ändringsdata använder. Dessa tre parametrar är följande:

• Startdatum/tid-värdet och slutdatum/tid-värdet för intervallet. Även om omslutningsfunktionerna använder datum-/tidsvärden som slutpunkter för frågeintervallet använder funktionerna för insamling av ändringsdata två LSN-värden som slutpunkter.

• Radfiltret. För både omslutningsfunktionerna och funktionerna för att hämta ändringsdata är parametern @row_filter_option densamma. Mer information finns i cdc.fn_cdc_get_all_changes_<capture_instance> (Transact-SQL) och cdc.fn_cdc_get_net_changes_<capture_instance> (Transact-SQL).

Resultatuppsättningen som returneras av omslutningsfunktionerna innehåller följande data:

• Alla begärda kolumner med ändringsdata.

• En kolumn med namnet __CDC_OPERATION som använder ett fält med ett eller två tecken för att identifiera den åtgärd som är associerad med raden. De giltiga värdena för det här fältet är följande: "I" för insert, "D" för borttagning, "UO" för att uppdatera gamla värden och "UN" för uppdatering av nya värden.

• Uppdatera flaggor, när du begär dem, som visas som bitkolumner efter åtgärdskoden och i den ordning som anges i parametern @update_flag_list . Dessa kolumner namnges genom att "_uflag" läggs till i det associerade kolumnnamnet.

Om paketet anropar en omslutningsfunktion som frågar efter alla ändringar returnerar omslutningsfunktionen även kolumnerna, __CDC_STARTLSN och __CDC_SEQVAL. Dessa två kolumner blir de första respektive andra kolumnerna i resultatuppsättningen. Omslutningsfunktionen sorterar också resultatuppsättningen baserat på dessa två kolumner.

Skriva en egen tabellvärdesfunktion

Du kan också använda SQL Server Management Studio för att skriva en egen tabellvärdeshanteringsfunktion som anropar frågefunktionen för ändringsdatainsamling och lagra funktionen för tabellvärdeshantering i SQL Server. Mer information om hur du skapar en Transact-SQL-funktion CREATE FUNCTION finns i (Transact-SQL).

I följande exempel definieras en tabellvärdesfunktion som hämtar ändringar från en kundtabell för det angivna ändringsintervallet. Den här funktionen använder funktionerna för att hämta ändringsdata för att mappa datetime-värdena till de värden för binär loggsekvensnummer (LSN) som ändringstabellerna använder internt. Den här funktionen hanterar också flera särskilda villkor:

• När ett null-värde skickas för starttiden använder den här funktionen det tidigaste tillgängliga värdet.

• När ett null-värde skickas för sluttiden använder den här funktionen det senaste tillgängliga värdet.

• När start-LSN är lika med slut-LSN, vilket vanligtvis indikerar att det inte finns några poster för det valda intervallet, avslutas den här funktionen.

Exempel på en tabellvärdesfunktion som frågar efter ändringsdata

CREATE function CDCSample.uf_Customer (  
     @start_time datetime  
    ,@end_time datetime  
)  
returns @Customer table (  
     CustomerID int  
    ,TerritoryID int  
    ,CustomerType nchar(1)  
    ,rowguid uniqueidentifier  
    ,ModifiedDate datetime  
    ,CDC_OPERATION varchar(1)  
) as  
begin  
    declare @from_lsn binary(10), @to_lsn binary(10)  
  
    if (@start_time is null)  
        select @from_lsn = sys.fn_cdc_get_min_lsn('Customer')  
    else  
        select @from_lsn = sys.fn_cdc_increment_lsn(sys.fn_cdc_map_time_to_lsn('largest less than or equal',@start_time))  
  
    if (@end_time is null)  
        select @to_lsn = sys.fn_cdc_get_max_lsn()  
    else  
        select @to_lsn = sys.fn_cdc_map_time_to_lsn('largest less than or equal',@end_time)  
  
    if (@from_lsn = sys.fn_cdc_increment_lsn(@to_lsn))  
        return  
  
    -- Query for change data  
    insert into @Customer  
    select   
        CustomerID,      
        TerritoryID,   
        CustomerType,   
        rowguid,   
        ModifiedDate,   
        case __$operation  
                when 1 then 'D'  
                when 2 then 'I'  
                when 4 then 'U'  
                else null  
         end as CDC_OPERATION  
    from   
        cdc.fn_cdc_get_net_changes_Customer(@from_lsn, @to_lsn, 'all')  
  
    return  
end   
go  
  

Hämtar ytterligare metadata med ändringsdata

Även om den användarskapade tabellvärdesfunktionen som visades tidigare endast använder kolumnen __$operation , cdc.fn_cdc_get_net_changes_<capture_instance> returnerar funktionen fyra kolumner med metadata för varje ändringsrad. Om du vill använda dessa värden i dataflödet kan du returnera dem som ytterligare kolumner från funktionen för tabellvärdesomslutning.

Mer information finns i cdc.fn_cdc_get_net_changes_<capture_instance> (Transact-SQL).

Nästa steg

När du har skapat den tabellvärdesfunktion som frågar efter ändringsdata är nästa steg att börja utforma dataflödet i paketet.

Nästa ämne:Hämta och förstå ändringsdata