Übung: Entwerfen und Implementieren einer langsam veränderlichen Dimension vom Typ 1 mit Zuordnungsdatenflüssen
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.
Navigieren Sie in Synapse Studio zum Hub Daten.
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.
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 )
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.
Navigieren Sie zum Hub Entwickeln.
Wählen Sie + und dann Datenfluss aus.
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.Klicken Sie auf dem Canvas auf Quelle hinzufügen.
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.
- Ausgabestreamname: Geben Sie
Wählen Sie im Dialogfeld „Neues Integrationsdataset“ die Option Azure Synapse Analytics aus, und klicken Sie dann auf Weiter.
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.
- Name: Geben Sie
Geben Sie im Feld Wert den Namen Ihres SQL-Pools ein, und klicken Sie dann auf OK.
Wählen Sie unter Tabellenname die Option
dbo.CustomerSource
aus, wählen Sie unter Importschema die OptionFrom connection/store
aus, und klicken Sie dann auf OK, um das Dataset zu erstellen.Klicken Sie neben dem Dataset
CustomerSource
, das Sie hinzugefügt haben, auf Öffnen.Geben Sie den Namen Ihres SQL-Pools in das Feld Wert neben
DBName
ein.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.
- Ausgabestreamname: Geben Sie
Hinzufügen von Transformationen zum Datenfluss
Klicken Sie rechts neben der Quelle
SourceDB
auf dem Canvas auf +, und wählen Sie dann die Option Abgeleitete Spalte aus.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 FunktioniifNull
ersetzt NULL-Werte durch leere Zeichenfolgen. Andernfalls werden die Hashwerte in der Regel dupliziert, wenn Einträge mit NULL-Werten vorhanden sind.- Ausgabestreamname: Geben Sie
Klicken Sie rechts neben der abgeleiteten Spalte + auf dem Canvas auf
CreateCustomerHash
, und wählen Sie dann die Option Exists (Vorhanden) aus.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
- Ausgabestreamname: Geben Sie
Wählen Sie rechts von
Exists
auf dem Canvas + und dann Nachschlagen aus.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
- Ausgabestreamname: Geben Sie
Wählen Sie rechts von
LookupCustomerID
auf dem Canvas + und dann Abgeleitete Spalte aus.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 WertInsertedDate
.Wählen Sie ModifiedDate
aus.currentTimestamp()
Aktualisieren Sie den Wert ModifiedDate
immer mit dem aktuellen Zeitstempel.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.
- Ausgabestreamname: Geben Sie
Wählen Sie rechts von der abgeleiteten Spalte
SetDates
auf dem Canvas + und dann Zeile ändern aus.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 BedingungUpsert 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.- Ausgabestreamname: Geben Sie
Wählen Sie rechts neben dem Schritt
AllowUpserts
für „Zeile ändern“ auf der Canvas + und dann Senke aus.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 SieValidate schema
.
- Ausgabestreamname: Geben Sie
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 SieCustomerID
in der Liste aus. - Tabellenaktion: Wählen Sie
None
aus. - Staging aktivieren: Deaktiviert
- Updatemethode: Aktivieren Sie
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
Der vollständige Zuordnungsfluss sollte in etwa wie folgt aussehen. Klicken Sie auf Alle veröffentlichen, um Ihre Änderungen zu speichern.
Wählen Sie Veröffentlichen aus.
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.
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.
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.