Gyakorlat – 1. típusú, lassan változó dimenzió tervezése és implementálása adatfolyamok leképezésével

  • 10 perc

Ebben a gyakorlatban egy 1. típusú SCD-hez hoz létre adatfolyamot az Azure Synapse dedikált SQL-készletének forrásként és célként való használatával. Ez az adatfolyam ezután hozzáadható egy Synapse-folyamathoz, és a kinyerési, átalakítási, betöltési (ETL) folyamat részeként futtatható.

Forrás- és dimenziótábla beállítása

Ebben a gyakorlatban egy dimenziótáblát szeretne betölteni az Azure Synapse-ban olyan forrásadatokból, amelyek számos különböző rendszertípusból származhatnak, például az Azure SQL-ből, az Azure Storage-ból stb. Ebben a példában egyszerűnek tarthatja a forrásadatokat az Azure Synapse-adatbázisban.

  1. A Synapse Studióban lépjen a Data Hubra.

    Data hub.

  2. Válassza a Munkaterület lapot (1), bontsa ki az Adatbázisok elemet, majd kattintson a jobb gombbal az SQLPool01 (2) elemre. Válassza az Új SQL-szkript (3), majd az Üres szkript (4) lehetőséget.

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

  3. Illessze be a következő szkriptet az üres szkriptablakba, majd válassza a Futtatás vagy a Találat F5 lehetőséget a lekérdezés végrehajtásához:

    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.

Leképezési adatfolyam létrehozása

Az adatfolyamok leképezése olyan folyamattevékenységek, amelyek vizuális módot biztosítanak az adatok kód nélküli átalakításának megadására. Ezután létrehoz egy leképezési adatfolyamot egy 1. típusú SCD létrehozásához.

  1. Lépjen a Fejlesztési központhoz.

    Develop hub.

  2. Válassza ki +az adatfolyamot, majd válassza az Adatfolyam lehetőséget.

    The plus button and data flow menu item are highlighted.

  3. Az új adatfolyam tulajdonságok paneljén adja meg a Név mezőt (1), majd a Tulajdonságok gombra (2) kattintva elrejtheti a tulajdonságok panelt.UpdateCustomerDimension

    The data flow properties pane is displayed.

  4. Válassza a Forrás hozzáadása lehetőséget a vásznon.

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

  5. A csoportban Source settingskonfigurálja a következő tulajdonságokat:

    • Kimeneti stream neve: Enter SourceDB
    • Forrás típusa: Kiválasztás Dataset
    • Beállítások: Jelölje be Allow schema drift a többi beállítást, és hagyja bejelölve
    • Mintavételezés: Kiválasztás Disable
    • Adatkészlet: Új adatkészlet létrehozásához válassza az + Új lehetőséget

    The New button is highlighted next to Dataset.

  6. Az új integrációs adathalmaz párbeszédpanelen válassza az Azure Synapse Analytics, majd a Folytatás lehetőséget.

    Azure SQL Database and the Continue button are highlighted.

  7. Az adathalmaz tulajdonságaiban konfigurálja a következőket:

    • Név: Enter CustomerSource
    • Társított szolgáltatás: Válassza ki a Synapse-munkaterület társított szolgáltatását
    • Tábla neve: Válassza a Legördülő lista Melletti Frissítés gombot

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

  8. Az Érték mezőben adja meg az SQL-készlet nevét, majd kattintson az OK gombra.

    The SQLPool01 parameter is highlighted.

  9. Válassza a Táblanév területen, válassza az From connection/store Importálás séma területen, majd az OK gombra kattintva hozza létre az dbo.CustomerSource adathalmazt.

    The form is completed as described.

  10. Válassza a Megnyitás gombot a CustomerSource hozzáadott adathalmaz mellett.

    The open button is highlighted next to the new dataset.

  11. Adja meg az SQL-készlet nevét a mellette DBNamelévő Érték mezőben.

  12. Az adatfolyam-szerkesztőben válassza a Forrás hozzáadása mezőt a SourceDB-tevékenység alatt. Konfigurálja ezt a forrást DimCustomer-táblaként a CustomerSource-hoz használt lépéseket követve.

    • Kimeneti stream neve: Enter DimCustomer
    • Forrás típusa: Kiválasztás Dataset
    • Beállítások: Jelölje be Allow schema drift a többi beállítást, és hagyja bejelölve
    • Mintavételezés: Kiválasztás Disable
    • Adatkészlet: Új adatkészlet létrehozásához válassza az + Új lehetőséget. Használja az Azure Synapse társított szolgáltatást, és válassza a DimCustomer táblát. Ügyeljen arra, hogy a DBName értéke az SQL-készlet neve legyen.

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

Átalakítások hozzáadása az adatfolyamhoz

  1. Válassza + ki a forrástól jobbra SourceDB a vásznon, majd válassza a Származtatott oszlop lehetőséget.

    The plus button and derived column menu item are highlighted.

  2. A csoportban Derived column's settingskonfigurálja a következő tulajdonságokat:

    • Kimeneti stream neve: Enter CreateCustomerHash
    • Bejövő stream: Válassza a SourceDB
    • Oszlopok: Adja meg a következőket:
    Column Expression Leírás
    Írja be a következőt: HashKey sha2(256, iifNull(Title,'') +FirstName +iifNull(MiddleName,'') +LastName +iifNull(Suffix,'') +iifNull(CompanyName,'') +iifNull(SalesPerson,'') +iifNull(EmailAddress,'') +iifNull(Phone,'')) Létrehoz egy SHA256 kivonatot a táblaértékek közül. Ezzel észleljük a sorváltozásokat, ha összehasonlítjuk a bejövő rekordok kivonatát a célrekordok kivonatértékével, és megfeleltetjük az CustomerID értéket. A iifNull függvény üres sztringekre cseréli a null értékeket. Ellenkező esetben a kivonatértékek ismétlődőek, ha null bejegyzések vannak jelen.

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

  3. Válassza + a vásznon a CreateCustomerHash származtatott oszloptól jobbra, majd válassza a Létező lehetőséget.

    The plus button and exists menu item are both highlighted.

  4. A csoportban Exists settingskonfigurálja a következő tulajdonságokat:

    • Kimeneti stream neve: Enter Exists
    • Bal oldali stream: Kiválasztás CreateCustomerHash
    • Jobb oldali stream: Válassza a SynapseDimCustomer
    • Létező típus: Válassza a Doesn't exist
    • Létező feltételek: Állítsa be a következőket a bal és a jobb oldalon:
    Balra: CreateCustomerHash oszlopa Jobbra: SynapseDimCustomer oszlopa
    HashKey HashKey

    The Exists settings form is configured as described.

  5. Válassza + a vászon jobb oldalánExists, majd a Keresés lehetőséget.

    The plus button and lookup menu item are both highlighted.

  6. A csoportban Lookup settingskonfigurálja a következő tulajdonságokat:

    • Kimeneti stream neve: Enter LookupCustomerID
    • Elsődleges stream: Kiválasztás Exists
    • Keresési stream: Kiválasztás SynapseDimCustomer
    • Több sor egyeztetése: Nincs bejelölve
    • Egyezés bekapcsolva: Válassza a Any row
    • Keresési feltételek: Állítsa be a következőket a bal és a jobb oldalon:
    Balra: Létezik oszlop Jobbra: SynapseDimCustomer oszlopa
    CustomerID CustomerID

    The Lookup settings form is configured as described.

  7. Válassza + ki a vászon jobb oldalán LookupCustomerID , majd válassza a Származtatott oszlop lehetőséget.

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

  8. A csoportban Derived column's settingskonfigurálja a következő tulajdonságokat:

    • Kimeneti stream neve: Enter SetDates
    • Bejövő stream: Válassza a LookupCustomerID
    • Oszlopok: Adja meg a következőket:
    Column Expression Leírás
    A következők szerint válasszon: InsertedDate iif(isNull(InsertedDate), currentTimestamp(), {InsertedDate}) Ha az InsertedDate érték null, szúrja be az aktuális időbélyeget. Ellenkező esetben használja az InsertedDate értéket.
    A következők szerint válasszon: ModifiedDate currentTimestamp() Mindig frissítse az ModifiedDate értéket az aktuális időbélyeggel.

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

    Megjegyzés:

    A második oszlop beszúrásához válassza a + Hozzáadás lehetőséget az Oszlopok lista fölé, majd az Oszlop hozzáadása lehetőséget.

  9. Válassza + ki a vásznon a SetDates származtatott oszlop lépésétől jobbra, majd válassza az Alter Row (Sor módosítása) lehetőséget.

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

  10. A csoportban Alter row settingskonfigurálja a következő tulajdonságokat:

    • Kimeneti stream neve: Enter AllowUpserts
    • Bejövő stream: Válassza a SetDates
    • Sorfeltételek módosítása: Adja meg a következőket:
    Feltétel Expression Leírás
    A következők szerint válasszon: Upsert if true() Állítsa a feltételt true() a feltételre az Upsert if upserts engedélyezéséhez. Ez biztosítja, hogy a leképezési adatfolyam lépésein áthaladó összes adat beszúrva vagy frissítve legyen a fogadóba.

    The alter row settings form is configured as described.

  11. Válassza + a vászon alter row lépésének AllowUpserts jobb oldalán, majd válassza a Fogadó lehetőséget.

    The plus button and sink menu item are both highlighted.

  12. A csoportban Sinkkonfigurálja a következő tulajdonságokat:

    • Kimeneti stream neve: Enter Sink
    • Bejövő stream: Válassza a AllowUpserts
    • Fogadó típusa: Válassza a Dataset
    • Adatkészlet: Kiválasztás DimCustomer
    • Beállítások: Ellenőrzés Allow schema drift és jelölés törlése Validate schema

    The sink properties form is configured as described.

  13. Válassza a Gépház lapot, és konfigurálja a következő tulajdonságokat:

    • Frissítési módszer: Az összes többi beállítás ellenőrzése Allow upsert és törlése
    • Kulcsoszlopok: Kijelölés List of columns, majd kijelölés CustomerID a listában
    • Táblaművelet: Kiválasztás None
    • Előkészítés engedélyezése: Nincs bejelölve

    The sink settings are configured as described.

  14. Válassza a Leképezés lapot, majd törölje az Automatikus leképezés jelölőnégyzet jelölését. Konfigurálja a bemeneti oszlopok leképezését az alábbiak szerint:

    Bemeneti oszlopok Kimeneti oszlopok
    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. A befejezett leképezési folyamatnak az alábbihoz hasonlóan kell kinéznie. A módosítások mentéséhez válassza az Összes közzététele lehetőséget.

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

  16. Válassza a Közzététel lehetőséget.

    The publish button is highlighted.

Az adatfolyam tesztelése

Végrehajtott egy 1. típusú SCD-adatfolyamot. Ha úgy dönt, hogy teszteli, hozzáadhatja ezt az adatfolyamot egy Synapse-integrációs folyamathoz. Ezután egyszer futtathatja a folyamatot, hogy elvégezhesse az ügyfél forrásadatainak kezdeti terhelését a DimCustomer-célhelyre.

A folyamat minden további futtatása összehasonlítja a forrástábla adatait a dimenziótáblában lévő adatokkal (a Kivonatkulcs használatával), és csak a módosított rekordokat frissíti. Ennek teszteléséhez frissíthet egy rekordot a forrástáblában, majd újra futtathatja a folyamatot, és ellenőrizheti a rekordfrissítéseket a dimenziótáblában.

Vegyük példaként Janet Gates ügyfelet. A kezdeti terhelés a LastName Gates és a CustomerId 4 terhelést mutatja.

The script is displayed with the initial customer record.

Íme egy példautasítás, amely frissíti az ügyfél vezetéknevét a forrástáblában.

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

A rekord frissítése és a folyamat ismételt futtatása után a DimCustomer megjeleníti ezeket a frissített adatokat.

The script is displayed with the updated customer record.

Az ügyfélrekord sikeresen frissítette az értéket a LastName forrásrekordnak megfelelően, és a ModifiedDaterégi LastName érték nyomon követése nélkül frissítette az értéket. Ez az 1. típusú SCD várt viselkedése. Ha előzményre volt szükség a LastName mezőhöz, akkor a táblát és az adatfolyamot úgy kell módosítania, hogy az a többi tanult SCD-típus egyike legyen.