Copiare e trasformare i dati da/in SQL Server usando Azure Data Factory o Azure Synapse Analytics

SI APPLICA A: Azure Data Factory Azure Synapse Analytics

In questo articolo viene illustrato come usare l'attività di copia in pipeline di Azure Data Factory e Azure Synapse Analytics per copiare i dati da/in un database SQL Server e usare Flusso di dati per trasformare o dati in un database SQL Server. Per altre informazioni, vedere l'articolo introduttivo per Azure Data Factory o Azure Synapse Analytics.

Funzionalità supportate

Questo connettore SQL Server è supportato per le funzionalità seguenti:

① Azure Integration Runtime ② Runtime di integrazione self-hosted

Per un elenco degli archivi dati supportati come origini o sink dall'attività di copia, vedere la tabella relativa agli archivi dati supportati.

In particolare, il connettore SQL Server supporta:

• SQL Server versione 2005 e successive.
• La copia dei dati tramite l'autenticazione di SQL o di Windows.
• Come origine, recupero dati tramite una query SQL o una stored procedure. È anche possibile scegliere di eseguire la copia parallela dall'origine di SQL Server. Per informazioni dettagliate, vedere la sezione Copia parallela dal database SQL.
• Come sink, creazione automatica della tabella di destinazione se non esiste in base allo schema di origine; accodamento di dati a una tabella o richiamo di una stored procedure con logica personalizzata durante la copia.

LocalDB di SQL Server Express non è supportato.

Prerequisiti

Se l'archivio dati si trova all'interno di una rete locale, una rete virtuale di Azure o un cloud privato virtuale di Amazon, è necessario configurare un runtime di integrazione self-hosted per connettersi.

Se l'archivio dati è un servizio dati del cloud gestito, è possibile usare Azure Integration Runtime. Se l'accesso è limitato solo agli indirizzi IP approvati nelle regole del firewall, è possibile aggiungere IP di Azure Integration Runtime nell'elenco Consentiti.

È anche possibile usare la funzionalitàruntime di integrazione della rete virtuale gestita in Azure Data Factory per accedere alla rete locale senza installare e configurare un runtime di integrazione self-hosted.

Per altre informazioni sui meccanismi di sicurezza di rete e sulle opzioni supportate da Data Factory, vedere strategie di accesso ai dati.

Operazioni preliminari

Per eseguire l'attività di copia con una pipeline, è possibile usare uno degli strumenti o SDK seguenti:

• Strumento Copia dati
• Il portale di Azure
• .NET SDK
• SDK di Python
• Azure PowerShell
• API REST
• Modello di Azure Resource Manager

Creare il servizio collegato di SQL Server tramite l’interfaccia utente

Usare la procedura seguente per creare un servizio collegato di SQL Server nell'interfaccia utente del portale di Azure.

1. Passare alla scheda Gestisci nell'area di lavoro di Azure Data Factory o Synapse e selezionare Servizi collegati, quindi fare clic su Nuovo:

   • Azure Data Factory
   • Azure Synapse

   

2. Cercare SQL e selezionare il connettore SQL Server.

   

3. Configurare i dettagli del servizio, testare la connessione e creare il nuovo servizio collegato.

   

Dettagli di configurazione del connettore

Le sezioni seguenti forniscono informazioni dettagliate sulle proprietà usate per definire entità delle pipeline di Data Factory e Synapse specifiche per il connettore del database SQL Server.

Proprietà del servizio collegato

La versione consigliata di SQL Server supporta TLS 1.3. Fare riferimento a questa sezione per aggiornare il servizio collegato di SQL Server se si usa una versione legacy. Per informazioni dettagliate delle proprietà, vedere le sezioni corrispondenti.

• Versione consigliata
• Versione legacy

Versione consigliata

Queste proprietà generiche sono supportate per un servizio collegato di SQL Server quando si applica la versione consigliata:

Per altre proprietà di connessione, vedere la tabella seguente:

Autenticazione SQL

Per usare l’autenticazione SQL, oltre alle proprietà generiche descritte nella sezione precedente, specificare le proprietà seguenti:

Esempio: usare l’autenticazione SQL

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Esempio: usare l'autenticazione di SQL con una password in Azure Key Vault

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<secretName>" 
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Esempio: usare Always Encrypted

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "alwaysEncryptedSettings": {
            "alwaysEncryptedAkvAuthType": "ServicePrincipal",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalKey": {
                "type": "SecureString",
                "value": "<service principal key>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Autenticazione di Windows

Per usare l’autenticazione di Windows, oltre alle proprietà generiche descritte nella sezione precedente, specificare le proprietà seguenti:

Esempio: usare l’autenticazione di Windows

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "Windows",
            "userName": "<domain\\username>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Esempio: usare l'autenticazione di Windows con una password in Azure Key Vault

{
    "name": "SqlServerLinkedService",
    "properties": {
        "annotations": [],
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "Windows",
            "userName": "<domain\\username>",
            "password": {
                "type": "AzureKeyVaultSecret",
                "store": {
                    "referenceName": "<Azure Key Vault linked service name>",
                    "type": "LinkedServiceReference"
                },
                "secretName": "<secretName>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Autenticazione dell'identità gestita assegnata dall'utente

Una data factory o un'area di lavoro di Synapse può essere associata a un'identità gestita assegnata dall'utente che rappresenta il servizio durante l'autenticazione ad altre risorse in Azure. È possibile usare questa identità gestita per l'autenticazione di SQL Server nelle macchine virtuali di Azure. L'area di lavoro designata factory o Synapse può accedere e copiare dati da o nel database usando questa identità.

Per usare l'autenticazione dell'identità gestita assegnata dall'utente, oltre alle proprietà generiche descritte nella sezione precedente, specificare le proprietà seguenti:

È anche necessario seguire la procedura seguente:

1. Concedere le autorizzazioni per l'identità gestita assegnata dall'utente.

2. Abilitare l'autenticazione di Microsoft Entra per SQL Server nelle macchine virtuali di Azure.

3. Creare utenti di database indipendenti per l'identità gestita assegnata dall'utente. Connettersi al database da o a cui si desidera copiare dati usando strumenti come SQL Server Management Studio, con un'identità di Microsoft Entra che dispone almeno dell'autorizzazione ALTER ANY USER. Eseguire il codice T-SQL seguente:

   CREATE USER [your_resource_name] FROM EXTERNAL PROVIDER;
   

4. Creare una o più identità gestite assegnate dall'utente e concedere le autorizzazioni necessarie per l'identità gestita assegnata dall'utente, come normalmente si fa per utenti SQL e altri utenti. Eseguire il codice seguente. Per altre opzioni, vedere questo documento.

   ALTER ROLE [role name] ADD MEMBER [your_resource_name];
   

5. Assegnare una o più identità gestite assegnate dall'utente alla data factory e creare le credenziali per ogni identità gestita assegnata dall'utente.

6. Configurare un servizio collegato di SQL Server.

Esempio

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "UserAssignedManagedIdentity",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Versione legacy

Queste proprietà generiche sono supportate per un servizio collegato di SQL Server quando si applica la versione Legacy:

Il connettore SQL Server supporta i tipi di autenticazione seguenti. Per informazioni dettagliate, vedere le sezioni corrispondenti.

• Autenticazione di SQL per la versione legacy
• Autenticazione di Windows per la versione legacy

Autenticazione di SQL per la versione legacy

Per usare l’autenticazione SQL, oltre alle proprietà generiche descritte nella sezione precedente, specificare le proprietà seguenti:

Autenticazione di Windows per la versione legacy

Per usare l’autenticazione di Windows, oltre alle proprietà generiche descritte nella sezione precedente, specificare le proprietà seguenti:

Proprietà del set di dati

Per un elenco completo delle sezioni e delle proprietà disponibili per la definizione di set di dati, vedere l'articolo sui set di dati. Questa sezione fornisce un elenco delle proprietà supportate dal set di dati SQL Server.

Per copiare dati da/in un database di SQL Server, sono supportate le proprietà seguenti:

Esempio

{
    "name": "SQLServerDataset",
    "properties":
    {
        "type": "SqlServerTable",
        "linkedServiceName": {
            "referenceName": "<SQL Server linked service name>",
            "type": "LinkedServiceReference"
        },
        "schema": [ < physical schema, optional, retrievable during authoring > ],
        "typeProperties": {
            "schema": "<schema_name>",
            "table": "<table_name>"
        }
    }
}

Proprietà dell'attività di copia

Per un elenco completo delle sezioni e delle proprietà disponibili per definire le attività, vedere l'articolo sulle pipeline. Questa sezione fornisce un elenco delle proprietà supportate dall'origine e dal sink di SQL Server.

SQL Server come origine

Per copiare dati da un database SQL Server, impostare il tipo di origine nell'attività di copia su SqlSource. Nella sezione source dell'attività di copia sono supportate le proprietà seguenti:

Tenere presente quanto segue:

• Se è specificato sqlReaderQuery per SqlSource, l’attività di copia esegue questa query nell’origine di SQL Server per ottenere i dati. In alternativa, è possibile specificare una stored procedure indicando i parametri sqlReaderStoredProcedureName e storedProcedureParameters, se la stored procedure accetta parametri.
• Quando si usa la stored procedure nell'origine per recuperare dati, tenere presente che se la stored procedure è progettata per restituire schemi diversi quando viene passato un valore di parametro diverso, è possibile che si verifichi un errore o un risultato imprevisto durante l'importazione dello schema dall'interfaccia utente o quando si copiano dati nel database SQL con la creazione automatica della tabella.

Esempio: usare una query SQL

"activities":[
    {
        "name": "CopyFromSQLServer",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<SQL Server input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SqlSource",
                "sqlReaderQuery": "SELECT * FROM MyTable"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

Esempio: usare una stored procedure

"activities":[
    {
        "name": "CopyFromSQLServer",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<SQL Server input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SqlSource",
                "sqlReaderStoredProcedureName": "CopyTestSrcStoredProcedureWithParameters",
                "storedProcedureParameters": {
                    "stringData": { "value": "str3" },
                    "identifier": { "value": "$$Text.Format('{0:yyyy}', <datetime parameter>)", "type": "Int"}
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

Definizione della stored procedure

CREATE PROCEDURE CopyTestSrcStoredProcedureWithParameters
(
    @stringData varchar(20),
    @identifier int
)
AS
SET NOCOUNT ON;
BEGIN
    select *
    from dbo.UnitTestSrcTable
    where dbo.UnitTestSrcTable.stringData != stringData
    and dbo.UnitTestSrcTable.identifier != identifier
END
GO

SQL Server come sink

Per copiare dati da SQL Server, impostare il tipo di sink nell'attività di copia su SqlSink. Nella sezione sink dell'attività di copia sono supportate le proprietà seguenti:

Esempio 1: accodare dati

"activities":[
    {
        "name": "CopyToSQLServer",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<SQL Server output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SqlSink",
                "tableOption": "autoCreate",
                "writeBatchSize": 100000
            }
        }
    }
]

Esempio 2: richiamare una stored procedure durante la copia

Per altre informazioni, vedere Richiamare una stored procedure da un sink SQL.

"activities":[
    {
        "name": "CopyToSQLServer",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<SQL Server output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SqlSink",
                "sqlWriterStoredProcedureName": "CopyTestStoredProcedureWithParameters",
                "storedProcedureTableTypeParameterName": "MyTable",
                "sqlWriterTableType": "MyTableType",
                "storedProcedureParameters": {
                    "identifier": { "value": "1", "type": "Int" },
                    "stringData": { "value": "str1" }
                }
            }
        }
    }
]

Esempio 3: Upsert dei dati

"activities":[
    {
        "name": "CopyToSQLServer",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<SQL Server output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SqlSink",
                "tableOption": "autoCreate",
                "writeBehavior": "upsert",
                "upsertSettings": {
                    "useTempDB": true,
                    "keys": [
                        "<column name>"
                    ]
                },
            }
        }
    }
]

Copia parallela dal database SQL

Il connettore SQL Server nell'attività Copy fornisce il partizionamento dei dati predefinito per copiare i dati in parallelo. È possibile trovare le opzioni di partizionamento dei dati nella tabella Origine dell'attività di copia.

Quando si abilita la copia partizionata, l’attività di copia esegue query parallele sull'origine di SQL Server per caricare i dati in base alle partizioni. Il grado di parallelismo è controllato dall'impostazione parallelCopies sull'attività di copia. Se, ad esempio, si imposta parallelCopies su quattro, il servizio simultaneamente genera ed esegue quattro query in base all'opzione partizione specificata e alle impostazioni, e ogni query recupera una porzione di dati dal database di SQL Server.

È preferibile abilitare la copia parallela con il partizionamento dei dati specialmente quando si caricano notevoli quantità di dati dal database di SQL Server. Di seguito sono riportate le configurazioni consigliate per i diversi scenari. Quando si copiano dati in un archivio dati basato su file, è consigliabile scrivere in una cartella come file multipli (specificare solo il nome della cartella); in tal caso, le prestazioni risultano migliori rispetto alla scrittura in un singolo file.

Procedure consigliate per il caricamento di dati con opzione partizione:

1. Scegliere una colonna distintiva come colonna partizione (ad esempio, chiave primaria o chiave univoca) per evitare l'asimmetria dei dati.
2. Se la tabella include una partizione predefinita, usare l'opzione di partizione "Partizioni fisiche della tabella" per ottenere prestazioni migliori.
3. Se si usa Azure Integration Runtime per copiare i dati, è possibile impostare "Unità di integrazione dati (DIU)" (>4) perché utilizzi più risorse di calcolo. Controllare gli scenari applicabili.
4. “Grado di parallelismo copia” controlla i numeri partizione; se si imposta per questo numero un valore troppo grande, le prestazioni potrebbero talvolta risentirne. È preferibile impostare questo numero come (DIU o numero di nodi del runtime di integrazione self-hosted) * (2-4).

Esempio: caricamento completo da una tabella di grandi dimensioni con partizioni fisiche

"source": {
    "type": "SqlSource",
    "partitionOption": "PhysicalPartitionsOfTable"
}

Esempio: query con partizione a intervalli dinamici

"source": {
    "type": "SqlSource",
    "query": "SELECT * FROM <TableName> WHERE ?DfDynamicRangePartitionCondition AND <your_additional_where_clause>",
    "partitionOption": "DynamicRange",
    "partitionSettings": {
        "partitionColumnName": "<partition_column_name>",
        "partitionUpperBound": "<upper_value_of_partition_column (optional) to decide the partition stride, not as data filter>",
        "partitionLowerBound": "<lower_value_of_partition_column (optional) to decide the partition stride, not as data filter>"
    }
}

Query di esempio per controllare la partizione fisica

SELECT DISTINCT s.name AS SchemaName, t.name AS TableName, pf.name AS PartitionFunctionName, c.name AS ColumnName, iif(pf.name is null, 'no', 'yes') AS HasPartition
FROM sys.tables AS t
LEFT JOIN sys.objects AS o ON t.object_id = o.object_id
LEFT JOIN sys.schemas AS s ON o.schema_id = s.schema_id
LEFT JOIN sys.indexes AS i ON t.object_id = i.object_id 
LEFT JOIN sys.index_columns AS ic ON ic.partition_ordinal > 0 AND ic.index_id = i.index_id AND ic.object_id = t.object_id 
LEFT JOIN sys.columns AS c ON c.object_id = ic.object_id AND c.column_id = ic.column_id 
LEFT JOIN sys.partition_schemes ps ON i.data_space_id = ps.data_space_id 
LEFT JOIN sys.partition_functions pf ON pf.function_id = ps.function_id 
WHERE s.name='[your schema]' AND t.name = '[your table name]'

Se la tabella ha una partizione fisica, viene visualizzato "HasPartition" come "sì", come illustrato di seguito.

Procedura consigliata per il caricamento dei dati in SQL Server

Quando si copiano dati in SQL Server, potrebbe essere necessario un comportamento di scrittura diverso:

• Accodamento: i dati di origine hanno solo nuovi record.
• Upsert: i dati di origine hanno sia inserimenti che aggiornamenti.
• Sovrascrittura: si desidera ricaricare ogni volta l'intera tabella dimensioni.
• Scrittura con logica personalizzata: è necessaria un'elaborazione aggiuntiva prima dell'inserimento finale nella tabella di destinazione.

Per informazioni sulla modalità di configurazione e per procedure consigliate, vedere le sezioni pertinenti.

Accodare dati

L'accodamento dei dati è il comportamento predefinito di questo connettore sink di SQL Server. Il servizio esegue un inserimento bulk per scrivere nella tabella in modo efficiente. È possibile configurare opportunamente l'origine e il sink nell'attività di copia.

Eseguire l'upsert dei dati

L'attività di copia ora supporta il caricamento nativo dei dati in una tabella temporanea del database e quindi l'aggiornamento dei dati nella tabella sink, se la chiave esiste, altrimenti inserisce nuovi dati. Per altre informazioni sulle impostazioni dell’operazione upsert nelle attività di copia, vedere SQL Server come sink.

Sovrascrivere l'intera tabella

È possibile configurare la proprietà preCopyScript in un sink dell'attività di copia. In questo caso, per ogni attività di copia eseguita, il servizio esegue prima lo script. Quindi, esegue la copia per inserire i dati. Ad esempio, per sovrascrivere l'intera tabella con i dati più recenti, specificare uno script per eliminare tutti i record prima del caricamento bulk dei nuovi dati dall'origine.

Scrivere dati con logica personalizzata

I passaggi per scrivere dati con una logica personalizzata sono simili a quelli descritti nella sezione Upsert dei dati. Quando si necessita di applicare un'elaborazione aggiuntiva prima dell'inserimento finale dei dati di origine nella tabella di destinazione, è possibile caricare in una tabella di staging e quindi richiamare un'attività stored procedure, o richiamare una stored procedure nel sink dell'attività di copia per applicare i dati.

Richiamare una stored procedure da un sink SQL

Quando si copiano dati in un database SQL Server, è anche possibile configurare e richiamare una stored procedure specificata dall'utente con parametri aggiuntivi in ogni batch della tabella di origine. La funzionalità di stored procedure sfrutta i parametri valutati a livello di tabella. Tenere presente che il servizio esegue automaticamente il wrapping della stored procedure nella propria transazione, pertanto qualunque transazione creata all'interno della stored procedure diventerà una transazione nidificata e potrebbe avere implicazioni per la gestione delle eccezioni.

È possibile usare una stored procedure quando non si possono usare i meccanismi di copia predefiniti. Ad esempio, quando si desidera applicare un'elaborazione aggiuntiva prima dell'inserimento finale dei dati di origine nella tabella di destinazione. Alcuni esempi di elaborazione aggiuntivi sono: quando si desidera unire colonne, cercare valori aggiuntivi e inserire in più tabelle.

Nell'esempio seguente viene illustrato come usare una stored procedure per eseguire un'operazione upsert in una tabella del database SQL Server. Si presuppone che i dati di input e la tabella Marketing del sink abbiano tre colonne: ProfileID, Stato e Categoria. Eseguire l'operazione upsert nella colonna ProfileID e applicarla solo a una categoria specifica denominata “ProductA”.

1. Nel database, definire il tipo di tabella con lo stesso nome di sqlWriterTableType. Lo schema del tipo di tabella è identico allo schema restituito dai dati di input.

   CREATE TYPE [dbo].[MarketingType] AS TABLE(
       [ProfileID] [varchar](256) NOT NULL,
       [State] [varchar](256) NOT NULL,
       [Category] [varchar](256) NOT NULL
   )
   

2. Nel database, definire la stored procedure con lo stesso nome di sqlWriterStoredProcedureName. che gestisce i dati di input dell'origine specificata e li unisce nella tabella di output. Il nome del parametro del tipo di tabella nella stored procedure deve essere identico al valore tableName definito nel set di dati.

   CREATE PROCEDURE spOverwriteMarketing @Marketing [dbo].[MarketingType] READONLY, @category varchar(256)
   AS
   BEGIN
   MERGE [dbo].[Marketing] AS target
   USING @Marketing AS source
   ON (target.ProfileID = source.ProfileID and target.Category = @category)
   WHEN MATCHED THEN
       UPDATE SET State = source.State
   WHEN NOT MATCHED THEN
       INSERT (ProfileID, State, Category)
       VALUES (source.ProfileID, source.State, source.Category);
   END
   

3. Definire la sezione SqlSink nel file JSON dell'attività di copia come indicato di seguito:

   "sink": {
       "type": "SqlSink",
       "sqlWriterStoredProcedureName": "spOverwriteMarketing",
       "storedProcedureTableTypeParameterName": "Marketing",
       "sqlWriterTableType": "MarketingType",
       "storedProcedureParameters": {
           "category": {
               "value": "ProductA"
           }
       }
   }
   

Proprietà del flusso di dati per mapping

Durante la trasformazione dei dati in un flusso di dati per mapping, è possibile leggere e scrivere in tabelle del database di SQL Server. Per altre informazioni, vedere la trasformazione origine e la trasformazione sink nei flussi di dati per mapping.

Trasformazione origine

Nella tabella seguente sono elencate le proprietà supportate dall'origine di SQL Server. È possibile modificare queste proprietà nella scheda Opzioni origine.

CREATE PROC CTESP @query nvarchar(max)
AS
BEGIN
EXECUTE sp_executesql @query;
END

Esempio di script di origine di SQL Server

Quando si usa SQL Server come tipo di origine, lo script del flusso di dati associato è:

source(allowSchemaDrift: true,
    validateSchema: false,
    isolationLevel: 'READ_UNCOMMITTED',
    query: 'select * from MYTABLE',
    format: 'query') ~> SQLSource

Trasformazione sink

Nella tabella seguente sono elencate le proprietà supportate dal sink di SQL Server. È possibile modificare queste proprietà nella scheda Opzioni sink.

Esempio di script di sink di SQL Server

Quando si usa SQL Server come tipo di sink, lo script del flusso di dati associato è:

IncomingStream sink(allowSchemaDrift: true,
    validateSchema: false,
    deletable:false,
    insertable:true,
    updateable:true,
    upsertable:true,
    keys:['keyColumn'],
    format: 'table',
    skipDuplicateMapInputs: true,
    skipDuplicateMapOutputs: true) ~> SQLSink

Mapping dei tipi di dati per SQL Server

Quando si copiano dati da/in SQL Server, vengono usati i mapping seguenti da tipi di dati di SQL Server in tipi di dati provvisori di Azure Data Factory. Le pipeline di Synapse, che implementano Data Factory, usano gli stessi mapping. Vedere Mapping dello schema e del tipo di dati per informazioni su come l'attività di copia esegue il mapping dello schema di origine e del tipo di dati al sink.

Proprietà dell'attività Lookup

Per altre informazioni sulle proprietà, vedere Attività Lookup.

Proprietà dell'attività GetMetadata

Per altre informazioni sulle proprietà, vedere Attività GetMetadata

Uso di Always Encrypted

Quando si copiano dati da/in SQL Server con Always Encrypted, seguire questa procedura:

1. Memorizzare la chiave master della colonna (CMK) in un Azure Key Vault. Altre informazioni su come configurare Always Encrypted usando Azure Key Vault

2. Accertarsi di concedere l'accesso all'insieme di credenziali delle chiavi in cui è archiviata la chiave master della colonna (CMK). Per le autorizzazioni necessarie, fare riferimento a questo articolo.

3. Creare un servizio collegato per connettersi al database SQL e abilitare la funzione "Always Encrypted" usando l'identità gestita o l'entità servizio.

Change Data Capture nativo

Azure Data Factory può supportare funzionalità Change Data Capture native per SQL Server, il database di Azure SQL e l'istanza gestita di Azure SQL. I dati modificati, inclusi l’inserimento, l’aggiornamento e l’eliminazione di righe in archivi SQL possono essere rilevati ed estratti automaticamente dal flusso di dati per mapping di ADF. Senza alcuna esperienza di coding nel flusso di dati per mapping, gli utenti possono ottenere facilmente uno scenario di replica dei dati da archivi SQL accodando un database come archivio di destinazione. Inoltre, gli utenti possono anche comporre qualsiasi logica di trasformazione dati intermedia per ottenere uno scenario ETL incrementale dagli archivi SQL.

Accertarsi di mantenere invariato il nome della pipeline e dell'attività, in modo che il checkpoint possa essere registrato da ADF per ottenere automaticamente i dati modificati dall'ultima esecuzione. Se si modifica il nome della pipeline o il nome dell'attività, il checkpoint verrà reimpostato, il che porta a ricominciare dall'inizio o a ottenere le modifiche da subito nella successiva esecuzione. Se si desidera modificare il nome della pipeline o il nome dell'attività mantenendo comunque il checkpoint per ottenere automaticamente i dati modificati dall'ultima esecuzione, usare la propria chiave checkpoint nell'attività del flusso di dati per ottenere tale risultato.

Quando si esegue il debug della pipeline, questa funzionalità funziona allo stesso modo. Tenere presente che il checkpoint verrà reimpostato quando si aggiorna il browser durante l'esecuzione del debug. Quando il risultato della pipeline dall'esecuzione del debug è soddisfacente, è possibile procedere alla pubblicazione e all'attivazione della pipeline. Al momento della prima attivazione della pipeline pubblicata, la pipeline viene riavviata automaticamente dall'inizio o ottiene le modifiche da tale momento in avanti.

Nella sezione di monitoraggio, è sempre possibile eseguire nuovamente una pipeline. Quando si esegue questa operazione, i dati modificati vengono sempre acquisiti dal checkpoint precedente dell'esecuzione della pipeline selezionata.

Esempio 1:

Quando si concatena direttamente una trasformazione di origine che fa riferimento al set di dati abilitato per SQL CDC, con una trasformazione sink che fa riferimento a un database in un flusso di dati per mapping, le modifiche apportate all'origine SQL verranno applicate automaticamente al database di destinazione, in modo da ottenere facilmente uno scenario di replica dei dati tra database. È possibile usare il metodo di aggiornamento nella trasformazione sink per selezionare se si desideri consentire l'inserimento, l'aggiornamento o l'eliminazione nel database di destinazione. Lo script di esempio nel flusso di dati di mapping è il seguente.

source(output(
		id as integer,
		name as string
	),
	allowSchemaDrift: true,
	validateSchema: false,
	enableNativeCdc: true,
	netChanges: true,
	skipInitialLoad: false,
	isolationLevel: 'READ_UNCOMMITTED',
	format: 'table') ~> source1
source1 sink(allowSchemaDrift: true,
	validateSchema: false,
	deletable:true,
	insertable:true,
	updateable:true,
	upsertable:true,
	keys:['id'],
	format: 'table',
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true,
	errorHandlingOption: 'stopOnFirstError') ~> sink1

Esempio 2:

Se si desidera abilitare lo scenario ETL anziché la replica dei dati tra database tramite SQL CDC, è possibile usare espressioni nel flusso di dati per mapping isInsert(1), isUpdate(1) e isDelete(1) per distinguere le righe con tipi di operazione diversi. Di seguito è riportato uno degli script di esempio per il flusso di dati per mapping sulla derivazione di una colonna con il valore: 1 per indicare le righe inserite, 2 per indicare le righe aggiornate e 3 per indicare le righe eliminate per le trasformazioni downstream per l’elaborazione dei dati delta.

source(output(
		id as integer,
		name as string
	),
	allowSchemaDrift: true,
	validateSchema: false,
	enableNativeCdc: true,
	netChanges: true,
	skipInitialLoad: false,
	isolationLevel: 'READ_UNCOMMITTED',
	format: 'table') ~> source1
source1 derive(operationType = iif(isInsert(1), 1, iif(isUpdate(1), 2, 3))) ~> derivedColumn1
derivedColumn1 sink(allowSchemaDrift: true,
	validateSchema: false,
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true) ~> sink1

Limitazione nota:

• Solo net changes da SQL CDC verrà caricato da ADF tramite cdc.fn_cdc_get_net_changes_.

Risolvere i problemi di connessione

1. Configurare l’istanza di SQL Server per accettare connessioni remote. Avviare SQL Server Management Studio, fare clic con il pulsante destro del mouse su server e selezionare Proprietà. Selezionare Connessioni dall'elenco, quindi selezionare la casella di controllo Consenti connessioni remote a questo server.

   

   Per la procedura dettagliata, vedere Configurare l'opzione di configurazione del server di accesso remoto.

2. Avviare Gestione configurazione SQL Server. Espandere Configurazione di rete SQL Server per l'istanza prevista e selezionare Protocolli per MSSQLSERVER. I protocolli vengono visualizzati nel riquadro destro. Abilitare TCP/IP facendo clic con il pulsante destro del mouse su TCP/IP e selezionando Abilita.

   

   Per informazioni dettagliate e modi alternativi per abilitare il protocollo TCP/IP, vedere Abilitare o disabilitare un protocollo di rete del server.

3. Nella stessa finestra, fare doppio clic su TCP/IP per aprire la finestra Proprietà TCP/IP.

4. Passare alla scheda Indirizzi IP . Scorrere in basso per vedere la sezione IPAll. Prendere nota della Porta TCP. Il valore predefinito è 1433.

5. Creare una regola per Windows Firewall nel computer per consentire il traffico in ingresso attraverso questa porta.

6. Verifica connessione: per connettersi al server SQL usando un nome completo, usare SQL Server Management Studio da un computer diverso. Un esempio è "<machine>.<domain>.corp.<company>.com,1433".

Aggiornare la versione di SQL Server

Per aggiornare la versione di SQL Server, nella pagina Modifica servizio collegato, selezionare Consigliato in Versione e configurare il servizio collegato facendo riferimento a Proprietà del servizio collegato per la versione consigliata.

Differenze tra la versione consigliata e la versione legacy

La tabella seguente illustra le differenze tra SQL Server usando la versione consigliata e la versione legacy.

Contenuto correlato

Per un elenco degli archivi dati supportati come origini e sink dall'attività di copia, vedere Archivi dati supportati.