Inkrementelles Kopieren von Daten aus Azure SQL-Datenbank in Blob Storage mit der Änderungsnachverfolgung im Azure-Portal

GILT FÜR: Azure Data Factory Azure Synapse Analytics

In einer Datenintegrationslösung ist das inkrementelle Laden von Daten nach anfänglichen vollständigen Ladevorgängen ein häufig verwendetes Szenario. Die geänderten Daten können für einen Zeitraum Ihres Quelldatenspeichers leicht aufgeteilt werden (z. B. LastModifyTime, CreationTime). Aber manchmal gibt es keine explizite Möglichkeit, die Deltadaten seit der letzten Verarbeitung der Daten zu identifizieren. Sie können die Technologie für die Änderungsnachverfolgung verwenden, die von Datenspeichern wie Azure SQL-Datenbank und SQL Server unterstützt wird, um die Deltadaten zu identifizieren.

In diesem Tutorial wird beschrieben, wie Sie Azure Data Factory zusammen mit der Änderungsnachverfolgung nutzen, um Deltadaten inkrementell aus Azure SQL-Datenbank in Azure Blob Storage zu laden. Weitere Informationen zur Änderungsnachverfolgung finden Sie unter Informationen zur Änderungsnachverfolgung (SQL Server).

In diesem Tutorial führen Sie die folgenden Schritte aus:

Allgemeine Lösung

In diesem Tutorial erstellen Sie zwei Pipelines, mit denen die folgenden Vorgänge durchgeführt werden.

1. Erstes Laden von historischen Daten:: Sie erstellen eine Pipeline mit einer Copy-Aktivität, bei der die gesamten Daten aus dem Quelldatenspeicher (Azure SQL-Datenbank) in den Zieldatenspeicher (Azure Blob Storage) kopiert werden:

   1. Aktivieren Sie die Technologie für die Änderungsnachverfolgung in der Quelldatenbank in Azure SQL-Datenbank.
   2. Rufen Sie den Anfangswert von SYS_CHANGE_VERSION in der Datenbank als Baseline zum Erfassen von geänderten Daten ab.
   3. Laden Sie die vollständigen Daten aus der Quelldatenbank in Azure Blob Storage.

   

2. Inkrementelles Laden von Deltadaten nach einem Zeitplan: Sie erstellen eine Pipeline mit den folgenden Aktivitäten und führen sie in regelmäßigen Abständen aus:

   1. Erstellen Sie zwei Lookup-Aktivitäten, um den alten und neuen SYS_CHANGE_VERSION-Wert aus Azure SQL-Datenbank abzurufen.

   2. Erstellen Sie eine Copy-Aktivität, um die eingefügten, aktualisierten oder gelöschten Daten (die Deltadaten) zwischen den beiden SYS_CHANGE_VERSION-Werten aus Azure SQL-Datenbank nach Azure Blob Storage zu kopieren.

      Sie laden die Deltadaten, indem Sie die Primärschlüssel von geänderten Zeilen (zwischen zwei SYS_CHANGE_VERSION-Werten) aus sys.change_tracking_tables mit Daten in der Quelltabelle verknüpfen und dann die Deltadaten auf das Ziel verschieben.

   3. Erstellen Sie eine Aktivität der gespeicherten Prozedur, um den Wert von SYS_CHANGE_VERSION für die nächste Pipelineausführung zu aktualisieren.

   

Voraussetzungen

• Azure-Abonnement. Sollten Sie kein Abonnement besitzen, können Sie ein kostenloses Konto erstellen, bevor Sie beginnen.
• Azure SQL-Datenbank. Sie verwenden eine Datenbank in Azure SQL-Datenbank als Quelldatenspeicher. Wenn Sie noch keine haben, lesen Sie Erstellen einer Datenbank in Azure SQL-Datenbank. Dort finden Sie die Schritte zu ihrer Erstellung.
• Azure-Speicherkonto. Sie verwenden den Blob Storage als Senkendatenspeicher. Wenn Sie kein Azure-Speicherkonto haben, finden Sie unter Speicherkonto erstellen die erforderlichen Schritte zum Erstellen eines solchen Kontos. Erstellen Sie einen Container mit dem Namen Adftutorial.

Erstellen einer Datenquellentabelle in Azure SQL-Datenbank

1. Öffnen Sie SQL Server Management Studio, und stellen Sie eine Verbindung mit SQL-Datenbank her.

2. Klicken Sie im Server-Explorer mit der rechten Maustaste auf Ihre Datenbank, und wählen Sie dann Neue Abfrage.

3. Führen Sie den folgenden SQL-Befehl für Ihre Datenbank aus, um eine Tabelle mit dem Namen data_source_table als Quelldatenspeicher zu erstellen.

   create table data_source_table
   (
       PersonID int NOT NULL,
       Name varchar(255),
       Age int
       PRIMARY KEY (PersonID)
   );
   INSERT INTO data_source_table
       (PersonID, Name, Age)
   VALUES
       (1, 'aaaa', 21),
       (2, 'bbbb', 24),
       (3, 'cccc', 20),
       (4, 'dddd', 26),
       (5, 'eeee', 22);
   

4. Aktivieren Sie die Änderungsnachverfolgung in Ihrer Datenbank und der Quelltabelle (data_source_table), indem Sie die folgende SQL-Abfrage ausführen.

   Hinweis

   • Ersetzen Sie <your database name> durch den Namen der Datenbank in Azure SQL-Datenbank, in der data_source_table enthalten ist.
   • Im aktuellen Beispiel werden die geänderten Daten zwei Tage lang aufbewahrt. Wenn Sie die geänderten Daten für drei oder mehr Tage laden, sind einige geänderte Daten nicht enthalten. Sie müssen entweder den Wert von CHANGE_RETENTION in eine größere Zahl ändern oder sicherstellen, dass Ihr Zeitraum zum Laden der geänderten Daten innerhalb von zwei Tagen liegt. Weitere Informationen finden Sie unter Aktivieren der Änderungsnachverfolgung für eine Datenbank.

   ALTER DATABASE <your database name>
   SET CHANGE_TRACKING = ON  
   (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON)  
   ALTER TABLE data_source_table
   ENABLE CHANGE_TRACKING  
   WITH (TRACK_COLUMNS_UPDATED = ON)
   

5. Erstellen Sie eine neue Tabelle und einen neuen Speicher namens ChangeTracking_version mit einem Standardwert, indem Sie die folgende Abfrage ausführen:

   create table table_store_ChangeTracking_version
   (
       TableName varchar(255),
       SYS_CHANGE_VERSION BIGINT,
   );
   DECLARE @ChangeTracking_version BIGINT
   SET @ChangeTracking_version = CHANGE_TRACKING_CURRENT_VERSION();  
   INSERT INTO table_store_ChangeTracking_version
   VALUES ('data_source_table', @ChangeTracking_version)
   

   Hinweis

   Wenn sich die Daten nach dem Aktivieren der Änderungsnachverfolgung für SQL-Datenbank nicht geändert haben, lautet der Wert für die Version der Änderungsnachverfolgung 0.

6. Führen Sie die folgende Abfrage zum Erstellen einer gespeicherten Prozedur in Ihrer Datenbank aus. Die Pipeline ruft diese gespeicherte Prozedur auf, um die Version der Änderungsnachverfolgung in der Tabelle zu aktualisieren, die Sie im vorherigen Schritt erstellt haben.

   CREATE PROCEDURE Update_ChangeTracking_Version @CurrentTrackingVersion BIGINT, @TableName varchar(50)
   AS
   BEGIN
   UPDATE table_store_ChangeTracking_version
   SET [SYS_CHANGE_VERSION] = @CurrentTrackingVersion
   WHERE [TableName] = @TableName
   END    
   

Erstellen einer Data Factory

1. Öffnen Sie Microsoft Edge oder Google Chrome. Derzeit wird die Data Factory-Benutzeroberfläche nur in diesen Browsern unterstützt.

2. Wählen Sie im linken Menü des Azure-Portals die Option Ressource erstellen aus.

3. Wählen Sie Integration>Data Factory aus.

   

4. Geben Sie auf der Seite Neue Data Factory den Namen ADFTutorialDataFactory ein.

   Der Name der Data Factory muss global eindeutig sein. Wenn Sie eine Fehlermeldung erhalten, die besagt, dass der ausgewählte Name nicht verfügbar ist, ändern Sie den Namen (z. B. in IhrNameADFTutorialDataFactory), und versuchen Sie noch mal, die Data Factory zu erstellen. Weitere Informationen finden Sie in den Azure Data Factory-Benennungsregeln.

5. Wählen Sie das Azure-Abonnement aus, in dem die Data Factory erstellt werden soll.

6. Führen Sie unter Ressourcengruppe einen der folgenden Schritte aus:

   • Wählen Sie die Option Vorhandene verwenden und dann in der Dropdownliste eine vorhandene Ressourcengruppe.
   • Wählen Sie Neu erstellen aus, und geben Sie dann den Namen einer Ressourcengruppe ein.

   Weitere Informationen über Ressourcengruppen finden Sie unter Verwenden von Ressourcengruppen zum Verwalten von Azure-Ressourcen.

7. Wählen Sie V2 als Version aus.

8. Wählen Sie unter Region die Region für die Data Factory aus.

   In der Dropdownliste werden nur unterstützte Standorte angezeigt. Die Datenspeicher (etwa Azure Storage und Azure SQL-Datenbank) und Computes (etwa Azure HDInsight), die eine Data Factory verwendet, können sich in anderen Regionen befinden.

9. Wählen Sie Next: Git configuration (Weiter: Git-Konfiguration) aus. Richten Sie das Repository gemäß den Anweisungen unter Konfigurationsmethode 4: Während der Erstellung einer Factory ein, oder aktivieren Sie das Kontrollkästchen Git später konfigurieren.

10. Klicken Sie auf Überprüfen + erstellen.

11. Klicken Sie auf Erstellen.

    Auf dem Dashboard wird auf der Kachel Deploying Data Factory (Data Factory wird bereitgestellt...) der Status angezeigt.

    

12. Nach Abschluss der Erstellung wird die Seite Data Factory angezeigt. Wählen Sie die Kachel Studio starten aus, um die Azure Data Factory-Benutzeroberfläche auf einer separaten Registerkarte zu öffnen.

Erstellen von verknüpften Diensten

Um Ihre Datenspeicher und Compute Services mit der Data Factory zu verknüpfen, können Sie verknüpfte Dienste in einer Data Factory erstellen. In diesem Abschnitt erstellen Sie verknüpfte Dienste mit Ihrem Azure Storage-Konto und mit Ihrer Datenbank in Azure SQL-Datenbank.

Erstellen eines verknüpften Azure Storage-Diensts

So verknüpfen Sie Ihr Speicherkonto mit der Data Factory

1. Wählen Sie auf der Data Factory-Benutzeroberfläche auf der Registerkarte Verwalten unter Verbindungen die Option Verknüpfte Dienste aus. Wählen Sie dann + Neu oder die Schaltfläche Verknüpften Dienst erstellen aus.
2. Wählen Sie im Fenster Neuer verknüpfter Dienst die Option Azure Blob Storage, und klicken Sie dann auf Weiter.
3. Geben Sie die folgenden Informationen ein:
   1. Geben Sie unter Name den Namen AzureStorageLinkedService ein.
   2. Wählen Sie für Verbindung über Integration Runtime herstellen die Integration Runtime aus.
   3. Wählen Sie für Authentifizierungstyp eine Authentifizierungsmethode aus.
   4. Wählen Sie für Speicherkontoname Ihr Azure-Speicherkonto aus.
4. Klicken Sie auf Erstellen.

Erstellen eines verknüpften Azure SQL-Datenbank-Diensts

So verknüpfen Sie Ihre Datenbank mit der Data Factory

1. Wählen Sie auf der Data Factory-Benutzeroberfläche auf der Registerkarte Verwalten unter Verbindungen die Option Verknüpfte Dienste aus. Wählen Sie dann + Neu aus.

2. Wählen Sie im Fenster New Linked Service (Neuer verknüpfter Dienst) die Option Azure SQL-Datenbank aus, und klicken Sie auf Weiter.

3. Geben Sie Folgendes ein:

   1. Geben Sie für Name den Namen AzureSqlDatabaseLinkedService ein.
   2. Wählen Sie für Servername Ihren Server aus.
   3. Wählen Sie für Datenbankname Ihre Datenbank aus.
   4. Wählen Sie für Authentifizierungstyp eine Authentifizierungsmethode aus. In diesem Tutorial wird die SQL-Authentifizierung zur Demonstration verwendet.
   5. Geben Sie für Benutzername den Namen des Benutzers ein.
   6. Geben Sie für Kennwort ein Kennwort für den Benutzer ein. Oder geben Sie die Informationen für Azure Key Vault – Verknüpfter AKV-Dienst, Geheimnisname und Geheimnisversion an.

4. Klicken Sie auf Verbindung testen, um die Verbindung zu testen.

5. Wählen Sie Erstellen aus, um den verknüpften Dienst zu erstellen.

   

Erstellen von Datasets

In diesem Abschnitt erstellen Sie Datasets zur Darstellung der Datenquelle und des Datenziels, zusammen mit dem Speicherort für die SYS_CHANGE_VERSION-Werte.

Erstellen eines Datasets zum Darstellen von Quelldaten

1. Wählen Sie auf der Data Factory-Benutzeroberfläche auf der Registerkarte Autor das Pluszeichen (+) aus. Wählen Sie dann Dataset aus, oder wählen Sie die Auslassungspunkte für Datasetaktionen aus.

   

2. Wählen Sie Azure SQL-Datenbank und dann Weiter aus.

3. Führen Sie im Fenster Eigenschaften festlegen die folgenden Schritte aus:

   1. Geben Sie für Name den Namen SourceDataset ein.
   2. Wählen Sie für Verknüpfter Dienst die Option AzureSqlDatabaseLinkedService aus.
   3. Wählen Sie für Tabellenname die Option dbo.data_source_table aus.
   4. Wählen Sie für Schema importieren die Option Aus Verbindung/Speicher aus.
   5. Klicken Sie auf OK.

   

Erstellen eines Datasets zum Darstellen von Daten, die in den Senkendatenspeicher kopiert werden

Im folgenden Verfahren erstellen Sie ein Dataset, das für die Daten steht, die aus dem Quelldatenspeicher kopiert werden. Sie haben den Container adftutorial in Azure Blob Storage erstellt, als Sie die Voraussetzungen erfüllt haben. Erstellen Sie den Container, wenn er noch nicht vorhanden ist, oder geben Sie den Namen eines bereits vorhandenen ein. In diesem Tutorial wird der Name der Ausgabedatei dynamisch aus dem Ausdruck @CONCAT('Incremental-', pipeline().RunId, '.txt') generiert.

1. Wählen Sie auf der Data Factory-Benutzeroberfläche auf der Registerkarte Autor das Pluszeichen (+) aus. Wählen Sie dann Dataset aus, oder wählen Sie die Auslassungspunkte für Datasetaktionen aus.

   

2. Wählen Sie Azure Blob Storage aus, und klicken Sie dann auf Weiter.

3. Wählen Sie das Format des Datentyps als DelimitedText aus, und klicken Sie dann auf Weiter.

4. Führen Sie im Fenster Eigenschaften festlegen die folgenden Schritte aus:

   1. Geben Sie für Name den Namen SinkDataset ein.
   2. Wählen Sie für Verknüpfter Dienst die Option AzureBlobStorageLinkedService aus.
   3. Geben Sie für Dateipfad den Pfad adftutorial/incchgtracking ein.
   4. Klicken Sie auf OK.

5. Nachdem das Dataset in der Strukturansicht angezeigt wird, wechseln Sie zur Registerkarte Verbindung, und wählen Sie das Textfeld Dateiname aus. Wenn die Option Dynamischen Inhalt hinzufügen angezeigt wird, wählen Sie sie aus.

   

6. Das Fenster Pipeline Ausdrucksgenerator wird angezeigt. Fügen Sie @concat('Incremental-',pipeline().RunId,'.csv') in das Textfeld ein.

7. Klicken Sie auf OK.

Erstellen eines Datasets zum Darstellen von Daten der Änderungsnachverfolgung

Im folgenden Verfahren erstellen Sie ein Dataset zum Speichern der Version für die Änderungsnachverfolgung. Die Tabelle table_store_ChangeTracking_version haben Sie im Rahmen der Vorbereitung erstellt.

1. Wählen Sie auf der Data Factory-Benutzeroberfläche auf der Registerkarte Autor das Pluszeichen (+) aus, und wählen Sie dann Dataset aus.
2. Wählen Sie Azure SQL-Datenbank und dann Weiter aus.
3. Führen Sie im Fenster Eigenschaften festlegen die folgenden Schritte aus:
   1. Geben Sie für Name den Namen ChangeTrackingDataset ein.
   2. Wählen Sie für Verknüpfter Dienst die Option AzureSqlDatabaseLinkedService aus.
   3. Wählen Sie für Tabellenname die Option dbo.table_store_ChangeTracking_version aus.
   4. Wählen Sie für Schema importieren die Option Aus Verbindung/Speicher aus.
   5. Klicken Sie auf OK.

Erstellen einer Pipeline für den vollständigen Kopiervorgang

Im folgenden Verfahren erstellen Sie eine Pipeline mit einer Copy-Aktivität, bei der die gesamten Daten aus dem Quelldatenspeicher (Azure SQL-Datenbank) in den Zieldatenspeicher (Azure Blob Storage) kopiert werden.

1. Wählen Sie auf der Data Factory-Benutzeroberfläche auf der Registerkarte Autor das Pluszeichen (+) aus, und wählen Sie dann Pipeline>Pipeline aus.

   

2. Eine neue Registerkarte zum Konfigurieren der Pipeline wird angezeigt. Die Pipeline wird auch in der Strukturansicht angezeigt. Ändern Sie im Eigenschaftenfenster den Namen der Pipeline in FullCopyPipeline.

3. Erweitern Sie in der Toolbox Aktivitäten die Option Move & Transform (Verschieben und transformieren). Führen Sie einen der folgenden Schritte aus:

   • Ziehen Sie die Copy-Aktivität auf die Oberfläche des Pipeline-Designers.
   • Suchen Sie auf der Suchleiste unter Aktivitäten nach der Aktivität zum Kopieren von Daten, und legen Sie den Namen auf FullCopyActivity fest.

4. Wechseln Sie zur Registerkarte Quelle, und wählen Sie für Source Dataset (Quelldataset) die Option SourceDataset aus.

5. Wechseln Sie zur Registerkarte Senke, und wählen Sie für Sink Dataset (Senkendataset) die Option SinkDataset aus.

6. Klicken Sie zum Überprüfen der Pipelinedefinition in der Symbolleiste auf Überprüfen. Vergewissern Sie sich, dass keine Validierungsfehler vorliegen. Schließen Sie die Pipelineüberprüfungsausgabe.

7. Wählen Sie zum Veröffentlichen von Entitäten (verknüpfte Dienste, Datasets und Pipelines) Alle veröffentlichen aus. Warten Sie, bis die Meldung Erfolgreich veröffentlicht angezeigt wird.

   

8. Klicken Sie zum Anzeigen von Benachrichtigungen auf die Schaltfläche Benachrichtigungen anzeigen.

Ausführen der vollständigen Kopierpipeline

1. Wählen Sie auf der Data Factory-Benutzeroberfläche auf der Symbolleiste für die Pipeline Trigger hinzufügen und dann Jetzt auslösen aus.

   

2. Klicken Sie im Fenster für die Pipelineausführung auf OK.

   

Überwachen der vollständigen Kopierpipeline

1. Wählen Sie auf der Data Factory-Benutzeroberfläche die Registerkarte Überwachen aus. Die Pipelineausführung und der zugehörige Status werden in der Liste angezeigt. Klicken Sie zum Aktualisieren der Liste auf Aktualisieren. Zeigen Sie mit der Maus auf die Pipelineausführung, um die Option Erneut ausführen oder Verbrauch zu erhalten.

   

2. Wählen Sie in der Spalte Pipelinename den Namen der Pipeline aus, um mit der Pipelineausführung verknüpfte Aktivitätsausführungen anzuzeigen. Da die Pipeline nur eine Aktivität enthält, enthält die Liste nur einen Eintrag. Wählen Sie oben den Link Alle Pipelineausführungen aus, um zurück zur Ansicht mit den Pipelineausführungen zu wechseln.

Überprüfen der Ergebnisse

Der Ordner incchgtracking des Containers adftutorial enthält eine Datei namens incremental-<GUID>.csv.

Die Datei sollte die Daten aus Ihrer Datenbank enthalten:

PersonID,Name,Age
1,"aaaa",21
2,"bbbb",24
3,"cccc",20
4,"dddd",26
5,"eeee",22

5,eeee,PersonID,Name,Age
1,"aaaa",21
2,"bbbb",24
3,"cccc",20
4,"dddd",26
5,"eeee",22

Hinzufügen von weiteren Daten zur Quelltabelle

Führen Sie die folgende Abfrage für Ihre Datenbank aus, um eine Zeile hinzuzufügen und eine Zeile zu aktualisieren:

INSERT INTO data_source_table
(PersonID, Name, Age)
VALUES
(6, 'new','50');


UPDATE data_source_table
SET [Age] = '10', [name]='update' where [PersonID] = 1

Erstellen einer Pipeline für die Deltakopie

Im folgenden Verfahren erstellen Sie eine Pipeline mit Aktivitäten und führen sie regelmäßig aus. Beim Ausführen der Pipeline geschieht Folgendes:

• Mit den Lookup-Aktivitäten werden der alte und der neue SYS_CHANGE_VERSION-Wert aus Azure SQL-Datenbank abgerufen und an die Copy-Aktivität übergeben.
• Die Copy-Aktivität kopiert die eingefügten, aktualisierten oder gelöschten Daten zwischen den beiden SYS_CHANGE_VERSION-Werten aus Azure SQL-Datenbank nach Azure Blob Storage.
• Die Aktivität der gespeicherten Prozedur aktualisiert den Wert von SYS_CHANGE_VERSION für die nächste Pipelineausführung.

1. Wechseln Sie auf der Data Factory-Benutzeroberfläche zur Registerkarte Autor, wählen Sie das Pluszeichen (+) und dann Pipeline>Pipeline aus.

   

2. Eine neue Registerkarte zum Konfigurieren der Pipeline wird angezeigt. Die Pipeline wird auch in der Strukturansicht angezeigt. Ändern Sie im Eigenschaftenfenster den Namen der Pipeline in IncrementalCopyPipeline.

3. Erweitern Sie in der Toolbox Aktivitäten die Option Allgemein. Ziehen Sie die Lookup-Aktivität auf die Oberfläche des Pipeline-Designers, oder suchen Sie im Feld Nach Aktivitäten suchen. Legen Sie den Namen der Aktivität auf LookupLastChangeTrackingVersionActivity fest. Mit dieser Aktivität wird die Version für die Änderungsnachverfolgung des letzten Kopiervorgangs abgerufen, die in der Tabelle table_store_ChangeTracking_version gespeichert ist.

4. Wechseln Sie im Fenster Eigenschaften zur Registerkarte Einstellungen. Wählen Sie für Source Dataset (Quelldataset) die Option ChangeTrackingDataset aus.

5. Ziehen Sie die Lookup-Aktivität aus der Toolbox Aktivitäten auf die Oberfläche des Pipeline-Designers. Legen Sie den Namen der Aktivität auf LookupCurrentChangeTrackingVersionActivity fest. Mit dieser Aktivität wird die aktuelle Version der Änderungsnachverfolgung abgerufen.

6. Wechseln Sie im Fenster Eigenschaften zur Registerkarte Einstellungen, und führen Sie dann die folgenden Schritte aus:

   1. Wählen Sie für Source Dataset (Quelldataset) die Option SourceDataset.

   2. Wählen Sie für Abfrage verwenden die Option Abfrage aus.

   3. Geben Sie für Abfrage die folgende SQL-Abfrage ein:

      SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion
      

   

7. Erweitern Sie in der Toolbox Aktivitäten die Option Move & Transform (Verschieben und transformieren). Ziehen Sie die Aktivität zum Kopieren von Daten auf die Oberfläche des Pipeline-Designers. Legen Sie den Namen der Aktivität auf IncrementalCopyActivity fest. Mit dieser Aktivität werden die Daten, die zwischen der letzten Version und der aktuellen Version der Änderungsnachverfolgung angefallen sind, in den Zieldatenspeicher kopiert.

8. Wechseln Sie im Fenster Eigenschaften zur Registerkarte Quelle, und führen Sie dann die folgenden Schritte aus:

   1. Wählen Sie für Source Dataset (Quelldataset) die Option SourceDataset.

   2. Wählen Sie für Abfrage verwenden die Option Abfrage aus.

   3. Geben Sie für Abfrage die folgende SQL-Abfrage ein:

      SELECT data_source_table.PersonID,data_source_table.Name,data_source_table.Age, CT.SYS_CHANGE_VERSION, SYS_CHANGE_OPERATION from data_source_table RIGHT OUTER JOIN CHANGETABLE(CHANGES data_source_table, @{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.SYS_CHANGE_VERSION}) AS CT ON data_source_table.PersonID = CT.PersonID where CT.SYS_CHANGE_VERSION <= @{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}
      

   

9. Wechseln Sie zur Registerkarte Senke, und wählen Sie für Sink Dataset (Senkendataset) die Option SinkDataset aus.

10. Verbinden Sie beide Lookup-Aktivitäten nacheinander mit der Copy-Aktivität. Ziehen Sie die grüne Schaltfläche, die der Lookup-Aktivität zugeordnet ist, auf die Copy-Aktivität.

11. Ziehen Sie die Aktivität der gespeicherten Prozedur aus der Toolbox Aktivitäten in die Oberfläche des Pipeline-Designers. Legen Sie den Namen der Aktivität auf StoredProceduretoUpdateChangeTrackingActivity fest. Mit dieser Aktivität wird die Version der Änderungsnachverfolgung in der Tabelle table_store_ChangeTracking_version geändert.

12. Wechseln Sie zur Registerkarte Einstellungen, und führen Sie dann die folgenden Schritte aus:

    1. Wählen Sie für Verknüpfter Dienst die Option AzureSqlDatabaseLinkedService aus.
    2. Wählen Sie unter Name der gespeicherten Prozedur den Namen Update_ChangeTracking_Version.
    3. Wählen Sie Importieren aus.
    4. Geben Sie im Abschnitt Parameter der gespeicherten Prozedur die folgenden Werte für die Parameter an:

    Name	type	Wert
    CurrentTrackingVersion	Int64	@{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}
    TableName	String	@{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.TableName}

    

13. Verbinden Sie die Copy-Aktivität mit der Aktivität der gespeicherten Prozedur. Ziehen Sie die grüne Schaltfläche, die an die Copy-Aktivität angefügt ist, auf die Aktivität der gespeicherten Prozedur.

14. Klicken Sie in der Symbolleiste auf Überprüfen. Vergewissern Sie sich, dass keine Validierungsfehler vorliegen. Schließen Sie das Fenster Pipeline Validation Report (Pipelineüberprüfungsbericht).

15. Veröffentlichen Sie Entitäten (verknüpfte Dienste, Datasets und Pipelines) für den Data Factory-Dienst, indem Sie die Schaltfläche Alle veröffentlichen auswählen. Warten Sie, bis die Meldung Veröffentlichung erfolgreich angezeigt wird.

    

Ausführen der inkrementellen Kopierpipeline

1. Wählen Sie auf der Symbolleiste für die Pipeline die Option Trigger hinzufügen und dann Trigger Now (Jetzt auslösen) aus.

   

2. Wählen Sie im Fenster Pipelineausführung die Option OK aus.

Überwachen der inkrementellen Kopierpipeline

1. Wählen Sie die Registerkarte Überwachen aus. Die Pipelineausführung und der zugehörige Status werden in der Liste angezeigt. Klicken Sie zum Aktualisieren der Liste auf Aktualisieren.

   

2. Wählen Sie in der Spalte Pipelinename den Link IncrementalCopyPipeline aus, um mit der Pipelineausführung verknüpfte Aktivitätsausführungen anzuzeigen. Die Aktivitätsausführungen werden in einer Liste angezeigt.

   

Überprüfen der Ergebnisse

Die zweite Datei wird im Ordner incchgtracking des Containers adftutorial angezeigt.

Die Datei sollte nur die Deltadaten aus Ihrer Datenbank enthalten. Der Datensatz mit U ist die aktualisierte Zeile in der Datenbank, und mit I wird die hinzugefügte Zeile angegeben.

PersonID,Name,Age,SYS_CHANGE_VERSION,SYS_CHANGE_OPERATION
1,update,10,2,U
6,new,50,1,I

Die ersten drei Spalten enthalten geänderte Daten aus data_source_table. Die letzten beiden Spalten enthalten die Metadaten aus der Tabelle für das Änderungsnachverfolgungssystem. Die vierte Spalte enthält den SYS_CHANGE_VERSION-Wert für die einzelnen geänderten Zeilen. Die fünfte Spalte enthält den Vorgang: U = update (aktualisieren), I = insert (einfügen). Weitere Informationen zu den Informationen zur Änderungsnachverfolgung finden Sie unter CHANGETABLE.

==================================================================
PersonID Name    Age    SYS_CHANGE_VERSION    SYS_CHANGE_OPERATION
==================================================================
1        update  10            2                                 U
6        new     50	       1                            	 I

Zugehöriger Inhalt

Im folgenden Tutorial erfahren Sie mehr über das Kopieren nur von neuen und geänderten Dateien auf Grundlage von LastModifiedDate: