Copiare e trasformare i dati da e verso SQL Server usando Azure Data Factory o Azure Synapse Analytics
SI APPLICA A:
Azure Data Factory Azure Synapse Analytics
Suggerimento
Provare Data Factory in Microsoft Fabric, una soluzione di analisi completa per le aziende. Microsoft Fabric copre tutti gli elementi, dallo spostamento dei dati all'analisi scientifica dei dati, all'analisi in tempo reale, alla business intelligence e alla creazione di report. Scopri come avviare gratuitamente una nuova versione di valutazione .
Questo articolo illustra come usare l'attività di copia nelle pipeline di Azure Data Factory e Azure Synapse per copiare dati da e verso il database di SQL Server e usare Flusso di dati per trasformare i dati nel database di 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:
| Funzionalità supportate | IR |
|---|---|
| Attività di copia (origine/sink) | (1) (2) |
| Flusso di dati di mapping (origine/sink) | (1) |
| Attività Lookup | (1) (2) |
| Attività GetMetadata | (1) (2) |
| Attività script | (1) (2) |
| Attività stored procedure | (1) (2) |
(1) Runtime di integrazione di Azure (2) 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, il recupero dei dati tramite una query SQL o una stored procedure. Per informazioni dettagliate, è 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, la creazione automatica della tabella di destinazione se non esiste in base allo schema di origine; aggiunta di dati a una tabella o richiamo di una stored procedure con logica personalizzata durante la copia.
Sql Server Express Local DB non è supportato.
Importante
L'origine dati deve supportare il tipo di dati NVARCHAR perché influisce sulla codifica dei dati quando viene applicata una codifica non universale ai dati.
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 cloud gestito, è possibile usare Azure Integration Runtime. Se l'accesso è limitato agli indirizzi IP approvati nelle regole del firewall, è possibile aggiungere indirizzi IP del runtime di integrazione di Azure all'elenco elementi consentiti.
È anche possibile usare la funzionalità di 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
- The Python SDK
- Azure PowerShell
- The REST API
- Modello di Azure Resource Manager
Creare un servizio collegato di SQL Server usando l'interfaccia utente
Usare la procedura seguente per creare un servizio collegato di SQL Server nell'interfaccia utente di portale di Azure.
Passare alla scheda Gestisci nell'area di lavoro di Azure Data Factory o Synapse e selezionare Servizi collegati, quindi fare clic su Nuovo:
Cercare SQL e selezionare il connettore SQL Server.
Configurare i dettagli del servizio, testare la connessione e creare il nuovo servizio collegato.
Dettagli di configurazione di Connessione or
Le sezioni seguenti forniscono informazioni dettagliate sulle proprietà usate per definire entità della pipeline di Data Factory e Synapse specifiche per il connettore di database di SQL Server.
Proprietà del servizio collegato
Questo connettore SQL Server supporta i tipi di autenticazione seguenti. Per informazioni dettagliate, vedere le sezioni corrispondenti.
Suggerimento
Se si verifica un errore con il codice di errore "UserErrorFailedTo Connessione ToSqlServer" e un messaggio simile a "Il limite di sessione per il database è XXX ed è stato raggiunto", aggiungere Pooling=false al stringa di connessione e riprovare.
Autenticazione SQL
Per usare l'autenticazione SQL, sono supportate le proprietà seguenti:
| Proprietà | Descrizione | Richiesto |
|---|---|---|
| type | La proprietà type deve essere impostata su SqlServer. | Sì |
| connectionString | Specificare le informazioni connectionString necessarie per connettersi al database di SQL Server. Specificare un nome di accesso come nome utente e assicurarsi che il database a cui si vuole connettersi sia mappato a questo account di accesso. Vedere gli esempi seguenti. | Sì |
| password | Per inserire una password in Azure Key Vault, estrarre la password configurazione dal stringa di connessione. Per altre informazioni, vedere l'esempio JSON che segue la tabella e Archiviare le credenziali in Azure Key Vault. |
No |
| alwaysEncrypted Impostazioni | Specificare le informazioni alwaysencryptedsettings necessarie per abilitare Always Encrypted per proteggere i dati sensibili archiviati in SQL Server usando l'identità gestita o l'entità servizio. Per altre informazioni, vedere l'esempio JSON che segue la tabella e la sezione Uso di Always Encrypted . Se non specificato, l'impostazione predefinita è sempre crittografata è disabilitata. | No |
| connectVia | Questo runtime di integrazione viene usato per connettersi all'archivio dati. Per altre informazioni, vedere la sezione Prerequisiti. Se non specificato, viene usato il runtime di integrazione di Azure predefinito. | No |
Esempio: Usare l'autenticazione SQL
{
"name": "SqlServerLinkedService",
"properties": {
"type": "SqlServer",
"typeProperties": {
"connectionString": "Data Source=<servername>\\<instance name if using named instance>;Initial Catalog=<databasename>;Integrated Security=False;User ID=<username>;Password=<password>;"
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
Esempio: Usare l'autenticazione SQL con una password in Azure Key Vault
{
"name": "SqlServerLinkedService",
"properties": {
"type": "SqlServer",
"typeProperties": {
"connectionString": "Data Source=<servername>\\<instance name if using named instance>;Initial Catalog=<databasename>;Integrated Security=False;User ID=<username>;",
"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": {
"connectionString": "Data Source=<servername>\\<instance name if using named instance>;Initial Catalog=<databasename>;Integrated Security=False;User ID=<username>;Password=<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 autenticazione di Windows, sono supportate le proprietà seguenti:
| Proprietà | Descrizione | Richiesto |
|---|---|---|
| type | La proprietà type deve essere impostata su SqlServer. | Sì |
| connectionString | Specificare le informazioni connectionString necessarie per connettersi al database di SQL Server. Vedere gli esempi seguenti. | |
| userName | Consente di specificare un nome utente. Un esempio è domainname\username. | Sì |
| password | Specificare una password per l'account utente specificato per il nome utente. Contrassegnare questo campo come SecureString per archiviarlo in modo sicuro. In alternativa, fare riferimento a un segreto archiviato in Azure Key Vault. | Sì |
| alwaysEncrypted Impostazioni | Specificare le informazioni alwaysencryptedsettings necessarie per abilitare Always Encrypted per proteggere i dati sensibili archiviati in SQL Server usando l'identità gestita o l'entità servizio. Per altre informazioni, vedere La sezione Uso di Always Encrypted . Se non specificato, l'impostazione predefinita è sempre crittografata è disabilitata. | No |
| connectVia | Questo runtime di integrazione viene usato per connettersi all'archivio dati. Per altre informazioni, vedere la sezione Prerequisiti. Se non specificato, viene usato il runtime di integrazione di Azure predefinito. | No |
Nota
autenticazione di Windows non è supportato nel flusso di dati.
Esempio: Usare autenticazione di Windows
{
"name": "SqlServerLinkedService",
"properties": {
"type": "SqlServer",
"typeProperties": {
"connectionString": "Data Source=<servername>\\<instance name if using named instance>;Initial Catalog=<databasename>;Integrated Security=True;",
"userName": "<domain\\username>",
"password": {
"type": "SecureString",
"value": "<password>"
}
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
Esempio: Usare autenticazione di Windows con una password in Azure Key Vault
{
"name": "SqlServerLinkedService",
"properties": {
"annotations": [],
"type": "SqlServer",
"typeProperties": {
"connectionString": "Data Source=<servername>\\<instance name if using named instance>;Initial Catalog=<databasename>;Integrated Security=True;",
"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"
}
}
}
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. In questa sezione viene fornito un elenco delle proprietà supportate dal set di dati di SQL Server.
Per copiare dati da e in un database di SQL Server, sono supportate le proprietà seguenti:
| Proprietà | Descrizione | Richiesto |
|---|---|---|
| type | La proprietà type del set di dati deve essere impostata su SqlServerTable. | Sì |
| schema | Nome dello schema. | No per l'origine, Sì per il sink |
| table | Nome della tabella/vista. | No per l'origine, Sì per il sink |
| tableName | Nome della tabella/vista con schema. Questa proprietà è supportata per garantire la compatibilità con le versioni precedenti. Per i nuovi carichi di lavoro, usare schema e table. |
No per l'origine, Sì per il sink |
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. In questa sezione viene fornito un elenco delle proprietà supportate dall'origine e dal sink di SQL Server.
SQL Server come origine
Suggerimento
Per caricare i dati da SQL Server in modo efficiente usando il partizionamento dei dati, vedere Copia parallela dal database SQL.
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:
| Proprietà | Descrizione | Richiesto |
|---|---|---|
| type | La proprietà type dell'origine dell'attività di copia deve essere impostata su SqlSource. | Sì |
| sqlReaderQuery | Usare la query SQL personalizzata per leggere i dati. Un esempio è select * from MyTable. |
No |
| sqlReaderStoredProcedureName | Questa proprietà definisce il nome della stored procedure che legge i dati dalla tabella di origine. L'ultima istruzione SQL deve essere un'istruzione SELECT nella stored procedure. | No |
| storedProcedureParameters | Questi parametri sono relativi alla stored procedure. I valori consentiti sono coppie nome-valore. I nomi e le maiuscole/minuscole dei parametri devono corrispondere ai nomi e alla combinazione di maiuscole e minuscole dei parametri della stored procedure. |
No |
| isolationLevel | Specifica il comportamento di blocco della transazione per l'origine SQL. I valori consentiti sono: ReadCommitted, ReadUncommitted, RepeatableRead, Serializable, Snapshot. Se non specificato, viene utilizzato il livello di isolamento predefinito del database. Per altre informazioni dettagliate, vedere questo documento. | No |
| partitionOptions | Specifica le opzioni di partizionamento dei dati usate per caricare dati da SQL Server. I valori consentiti sono: Nessuno (impostazione predefinita), PhysicalPartitionsOfTable e DynamicRange. Quando un'opzione di partizione è abilitata ( ovvero , non None), il grado di parallelismo per caricare simultaneamente i dati da SQL Server è controllato dall'impostazione nell'attività parallelCopies di copia. |
No |
| partitionSettings | Specifica il gruppo di impostazioni per il partizionamento dei dati. Applicare quando l'opzione di partizione non Noneè . |
No |
In partitionSettings: |
||
| partitionColumnName | Specificare il nome della colonna di origine nel tipo integer o date/datetime (int, smallint, datebigint, smalldatetimedatetimedatetime2, o datetimeoffset) che verrà usato dal partizionamento dell'intervallo per la copia parallela. Se non specificato, l'indice o la chiave primaria della tabella vengono rilevati automaticamente e usati come colonna di partizione.Si applica quando l'opzione di partizione è DynamicRange. Se si usa una query per recuperare i dati di origine, associare ?AdfDynamicRangePartitionCondition nella clausola WHERE. Per un esempio, vedere la sezione Copia parallela dal database SQL. |
No |
| partitionUpperBound | Valore massimo della colonna di partizione per la suddivisione dell'intervallo di partizioni. Questo valore viene usato per decidere lo stride della partizione, non per filtrare le righe nella tabella. Tutte le righe nella tabella o nel risultato della query verranno partizionate e copiate. Se non specificato, l'attività di copia rileva automaticamente il valore. Si applica quando l'opzione di partizione è DynamicRange. Per un esempio, vedere la sezione Copia parallela dal database SQL. |
No |
| partitionLowerBound | Valore minimo della colonna di partizione per la suddivisione dell'intervallo di partizioni. Questo valore viene usato per decidere lo stride della partizione, non per filtrare le righe nella tabella. Tutte le righe nella tabella o nel risultato della query verranno partizionate e copiate. Se non specificato, l'attività di copia rileva automaticamente il valore. Si applica quando l'opzione di partizione è DynamicRange. Per un esempio, vedere la sezione Copia parallela dal database SQL. |
No |
Tenere presente quanto segue:
- Se sqlReaderQuery viene specificato per SqlSource, l'attività di copia esegue questa query sull'origine 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 i dati, si noti se la stored procedure è progettata come restituzione di uno schema diverso quando viene passato un valore di parametro diverso, è possibile che si verifichi un errore o si verifichi 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
Suggerimento
Altre informazioni sui comportamenti di scrittura, le configurazioni e le procedure consigliate supportate sono disponibili in Procedure consigliate per il caricamento dei dati in SQL Server.
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:
| Proprietà | Descrizione | Richiesto |
|---|---|---|
| type | La proprietà type del sink dell'attività di copia deve essere impostata su SqlSink. | Sì |
| preCopyScript | Questa proprietà specifica una query SQL per l'attività di copia da eseguire prima di scrivere dati in SQL Server. Viene richiamata solo una volta per ogni esecuzione della copia. È possibile usare questa proprietà per pulire i dati precaricati. | No |
| tableOption | Specifica se creare automaticamente la tabella sink se non esiste in base allo schema di origine. La creazione automatica della tabella non è supportata quando il sink specifica la stored procedure. I valori consentiti sono: none (impostazione predefinita), autoCreate. |
No |
| sqlWriterStoredProcedureName | Il nome della stored procedure che definisce come applicare i dati di origine in una tabella di destinazione. Questa stored procedure viene richiamata per batch. Per le operazioni eseguite una sola volta e non hanno nulla a che fare con i dati di origine, ad esempio, eliminare o troncare, usare la preCopyScript proprietà .Vedere l'esempio di Richiamare una stored procedure da un sink SQL. |
No |
| storedProcedureTableTypeParameterName | Nome del parametro del tipo di tabella specificato nella stored procedure. | No |
| sqlWriterTableType | Nome del tipo di tabella da utilizzare nella stored procedure. Nel corso dell'attività di copia, i dati spostati vengono resi disponibili in una tabella temporanea di questo tipo. Il codice della stored procedure può quindi unire i dati di cui è in corso la copia con i dati esistenti. | No |
| storedProcedureParameters | Parametri per la stored procedure. I valori consentiti sono coppie nome-valore. I nomi e le maiuscole e minuscole dei parametri devono corrispondere ai nomi e alle maiuscole e minuscole dei parametri della stored procedure. |
No |
| writeBatchSize | Numero di righe da inserire nella tabella SQL per batch. I valori consentiti sono integer per il numero di righe. Per impostazione predefinita, il servizio determina in modo dinamico le dimensioni del batch appropriate in base alle dimensioni della riga. |
No |
| writeBatchTimeout | Tempo di attesa per il completamento dell'operazione di inserimento, upsert e stored procedure prima del timeout. I valori consentiti sono relativi all'intervallo di tempo. Un esempio è "00:30:00" per 30 minuti. Se non viene specificato alcun valore, per impostazione predefinita il timeout è "00:30:00". |
No |
| maxConcurrent Connessione ions | Limite massimo di connessioni simultanee stabilite all'archivio dati durante l'esecuzione dell'attività. Specificare un valore solo quando si desidera limitare le connessioni simultanee. | No |
| WriteBehavior | Specificare il comportamento di scrittura per l'attività di copia per caricare i dati nel database di SQL Server. Il valore consentito è Insert e Upsert. Per impostazione predefinita, il servizio usa insert per caricare i dati. |
No |
| upsert Impostazioni | Specificare il gruppo delle impostazioni per il comportamento di scrittura. Applicare quando l'opzione WriteBehavior è Upsert. |
No |
In upsertSettings: |
||
| useTempDB | Specificare se utilizzare una tabella temporanea globale o una tabella fisica come tabella provvisoria per upsert. Per impostazione predefinita, il servizio usa una tabella temporanea globale come tabella provvisoria. value è true. |
No |
| interimSchemaName | Specificare lo schema provvisorio per la creazione di una tabella provvisoria se viene utilizzata la tabella fisica. Nota: l'utente deve disporre dell'autorizzazione per la creazione e l'eliminazione di una tabella. Per impostazione predefinita, la tabella provvisoria condividerà lo stesso schema della tabella sink. Applicare quando l'opzione useTempDB è False. |
No |
| keys | Specificare i nomi di colonna per l'identificazione univoca delle righe. È possibile usare una singola chiave o una serie di chiavi. Se non specificato, viene usata la chiave primaria. | No |
Esempio 1: Aggiungere 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 data
"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à di copia fornisce il partizionamento dei dati predefinito per copiare i dati in parallelo. È possibile trovare le opzioni di partizionamento dei dati nella scheda 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. Ad esempio, se si imposta parallelCopies su quattro, il servizio genera e esegue simultaneamente quattro query in base all'opzione e alle impostazioni di partizione specificate e ogni query recupera una parte di dati da SQL Server.
È consigliabile abilitare la copia parallela con il partizionamento dei dati, soprattutto quando si caricano grandi quantità di dati da 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 più file (specificare solo il nome della cartella), nel qual caso le prestazioni sono migliori rispetto alla scrittura in un singolo file.
| Scenario | Impostazioni consigliate |
|---|---|
| Caricamento completo da tabelle di grandi dimensioni, con partizioni fisiche. | Opzione di partizione: partizioni fisiche della tabella. Durante l'esecuzione, il servizio rileva automaticamente le partizioni fisiche e copia i dati in base alle partizioni. Per verificare se la tabella contiene o meno una partizione fisica, è possibile fare riferimento a questa query. |
| Caricamento completo da una tabella di grandi dimensioni, senza partizioni fisiche, mentre con una colonna integer o datetime per il partizionamento dei dati. | Opzioni di partizione: partizione intervallo dinamico. Colonna di partizione (facoltativa): specificare la colonna usata per partizionare i dati. Se non specificato, viene utilizzata la colonna chiave primaria. Limite superiore della partizione e limite inferiore della partizione (facoltativo): specificare se si desidera determinare lo stride della partizione. Non si tratta di filtrare le righe nella tabella, tutte le righe della tabella verranno partizionate e copiate. Se non specificato, l'attività di copia rileva automaticamente i valori e può richiedere molto tempo a seconda dei valori MIN e MAX. È consigliabile specificare un limite superiore e un limite inferiore. Ad esempio, se la colonna di partizione "ID" ha valori compresi tra 1 e 100 e si imposta il limite inferiore su 20 e il limite superiore come 80, con copia parallela come 4, il servizio recupera i dati rispettivamente per 4 partizioni - ID nell'intervallo <=20, [21, 50], [51, 80]e >=81. |
| Caricare una grande quantità di dati usando una query personalizzata, senza partizioni fisiche, mentre con una colonna integer o date/datetime per il partizionamento dei dati. | Opzioni di partizione: partizione intervallo dinamico. Query: SELECT * FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition AND <your_additional_where_clause>.Colonna di partizione: specificare la colonna usata per partizionare i dati. Limite superiore della partizione e limite inferiore della partizione (facoltativo): specificare se si desidera determinare lo stride della partizione. Non si tratta di filtrare le righe nella tabella, tutte le righe nel risultato della query verranno partizionate e copiate. Se non specificato, l'attività di copia rileva automaticamente il valore. Durante l'esecuzione, il servizio sostituisce ?AdfRangePartitionColumnName con il nome di colonna e gli intervalli di valori effettivi per ogni partizione e invia a SQL Server. Ad esempio, se la colonna di partizione "ID" include valori compresi tra 1 e 100 e si imposta il limite inferiore su 20 e il limite superiore come 80, con copia parallela come 4, il servizio recupera i dati per 4 partizioni- ID nell'intervallo <=20, [21, 50], [51, 80], e >=81 rispettivamente. Di seguito sono riportate altre query di esempio per diversi scenari: 1. Eseguire una query sull'intera tabella: SELECT * FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition2. Eseguire una query da una tabella con la selezione di colonne e filtri aggiuntivi della clausola where: SELECT <column_list> FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition AND <your_additional_where_clause>3. Eseguire una query con sottoquery: SELECT <column_list> FROM (<your_sub_query>) AS T WHERE ?AdfDynamicRangePartitionCondition AND <your_additional_where_clause>4. Eseguire una query con la partizione nella sottoquery: SELECT <column_list> FROM (SELECT <your_sub_query_column_list> FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition) AS T |
Procedure consigliate per caricare i dati con l'opzione di partizione:
- Scegliere una colonna distintiva come colonna di partizione (ad esempio chiave primaria o chiave univoca) per evitare l'asimmetria dei dati.
- Se la tabella include una partizione predefinita, usare l'opzione di partizione "Partizioni fisiche della tabella" per ottenere prestazioni migliori.
- Se si usa Azure Integration Runtime per copiare i dati, è possibile impostare unità di Integrazione dei dati (DIU) di dimensioni maggiori (>4) per usare più risorse di calcolo. Controllare gli scenari applicabili.
- "Grado di parallelismo di copia" controlla i numeri di partizione, impostando questo numero troppo grande a volte le prestazioni, è consigliabile impostare questo numero come (DIU o numero di nodi del runtime di integrazione self-hosted) * (da 2 a 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 ?AdfDynamicRangePartitionCondition 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:
- Append: i dati di origine hanno solo nuovi record.
- Upsert: i dati di origine hanno sia inserimenti che aggiornamenti.
- Sovrascrittura: si vuole ricaricare l'intera tabella delle dimensioni ogni volta.
- Scrivere con logica personalizzata: è necessaria un'elaborazione aggiuntiva prima dell'inserimento finale nella tabella di destinazione.
Vedere le rispettive sezioni per informazioni su come configurare e procedure consigliate.
Accodare dati
L'aggiunta 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 l'origine e il sink di conseguenza nell'attività di copia.
Eseguire l'upsert dei dati
attività Copy supporta ora il caricamento nativo dei dati in una tabella temporanea del database e quindi l'aggiornamento dei dati nella tabella sink se la chiave esiste e in caso contrario inserisce nuovi dati. Per altre informazioni sulle impostazioni 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. Esegue quindi la copia per inserire i dati. Ad esempio, per sovrascrivere l'intera tabella con i dati più recenti, specificare uno script per eliminare prima di eseguire il caricamento bulk dei nuovi dati dall'origine.
Scrivere dati con logica personalizzata
I passaggi per scrivere dati con logica personalizzata sono simili a quelli descritti nella sezione Dati Upsert . Quando è necessario 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 nel sink dell'attività di copia per applicare i dati.
Richiamare una stored procedure da un sink SQL
Quando si copiano dati nel database di 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. Si noti che il servizio esegue automaticamente il wrapping della stored procedure nella propria transazione, pertanto qualsiasi 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. Un 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 supponga che i dati di input e la tabella Marketing sink abbiano ognuna tre colonne: ProfileID, State e Category. Eseguire l'upsert in base alla colonna ProfileID e applicarlo solo per una categoria specifica denominata "ProductA".
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 )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 è uguale a 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); ENDDefinire la sezione sink SQL nell'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
Quando si trasformano i dati nel flusso di dati di mapping, è possibile leggere e scrivere nelle tabelle dal database di SQL Server. Per altre informazioni, vedere la trasformazione origine e la trasformazione sink nei flussi di dati per mapping.
Nota
Per accedere a SQL Server locale, è necessario usare Azure Data Factory o l'area di lavoro gestita di Synapse Rete virtuale usando un endpoint privato. Per informazioni dettagliate, vedere questa esercitazione .
Trasformazione origine
Nella tabella seguente sono elencate le proprietà supportate dall'origine SQL Server. È possibile modificare queste proprietà nella scheda Opzioni origine.
| Nome | Descrizione | Richiesto | Valori consentiti | Proprietà script flusso di dati |
|---|---|---|---|---|
| Tabella | Se si seleziona Tabella come input, il flusso di dati recupera tutti i dati dalla tabella specificata nel set di dati. | No | - | - |
| Query | Se si seleziona Query come input, specificare una query SQL per recuperare i dati dall'origine, che esegue l'override di qualsiasi tabella specificata nel set di dati. L'uso delle query è un ottimo modo per ridurre le righe per i test o le ricerche. La clausola Order By non è supportata, ma è possibile impostare un'istruzione FROM edizione Standard LECT completa. È possibile usare anche funzioni di tabella definite dall'utente. select * from udfGetData() è una funzione definita dall'utente in SQL che restituisce una tabella che è possibile usare nel flusso di dati. Esempio di query: Select * from MyTable where customerId > 1000 and customerId < 2000 |
No | String | query |
| Dimensioni batch | Specificare una dimensione batch per suddividere i dati di grandi dimensioni in letture. | No | Intero | batchSize |
| Livello di isolamento | Scegliere uno dei livelli di isolamento seguenti: - Read Committed - Read Uncommitted (impostazione predefinita) - Lettura ripetibile -Serializzabile - Nessuno (ignora il livello di isolamento) |
No | READ_COMMITTED READ_UNCOMMITTED REPEATABLE_READ SERIALIZABLE NONE |
isolationLevel |
| Abilitare l'estrazione incrementale | Usare questa opzione per indicare ad ADF di elaborare solo le righe modificate dall'ultima esecuzione della pipeline. | No | - | - |
| Colonna data incrementale | Quando si usa la funzionalità di estrazione incrementale, è necessario scegliere la colonna data/ora da usare come filigrana nella tabella di origine. | No | - | - |
| Abilitare Change Data Capture nativo (anteprima) | Usare questa opzione per indicare ad ADF di elaborare solo i dati differenziali acquisiti dalla tecnologia SQL Change Data Capture dall'ultima esecuzione della pipeline. Con questa opzione, i dati differenziali che includono inserimento di righe, aggiornamento ed eliminazione verranno caricati automaticamente senza alcuna colonna di data incrementale necessaria. È necessario abilitare Change Data Capture in SQL Server prima di usare questa opzione in Azure Data Factory. Per altre informazioni su questa opzione in Azure Data Factory, vedere Change Data Capture nativo. | No | - | - |
| Iniziare a leggere dall'inizio | L'impostazione di questa opzione con l'estrazione incrementale indicherà a ADF di leggere tutte le righe alla prima esecuzione di una pipeline con l'estrazione incrementale attivata. | No | - | - |
Suggerimento
L'espressione di tabella comune (CTE) in SQL non è supportata nella modalità query del flusso di dati di mapping, perché il prerequisito dell'uso di questa modalità è che le query possono essere usate nella clausola FROM della query SQL, ma non è possibile eseguire questa operazione. Per usare le CTE, è necessario creare una stored procedure usando la query seguente:
CREATE PROC CTESP @query nvarchar(max)
AS
BEGIN
EXECUTE sp_executesql @query;
END
Usare quindi la modalità Stored procedure nella trasformazione di origine del flusso di dati di mapping e impostare l'esempio @querywith CTE as (select 'test' as a) select * from CTE. È quindi possibile usare le CTE come previsto.
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.
| Nome | Descrizione | Richiesto | Valori consentiti | Proprietà script flusso di dati |
|---|---|---|---|---|
| Metodo di aggiornamento | Specificare le operazioni consentite nella destinazione del database. Per impostazione predefinita, sono consentiti solo gli inserimenti. Per aggiornare, upsert o eliminare righe, è necessaria una trasformazione Alter row per contrassegnare le righe per tali azioni. |
Sì | true oppure false |
deletable Inseribile Aggiornabile aggiornabile |
| Colonne chiave | Per gli aggiornamenti, gli upsert e le eliminazioni, le colonne chiave devono essere impostate per determinare quale riga modificare. Il nome della colonna selezionato come chiave verrà usato come parte dell'aggiornamento successivo, dell'upsert, dell'eliminazione. Pertanto, è necessario selezionare una colonna esistente nel mapping sink. |
No | Matrice | keys |
| Ignora scrittura colonne chiave | Se si desidera non scrivere il valore nella colonna chiave, selezionare "Ignora la scrittura di colonne chiave". | No | true oppure false |
skipKeyWrites |
| azione Tabella | determina se ricreare o rimuovere tutte le righe dalla tabella di destinazione prima della scrittura. - Nessuno: nessuna azione verrà eseguita nella tabella. - Ricrea: la tabella verrà eliminata e ricreata. Questa opzione è obbligatoria se si crea una nuova tabella in modo dinamico. - Troncamento: tutte le righe della tabella di destinazione verranno rimosse. |
No | true oppure false |
Ricreare truncate |
| Dimensioni batch | Specificare il numero di righe scritte in ogni batch. Dimensioni batch più grandi migliorano l'ottimizzazione della compressione e della memoria, ma rischiano di causare eccezioni di memoria insufficiente durante la memorizzazione nella cache dei dati. | No | Intero | batchSize |
| Pre e post-script SQL | Specificare script SQL su più righe che verranno eseguiti prima (pre-elaborazione) e dopo la scrittura dei dati (post-elaborazione) nel database Sink. | No | String | preSQLs postSQLs |
Suggerimento
- È consigliabile suddividere singoli script batch con più comandi in più batch.
- Possono essere eseguite come parte di un batch solo le istruzioni DDL (Data Definition Language) e DML (Data Manipulation Language) che restituiscono un semplice conteggio di aggiornamento. Per altre informazioni, vedere Esecuzione di operazioni batch
Esempio di script 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 e in SQL Server, vengono usati i mapping seguenti dai tipi di dati di SQL Server ai 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.
| Tipo di dati di SQL Server | Tipo di dati provvisorio di Data Factory |
|---|---|
| bigint | Int64 |
| binary | Byte[] |
| bit | Booleano |
| char | String, Char[] |
| data | DataOra |
| Datetime | DataOra |
| datetime2 | Data/Ora |
| Datetimeoffset | DateTimeOffset |
| Decimale | Decimale |
| FILESTREAM attribute (varbinary(max)) | Byte[] |
| Float | Double |
| image | Byte[] |
| int | Int32 |
| money | Decimale |
| nchar | String, Char[] |
| ntext | String, Char[] |
| numeric | Decimale |
| nvarchar | String, Char[] |
| real | Singola |
| rowversion | Byte[] |
| smalldatetime | Data/Ora |
| smallint | Int16 |
| smallmoney | Decimale |
| sql_variant | Object |
| Testo | String, Char[] |
| Ora | TimeSpan |
| timestamp | Byte[] |
| tinyint | Int16 |
| uniqueidentifier | GUID |
| varbinary | Byte[] |
| varchar | String, Char[] |
| xml | String |
Nota
Per i tipi di dati mappati al tipo provvisorio Decimal, attualmente attività Copy supporta la precisione fino a 28. Se si hanno dati che richiedono una precisione maggiore di 28, è consigliabile convertirli in una stringa in una query SQL.
Quando si copiano dati da SQL Server usando Azure Data Factory, viene eseguito il mapping del tipo di dati bit al tipo di dati provvisorio Boolean. Se sono presenti dati che devono essere mantenuti come tipo di dati di bit, usare query con T-SQL CAST o CONVERT.
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/a SQL Server con Always Encrypted, seguire questa procedura:
Archiviare la chiave master della colonna in un insieme di credenziali delle chiavi di Azure. Altre informazioni su come configurare Always Encrypted con Azure Key Vault
Assicurarsi di concedere l'accesso all'insieme di credenziali delle chiavi in cui è archiviata la chiave master della colonna . Per le autorizzazioni necessarie, vedere questo articolo .
Creare un servizio collegato per connettersi al database SQL e abilitare la funzione "Always Encrypted" usando l'identità gestita o l'entità servizio.
Nota
SQL Server Always Encrypted supporta gli scenari seguenti:
- Gli archivi dati di origine o sink usano l'identità gestita o l'entità servizio come tipo di autenticazione del provider di chiavi.
- Gli archivi dati di origine e sink usano l'identità gestita come tipo di autenticazione del provider di chiavi.
- Gli archivi dati di origine e sink usano la stessa entità servizio del tipo di autenticazione del provider di chiavi.
Nota
Sql Server Always Encrypted è attualmente supportato solo per la trasformazione dell'origine nei flussi di dati di mapping.
Acquisizione dei dati delle modifiche nativa
Azure Data Factory può supportare funzionalità native di Change Data Capture per SQL Server, il database SQL di Azure e l'istanza gestita di SQL di Azure. I dati modificati, inclusi inserimento di righe, aggiornamento ed eliminazione negli archivi SQL, possono essere rilevati ed estratti automaticamente dal flusso di dati di mapping di Azure Data Factory. Senza esperienza di codice nel flusso di dati di mapping, gli utenti possono ottenere facilmente uno scenario di replica dei dati dagli archivi SQL aggiungendo un database come archivio di destinazione. Inoltre, gli utenti possono anche comporre qualsiasi logica di trasformazione dei dati tra per ottenere uno scenario ETL incrementale dagli archivi SQL.
Assicurarsi di mantenere invariato il nome della pipeline e dell'attività, in modo che il checkpoint possa essere registrato da ADF per ottenere i dati modificati dall'ultima esecuzione automaticamente. Se si modifica il nome della pipeline o il nome dell'attività, il checkpoint verrà reimpostato, che consente di iniziare dall'inizio o ottenere modifiche da ora nell'esecuzione successiva. Se si vuole modificare il nome della pipeline o il nome dell'attività, ma mantenere comunque il checkpoint per ottenere i dati modificati dall'ultima esecuzione automaticamente, 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. Dopo aver soddisfatto il risultato della pipeline dall'esecuzione di debug, è 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 ora in poi.
Nella sezione di monitoraggio è sempre possibile eseguire di nuovo 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 a cui si fa riferimento al set di dati abilitato per SQL CDC con una trasformazione sink a cui si fa riferimento a un database in un flusso di dati di 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 i database. È possibile usare il metodo di aggiornamento nella trasformazione sink per selezionare se si desidera consentire l'inserimento, consentire l'aggiornamento o consentire 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 vuole abilitare lo scenario ETL anziché la replica dei dati tra database tramite SQL CDC, è possibile usare espressioni nel flusso di dati di mapping, tra cui 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 mapping del flusso di dati 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 elaborare i dati differenziali.
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 le modifiche nette da SQL CDC verranno caricate da ADF tramite cdc.fn_cdc_get_net_changes_.
Risolvere i problemi di connessione
Configurare l'istanza di SQL Server per accettare connessioni remote. Avviare SQL Server Management Studio, fare clic con il pulsante destro del mouse sul server e scegliere Proprietà. Selezionare Connessione ions dall'elenco e selezionare la casella di controllo Consenti connessioni remote a questo server.
Per i passaggi dettagliati, vedere Configurare l'opzione di configurazione del server di accesso remoto.
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 scegliendo Abilita.
Per altre informazioni e modi alternativi per abilitare il protocollo TCP/IP, vedere Abilitare o disabilitare un protocollo di rete del server.
Nella stessa finestra fare doppio clic su TCP/IP per avviare la finestra Proprietà TCP/IP.
Passare alla scheda Indirizzi IP. Scorrere verso il basso per visualizzare la sezione IPAll . Annotare la porta TCP. Il valore predefinito è 1433.
Creare una regola per Windows Firewall nel computer per consentire il traffico in ingresso attraverso questa porta.
Verificare la connessione: per connettersi a SQL Server usando un nome completo, usare SQL Server Management Studio da un computer diverso. Un esempio è
"<machine>.<domain>.corp.<company>.com,1433".
Contenuto correlato
Per un elenco degli archivi dati supportati come origini e sink dall'attività di copia, vedere Archivi dati supportati.