Übung: Entwerfen und Implementieren einer langsam veränderlichen Dimension vom Typ 1 mit Zuordnungsdatenflüssen

  • 10 Minuten

In dieser Übung erstellen Sie einen Datenfluss für eine langsam veränderliche Dimension vom Typ 1 mithilfe eines dedizierten SQL-Pools von Azure Synapse als Quelle und Ziel. Dieser Datenfluss kann anschließend zu einer Synapse-Pipeline hinzugefügt und im Rahmen des ETL-Prozesses (Extrahieren, Transformieren und Laden) ausgeführt werden.

Einrichten der Quell- und Dimensionstabelle

Für diese Übung müssen Sie eine Dimensionstabelle aus Quelldaten in Azure Synapse laden, die von vielen verschiedenen Systemtypen wie Azure SQL, Azure Storage und mehr stammen könnten. Dieses Beispiel ist jedoch relativ einfach gehalten, da Sie die Quelldaten in Ihrer Azure Synapse-Datenbank erstellen.

  1. Navigieren Sie in Synapse Studio zum Hub Daten.

    Data hub.

  2. Wählen Sie die Registerkarte Arbeitsbereich(1) aus, erweitern Sie „Datenbanken“, und klicken Sie dann mit der rechten Maustaste auf SQLPool01 (2). Wählen Sie Neues SQL-Skript (3) und dann Leeres Skript (4) aus.

    The data hub is displayed with the context menus to create a new SQL script.

  3. Fügen Sie das folgende Skript in das leere Skriptfenster ein, und klicken Sie dann auf Ausführen, oder drücken Sie F5, um die Abfrage auszuführen:

    CREATE TABLE [dbo].[CustomerSource] (
    [CustomerID] [int] NOT NULL,
    [Title] [nvarchar](8),
    [FirstName] [nvarchar](50),
    [MiddleName] [nvarchar](50),
    [LastName] [nvarchar](50),
    [Suffix] [nvarchar](10),
    [CompanyName] [nvarchar](128),
    [SalesPerson] [nvarchar](256),
    [EmailAddress] [nvarchar](50),
    [Phone] [nvarchar](25)
) WITH ( HEAP )

COPY INTO [dbo].[CustomerSource]
FROM 'https://solliancepublicdata.blob.core.windows.net/dataengineering/dp-203/awdata/CustomerSource.csv'
WITH (
    FILE_TYPE='CSV',
    FIELDTERMINATOR='|',
    FIELDQUOTE='',
    ROWTERMINATOR='0x0a',
    ENCODING = 'UTF16'
)

CREATE TABLE dbo.[DimCustomer](
    [CustomerID] [int] NOT NULL,
    [Title] [nvarchar](8) NULL,
    [FirstName] [nvarchar](50) NOT NULL,
    [MiddleName] [nvarchar](50) NULL,
    [LastName] [nvarchar](50) NOT NULL,
    [Suffix] [nvarchar](10) NULL,
    [CompanyName] [nvarchar](128) NULL,
    [SalesPerson] [nvarchar](256) NULL,
    [EmailAddress] [nvarchar](50) NULL,
    [Phone] [nvarchar](25) NULL,
    [InsertedDate] [datetime] NOT NULL,
    [ModifiedDate] [datetime] NOT NULL,
    [HashKey] [char](64)
)
WITH
(
    DISTRIBUTION = REPLICATE,
    CLUSTERED COLUMNSTORE INDEX
)

    The script and Run button are both highlighted.

Erstellen eines Zuordnungsdatenflusses

Zuordnungsdatenflüsse sind Pipelineaktivitäten, die auf visuelle Weise angeben, wie Daten transformiert werden sollen, und zwar ohne Code. Im nächsten Schritt erstellen Sie einen Zuordnungsdatenfluss, um eine langsam veränderliche Dimension vom Typ 1 zu erstellen.

  1. Navigieren Sie zum Hub Entwickeln.

    Develop hub.

  2. Wählen Sie + und dann Datenfluss aus.

    The plus button and data flow menu item are highlighted.

  3. Geben Sie im Eigenschaftenbereich des neuen Datenflusses UpdateCustomerDimension in das Feld Name(1) ein, klicken Sie dann auf die Schaltfläche Eigenschaften(2), um den Eigenschaftenbereich auszublenden.

    The data flow properties pane is displayed.

  4. Klicken Sie auf dem Canvas auf Quelle hinzufügen.

    The Add Source button is highlighted on the data flow canvas.

  5. Konfigurieren Sie unter Source settings die folgenden Eigenschaften:

    • Ausgabestreamname: Geben Sie SourceDB ein.
    • Quelltyp: Wählen Sie Dataset aus.
    • Optionen: Aktivieren Sie das Kontrollkästchen Allow schema drift, und lassen Sie die anderen Optionen deaktiviert.
    • Sampling: Wählen Sie Disable aus.
    • Dataset: Klicken Sie auf + Neu, um ein neues Dataset zu erstellen.

    The New button is highlighted next to Dataset.

  6. Wählen Sie im Dialogfeld „Neues Integrationsdataset“ die Option Azure Synapse Analytics aus, und klicken Sie dann auf Weiter.

    Azure SQL Database and the Continue button are highlighted.

  7. Konfigurieren Sie in den Dataseteigenschaften Folgendes:

    • Name: Geben Sie CustomerSource ein.
    • Verknüpfter Dienst: Wählen Sie den verknüpften Dienst für den Synapse-Arbeitsbereich aus.
    • Tabellenname: Klicken Sie neben dem Dropdownfeld auf Aktualisieren.

    The form is configured as described and the refresh button is highlighted.

  8. Geben Sie im Feld Wert den Namen Ihres SQL-Pools ein, und klicken Sie dann auf OK.

    The SQLPool01 parameter is highlighted.

  9. Wählen Sie unter Tabellenname die Option dbo.CustomerSource aus, wählen Sie unter Importschema die Option From connection/store aus, und klicken Sie dann auf OK, um das Dataset zu erstellen.

    The form is completed as described.

  10. Klicken Sie neben dem Dataset CustomerSource, das Sie hinzugefügt haben, auf Öffnen.

    The open button is highlighted next to the new dataset.

  11. Geben Sie den Namen Ihres SQL-Pools in das Feld Wert neben DBName ein.

  12. Klicken Sie im Datenfluss-Editor auf das Feld Quelle hinzufügen unter der SourceDB-Aktivität. Konfigurieren Sie diese Quelle als DimCustomer-Tabelle, indem Sie dieselben Schritte ausführen, die Sie für CustomerSource ausgeführt haben.

    • Ausgabestreamname: Geben Sie DimCustomer ein.
    • Quelltyp: Wählen Sie Dataset aus.
    • Optionen: Aktivieren Sie das Kontrollkästchen Allow schema drift, und lassen Sie die anderen Optionen deaktiviert.
    • Sampling: Wählen Sie Disable aus.
    • Dataset: Klicken Sie auf + Neu, um ein neues Dataset zu erstellen. Verwenden Sie den verknüpften Dienst für Azure Synapse, und wählen Sie die Tabelle „DimCustomer“ aus. Stellen Sie sicher, dass Sie den Namen Ihres SQL-Pools für „DBName“ angeben.

    The Add Source, Output stream name, and Dataset name are highlighted in the Source settings.

Hinzufügen von Transformationen zum Datenfluss

  1. Klicken Sie rechts neben der Quelle SourceDB auf dem Canvas auf +, und wählen Sie dann die Option Abgeleitete Spalte aus.

    The plus button and derived column menu item are highlighted.

  2. Konfigurieren Sie unter Derived column's settings die folgenden Eigenschaften:

    • Ausgabestreamname: Geben Sie CreateCustomerHash ein.
    • Eingehender Stream: Wählen Sie SourceDB aus.
    • Spalten: Geben Sie Folgendes ein:
    Column Ausdruck BESCHREIBUNG
    Eingabe in HashKey sha2(256, iifNull(Title,'') +FirstName +iifNull(MiddleName,'') +LastName +iifNull(Suffix,'') +iifNull(CompanyName,'') +iifNull(SalesPerson,'') +iifNull(EmailAddress,'') +iifNull(Phone,'')) Hiermit wird ein SHA256-Hash der Tabellenwerte erstellt. Dies wird verwendet, um Zeilenänderungen zu erkennen, indem der Hash der eingehenden Datensätze mit dem Hashwert der Zieldatensätze verglichen wird, der mit dem CustomerID-Wert übereinstimmt. Die Funktion iifNull ersetzt NULL-Werte durch leere Zeichenfolgen. Andernfalls werden die Hashwerte in der Regel dupliziert, wenn Einträge mit NULL-Werten vorhanden sind.

    The Derived column's settings form is configured as described.

  3. Klicken Sie rechts neben der abgeleiteten Spalte + auf dem Canvas auf CreateCustomerHash, und wählen Sie dann die Option Exists (Vorhanden) aus.

    The plus button and exists menu item are both highlighted.

  4. Konfigurieren Sie unter Exists settings die folgenden Eigenschaften:

    • Ausgabestreamname: Geben Sie Exists ein.
    • Linker Stream: Wählen Sie CreateCustomerHash aus.
    • Rechter Stream: Wählen Sie SynapseDimCustomer aus.
    • Typ für Vorhandensein: Wählen Sie Doesn't exist aus.
    • Exists-Bedingungen: Legen Sie Folgendes für links und rechts fest:
    Links: Spalte von CreateCustomerHash Rechts: Spalte von SynapseDimCustomer
    HashKey HashKey

    The Exists settings form is configured as described.

  5. Wählen Sie rechts von Exists auf dem Canvas + und dann Nachschlagen aus.

    The plus button and lookup menu item are both highlighted.

  6. Konfigurieren Sie unter Lookup settings die folgenden Eigenschaften:

    • Ausgabestreamname: Geben Sie LookupCustomerID ein.
    • Primärer Stream: Wählen Sie Exists aus.
    • Suchstream: Wählen Sie SynapseDimCustomer aus.
    • Mehrere Zeilen abgleichen: Deaktiviert
    • Match on (Übereinstimmung mit): Wählen Sie Any row aus.
    • Suchbedingungen: Legen Sie Folgendes für links und rechts fest:
    Links: Spalte von Exists Rechts: Spalte von SynapseDimCustomer
    CustomerID CustomerID

    The Lookup settings form is configured as described.

  7. Wählen Sie rechts von LookupCustomerID auf dem Canvas + und dann Abgeleitete Spalte aus.

    The plus button and derived column menu item are both highlighted.

  8. Konfigurieren Sie unter Derived column's settings die folgenden Eigenschaften:

    • Ausgabestreamname: Geben Sie SetDates ein.
    • Eingehender Stream: Wählen Sie LookupCustomerID aus.
    • Spalten: Geben Sie Folgendes ein:
    Column Ausdruck BESCHREIBUNG
    Wählen Sie InsertedDate aus. iif(isNull(InsertedDate), currentTimestamp(), {InsertedDate}) Wenn der Wert InsertedDate NULL ist, fügen Sie den aktuellen Zeitstempel ein. Verwenden Sie andernfalls den Wert InsertedDate.
    Wählen Sie ModifiedDate aus. currentTimestamp() Aktualisieren Sie den Wert ModifiedDate immer mit dem aktuellen Zeitstempel.

    Another Derived column's settings form is configured as described.

    Hinweis

    Klicken Sie über der Liste „Spalten“ auf + Hinzufügen, und wählen Sie dann Spalte hinzufügen aus, um die zweite Spalte einzufügen.

  9. Wählen Sie rechts von der abgeleiteten Spalte SetDates auf dem Canvas + und dann Zeile ändern aus.

    The plus button and alter row menu item are both highlighted.

  10. Konfigurieren Sie unter Alter row settings die folgenden Eigenschaften:

    • Ausgabestreamname: Geben Sie AllowUpserts ein.
    • Eingehender Stream: Wählen Sie SetDates aus.
    • Bedingungen für Zeilenänderung: Geben Sie Folgendes ein:
    Bedingung Ausdruck BESCHREIBUNG
    Wählen Sie Upsert if aus. true() Legen Sie die Bedingung auf true() für die Bedingung Upsert if fest, um Upserts zuzulassen. Dadurch wird sichergestellt, dass alle Daten, die die Schritte im Zuordnungsdatenfluss durchlaufen, in die Senke eingefügt oder geändert werden.

    The alter row settings form is configured as described.

  11. Wählen Sie rechts neben dem Schritt AllowUpserts für „Zeile ändern“ auf der Canvas + und dann Senke aus.

    The plus button and sink menu item are both highlighted.

  12. Konfigurieren Sie unter Sink die folgenden Eigenschaften:

    • Ausgabestreamname: Geben Sie Sink ein.
    • Eingehender Stream: Wählen Sie AllowUpserts aus.
    • Senkentyp: Wählen Sie Dataset aus.
    • Dataset: Wählen Sie DimCustomer aus.
    • Optionen: Aktivieren Sie Allow schema drift, und deaktivieren Sie Validate schema.

    The sink properties form is configured as described.

  13. Wählen Sie die Registerkarte Einstellungen aus, und konfigurieren Sie die folgenden Eigenschaften:

    • Updatemethode: Aktivieren Sie Allow upsert, und deaktivieren Sie alle anderen Optionen.
    • Schlüsselspalten: Klicken Sie auf List of columns, und wählen Sie CustomerID in der Liste aus.
    • Tabellenaktion: Wählen Sie None aus.
    • Staging aktivieren: Deaktiviert

    The sink settings are configured as described.

  14. Wählen Sie die Registerkarte Zuordnung aus, und deaktivieren Sie Auto mapping (automatische Zuordnung). Konfigurieren Sie die Zuordnung der Eingabespalten wie unten beschrieben:

    Eingabespalten Ausgabespalten
    SourceDB@CustomerID CustomerID
    SourceDB@Title Title
    SourceDB@FirstName FirstName
    SourceDB@MiddleName MiddleName
    SourceDB@LastName LastName
    SourceDB@Suffix Suffix
    SourceDB@CompanyName CompanyName
    SourceDB@SalesPerson SalesPerson
    SourceDB@EmailAddress EmailAddress
    SourceDB@Phone Phone
    InsertedDate InsertedDate
    ModifiedDate ModifiedDate
    CreateCustomerHash@HashKey HashKey

    Mapping settings are configured as described.

  15. Der vollständige Zuordnungsfluss sollte in etwa wie folgt aussehen. Klicken Sie auf Alle veröffentlichen, um Ihre Änderungen zu speichern.

    The completed data flow is displayed and Publish all is highlighted.

  16. Wählen Sie Veröffentlichen aus.

    The publish button is highlighted.

Testen des Datenflusses

Sie haben einen Datenfluss für langsam veränderliche Dimensionen vom Typ 1 fertiggestellt. Wenn Sie diesen testen möchten, können Sie den Datenfluss zu einer Synapse-Integrationspipeline hinzufügen. Anschließend können Sie die Pipeline einmal ausführen, um den anfänglichen Ladevorgang für die Quelldaten der Kunden in das DimCustomer-Ziel durchzuführen.

Bei jeder zusätzlichen Ausführung der Pipeline werden die Daten in der Quelltabelle mit den bereits in der Dimensionstabelle vorhandenen Daten verglichen (mithilfe des Hashschlüssels) und nur Datensätze werden aktualisiert, die geändert wurden. Dies können Sie testen, indem Sie einen Datensatz in der Quelltabelle ändern und dann die Pipeline noch mal ausführen. Anschließend überprüfen Sie die Datensatzänderungen in der Dimensionstabelle.

Sehen Sie sich beispielsweise den Kunden Janet Gates an. Der anfängliche Ladevorgang zeigt für LastName den Wert „Gates“ und für CustomerId den Wert „4“ an.

The script is displayed with the initial customer record.

Hier sehen Sie eine Beispielanweisung, die den Nachnamen des Kunden in der Quelltabelle ändert.

UPDATE [dbo].[CustomerSource]
SET LastName = 'Lopez'
WHERE [CustomerId] = 4

Nach der Änderung des Datensatzes und einer neuen Ausführung der Pipeline, zeigt DimCustomer die folgenden aktualisierten Daten an.

The script is displayed with the updated customer record.

Der Wert LastName im Kundendatensatz wurde erfolgreich geändert, sodass er dem Quelldatensatz entspricht. Außerdem wurde ModifiedDate aktualisiert, ohne dass der alte LastName-Wert aufgezeichnet wurde. Dies entspricht dem erwarteten Verhalten für langsam veränderliche Dimensionen vom Typ 1. Wenn der Verlauf für das Feld LastName erforderlich ist, müssen Sie die Tabelle und den Datenfluss so anpassen, dass sie einem der anderen Typen langsam veränderlicher Dimensionen aufweisen, die Sie kennengelernt haben.