Esercizio - Progettare e implementare una dimensione a modifica lenta di tipo 1 con flussi di dati per mapping

  • 10 minuti

In questo esercizio viene creato un flusso di dati per una dimensione a modifica lenta di tipo 1 usando il pool SQL dedicato di Azure Synapse come origine e destinazione. Questo flusso di dati può quindi essere aggiunto a un'istanza di Synapse Pipelines ed eseguito come parte del processo ETL (estrazione, trasformazione e caricamento).

Configurazione dei dati di origine e della tabella delle dimensioni

Per questo esercizio si vuole caricare una tabella delle dimensioni in Azure Synapse da dati di origine che possono provenire da molti tipi di sistemi diversi, ad esempio SQL di Azure, Archiviazione di Azure e così via. L'esempio viene mantenuto semplice creando i dati di origine nel database di Azure Synapse.

  1. In Synapse Studio passare all'hub Dati.

    Data hub.

  2. Selezionare la scheda Area di lavoro(1), espandere Database, quindi fare clic con il pulsante destro del mouse su SQLPool01 (2). Selezionare Nuovo script SQL (3), quindi Script vuoto (4).

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

  3. Incollare lo script seguente nella finestra Script vuoto, quindi selezionare Esegui o premere F5 per eseguire la query:

    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.

Creare un flusso di dati per mapping

I flussi di dati per mapping sono attività della pipeline che consentono di specificare in modo visivo come trasformare i dati, tramite un'esperienza che non richiede la scrittura di codice. Di seguito viene creato un flusso di dati per mapping per creare una dimensione a modifica lenta di tipo 1.

  1. Passare all'hub Sviluppo.

    Develop hub.

  2. Selezionare +, quindi selezionare Flusso di dati.

    The plus button and data flow menu item are highlighted.

  3. Nel riquadro delle proprietà del nuovo flusso di dati immettere UpdateCustomerDimension nel campo Nome(1), quindi selezionare il pulsante Proprietà(2) per nascondere il riquadro delle proprietà.

    The data flow properties pane is displayed.

  4. Selezionare Aggiungi origine nell'area di disegno.

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

  5. In Source settings configurare le proprietà seguenti:

    • Nome flusso di output: immettere SourceDB
    • Tipo di origine: selezionare Dataset
    • Opzioni: selezionare Allow schema drift e lasciare deselezionate le altre opzioni
    • Campionamento: selezionare Disable
    • Set di dati: selezionare + Nuovo per creare un nuovo set di dati

    The New button is highlighted next to Dataset.

  6. Nella finestra di dialogo Nuovo set di dati di integrazione selezionare Azure Synapse Analyticse quindi Continua.

    Azure SQL Database and the Continue button are highlighted.

  7. Nelle proprietà del set di dati configurare le impostazioni seguenti:

    • Nome: immettere CustomerSource
    • Servizio collegato: selezionare il servizio collegato dell'area di lavoro di Synapse
    • Nome tabella: selezionare il pulsante Aggiorna accanto all'elenco a discesa

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

  8. Nel campo Valore immettere il nome del pool SQL, quindi selezionare OK.

    The SQLPool01 parameter is highlighted.

  9. Selezionare dbo.CustomerSource in Nome tabella, selezionare From connection/store in Importa schema, quindi fare clic su OK per creare il set di dati.

    The form is completed as described.

  10. Selezionare Apri accanto al set di dati CustomerSource aggiunto.

    The open button is highlighted next to the new dataset.

  11. Immettere il nome del pool SQL nel campo Valore accanto a DBName.

  12. Nell'editor del flusso di dati selezionare la casella Aggiungi origine sotto l'attività SourceDB. Configurare questa origine come tabella DimCustomer seguendo la stessa procedura usata per CustomerSource.

    • Nome flusso di output: immettere DimCustomer
    • Tipo di origine: selezionare Dataset
    • Opzioni: selezionare Allow schema drift e lasciare deselezionate le altre opzioni
    • Campionamento: selezionare Disable
    • Set di dati: selezionare + Nuovo per creare un nuovo set di dati. Usare il servizio collegato Azure Synapse e scegliere la tabella DimCustomer. Assicurarsi di impostare DBName sul nome del pool SQL.

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

Aggiungere trasformazioni al flusso di dati

  1. Selezionare + a destra dell'origine SourceDB nell'area di disegno e quindi selezionare Colonna derivata.

    The plus button and derived column menu item are highlighted.

  2. In Derived column's settings configurare le proprietà seguenti:

    • Nome flusso di output: immettere CreateCustomerHash
    • Incoming stream (Flusso in ingresso): selezionare SourceDB
    • Colonne: immettere i valori seguenti:
    Column Expression Descrizione
    Digitare HashKey sha2(256, iifNull(Title,'') +FirstName +iifNull(MiddleName,'') +LastName +iifNull(Suffix,'') +iifNull(CompanyName,'') +iifNull(SalesPerson,'') +iifNull(EmailAddress,'') +iifNull(Phone,'')) Crea un hash SHA256 dei valori della tabella. Questo viene usato per rilevare le modifiche alle righe confrontando l'hash dei record in ingresso con il valore hash dei record di destinazione, corrispondente al valore CustomerID. La funzione iifNull sostituisce i valori Null con stringhe vuote. In caso contrario, i valori hash tendono a essere duplicati quando sono presenti voci Null.

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

  3. Selezionare + a destra della colonna derivata CreateCustomerHash nell'area di disegno, quindi selezionare Esiste.

    The plus button and exists menu item are both highlighted.

  4. In Exists settings configurare le proprietà seguenti:

    • Nome flusso di output: immettere Exists
    • Flusso di sinistra: selezionare CreateCustomerHash
    • Flusso di destra: selezionare SynapseDimCustomer
    • Tipo esistente: selezionare Doesn't exist
    • Condizioni Exists: impostare i valori seguenti per i flussi destro e sinistro:
    Sinistro: colonna di CreateCustomerHash Destro: colonna di SynapseDimCustomer
    HashKey HashKey

    The Exists settings form is configured as described.

  5. Selezionare + a destra di Exists nell'area di disegno, quindi selezionare Ricerca.

    The plus button and lookup menu item are both highlighted.

  6. In Lookup settings configurare le proprietà seguenti:

    • Nome flusso di output: immettere LookupCustomerID
    • Primary stream (Flusso primario): selezionare Exists
    • Lookup stream (Flusso di ricerca): selezionare SynapseDimCustomer
    • Corrispondenza in più righe: deselezionata
    • Corrispondenza in: selezionare Any row
    • Lookup conditions (Condizioni di ricerca): impostare i valori seguenti per i flussi destro e sinistro:
    Sinistro: colonna di Exists Destro: colonna di SynapseDimCustomer
    CustomerID CustomerID

    The Lookup settings form is configured as described.

  7. Selezionare + a destra di LookupCustomerID nell'area di disegno, quindi selezionare Colonna derivata.

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

  8. In Derived column's settings configurare le proprietà seguenti:

    • Nome flusso di output: immettere SetDates
    • Incoming stream (Flusso in ingresso): selezionare LookupCustomerID
    • Colonne: immettere i valori seguenti:
    Column Expression Descrizione
    Selezionare InsertedDate iif(isNull(InsertedDate), currentTimestamp(), {InsertedDate}) Se il valore di InsertedDate è Null, inserire il timestamp corrente. In caso contrario, usare il valore di InsertedDate.
    Selezionare ModifiedDate currentTimestamp() Aggiornare sempre il valore di ModifiedDate con il timestamp corrente.

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

    Nota

    Per inserire la seconda colonna, selezionare + Aggiungi sopra l'elenco Colonne, quindi selezionare Aggiungi colonna.

  9. Selezionare + a destra del passaggio della colonna derivata SetDates nell'area di disegno, quindi selezionare Modifica riga.

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

  10. In Alter row settings configurare le proprietà seguenti:

    • Nome flusso di output: immettere AllowUpserts
    • Incoming stream (Flusso in ingresso): selezionare SetDates
    • Modifica le condizioni di riga: immettere i valori seguenti:
    Condizione Expression Descrizione
    Selezionare Upsert if true() Impostare la condizione su true() nella condizione Upsert if per consentire le operazioni di upsert. In questo modo, si garantisce che tutti i dati che attraversano i passaggi nel flusso di dati per mapping vengano inseriti o aggiornati nel sink.

    The alter row settings form is configured as described.

  11. Selezionare + a destra del passaggio di modifica riga AllowUpserts nell'area di disegno, quindi selezionare Sink.

    The plus button and sink menu item are both highlighted.

  12. In Sink configurare le proprietà seguenti:

    • Nome flusso di output: immettere Sink
    • Incoming stream (Flusso in ingresso): selezionare AllowUpserts
    • Tipo di sink: selezionare Dataset
    • Set di dati: selezionare DimCustomer
    • Opzioni: selezionare Allow schema drift e deselezionare Validate schema

    The sink properties form is configured as described.

  13. Selezionare la scheda Impostazioni e configurare le proprietà seguenti:

    • Metodo di aggiornamento: selezionare Allow upsert e deselezionare tutte le altre opzioni
    • Colonne chiave: selezionare List of columns, quindi selezionare CustomerID nell'elenco
    • Azione tabella: selezionare None
    • Abilita staging: deselezionata

    The sink settings are configured as described.

  14. Selezionare la scheda Mapping, quindi deselezionare Auto mapping. Configurare il mapping delle colonne di input come descritto di seguito:

    Colonne di input Colonne di output
    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. Il flusso per mapping completo sarà simile al seguente. Selezionare Pubblica tutti per salvare le modifiche.

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

  16. Selezionare Pubblica.

    The publish button is highlighted.

Come testare il flusso di dati

Il flusso di dati per la dimensione a modifica lenta di tipo 1 è stato completato. Se si sceglie di testare il flusso di dati, è possibile aggiungerlo a una pipeline di integrazione di Synapse. È quindi possibile eseguire la pipeline una volta per effettuare il caricamento iniziale dei dati di origine del cliente nella destinazione DimCustomer.

Ogni esecuzione aggiuntiva della pipeline confronterà i dati nella tabella di origine con quelli già presenti nella tabella delle dimensioni (usando HashKey) e aggiornerà solo i record modificati. Per eseguire il test, è possibile aggiornare un record nella tabella di origine e quindi eseguire di nuovo la pipeline e verificare gli aggiornamenti dei record nella tabella delle dimensioni.

Esaminare il cliente Janet Gates come esempio. Il caricamento iniziale mostra che LastName è Gates e CustomerId è 4.

The script is displayed with the initial customer record.

Qui è mostrata un'istruzione di esempio che aggiorna il cognome del cliente nella tabella di origine.

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

Dopo aver aggiornato il record e aver eseguito di nuovo la pipeline, DimCustomer mostrerà questi dati aggiornati.

The script is displayed with the updated customer record.

Il record del cliente ha aggiornato il valore di LastName in modo che corrisponda al record di origine e ha aggiornato ModifiedDate, senza tenere traccia del valore precedente di LastName. Questo è il comportamento previsto per una dimensione a modifica lenta di tipo 1. Se per il campo LastName è necessaria la cronologia, modificare la tabella e il flusso di dati in modo che corrispondano agli altri tipi di dimensione a modifica lenta appresi qui.