Carga incremental de datos de Azure SQL Database a Azure Blob Storage mediante la información de seguimiento de cambios mediante PowerShell

SE APLICA A: Azure Data Factory Azure Synapse Analytics

Sugerencia

Data Factory en Microsoft Fabric es la próxima generación de Azure Data Factory, con una arquitectura más sencilla, inteligencia artificial integrada y nuevas características. Si no está familiarizado con la integración de datos, comience con Fabric Data Factory. Las cargas de trabajo de ADF existentes pueden actualizarse a Fabric para acceder a nuevas funcionalidades en ciencia de datos, análisis en tiempo real e informes.

En este tutorial, creará una factoría de datos de Azure con una canalización que carga los datos diferenciales según la información de control de cambios de la base de datos de origen de Azure SQL Database a una instancia de Azure Blob Storage.

En este tutorial, realizará los siguientes pasos:

Preparación del almacenamiento de datos de origen
Creación de una factoría de datos.
Cree servicios vinculados.
Cree conjuntos de datos de origen, receptor y seguimiento de cambios.
Creación, ejecución y supervisión de la canalización de copia completa
Adición o actualización de datos en la tabla de origen
Crea, ejecuta y supervisa el pipeline de copia incremental

Nota:

Se recomienda usar el módulo Az de PowerShell de Azure para interactuar con Azure. Para empezar, consulte Install Azure PowerShell. Para obtener información sobre cómo migrar al módulo Az PowerShell, consulte Migrate Azure PowerShell de AzureRM a Az.

Información general

En una solución de integración de datos, la carga incremental de datos tras las cargas iniciales de datos es un método ampliamente utilizado. En algunos casos, los datos modificados en un período en el almacén de datos de origen pueden ser fácilmente divididos en partes (por ejemplo, LastModifyTime, CreationTime). En algunos casos, no hay una manera explícita de identificar los datos delta desde la última vez que se procesaron. La tecnología Change Tracking compatible con almacenes de datos como Azure SQL Database y SQL Server se pueden usar para identificar los datos delta. En este tutorial se describe cómo usar Azure Data Factory con la tecnología sql Change Tracking para cargar incrementalmente datos delta de Azure SQL Database en Azure Blob Storage. Para obtener información más concreta sobre la tecnología de seguimiento de cambios de SQL, consulte Seguimiento de cambios en SQL Server.

Flujo de trabajo de un extremo a otro

Estos son los pasos del flujo de trabajo completo típico para cargar incrementalmente los datos mediante la tecnología de control de cambios.

Nota:

Tanto Azure SQL Database como SQL Server admiten la tecnología Change Tracking. En este tutorial se usa Azure SQL Database como almacén de datos de origen. También puede usar una instancia de SQL Server.

Carga inicial de datos históricos (ejecutar una vez):
1. Habilite la tecnología Change Tracking en la base de datos de origen en Azure SQL Database.
2. Obtenga el valor inicial de SYS_CHANGE_VERSION en la base de datos como base de referencia para capturar los datos que han cambiado.
3. Cargue datos completos de la base de datos de origen en un almacenamiento de blobs de Azure.
Carga incremental de los datos diferenciales según una programación (ejecutar periódicamente después de la carga inicial de datos):
1. Obtenga los valores SYS_CHANGE_VERSION antiguos y nuevos.
2. Cargue los datos diferenciales combinando las claves principales de las filas modificadas (entre dos valores SYS_CHANGE_VERSION) desde sys.change_tracking_tables con los datos de la tabla de origen y, a continuación, muévalos al destino.
3. Actualice el valor de SYS_CHANGE_VERSION para la carga diferencial la próxima vez.

Solución de alto nivel

En este tutorial, creará dos canalizaciones que llevan a cabo las dos operaciones siguientes:

Initial load: se crea una canalización con una actividad de copia que copia todos los datos del almacén de datos de origen (Azure SQL Database) en el almacén de datos de destino (Azure Blob Storage).
Carga incremental: creas una canalización con las siguientes actividades, y la ejecutas con regularidad.
1. Cree dos actividades de búsqueda para obtener las SYS_CHANGE_VERSION antigua y nueva de Azure SQL Database y pasarlas a la actividad de copia.
2. Cree una actividad de copia para copiar los datos insertados, actualizados o eliminados entre los dos valores de SYS_CHANGE_VERSION desde Azure SQL Database a Azure Blob Storage.
3. Cree una actividad de procedimiento almacenado para actualizar el valor SYS_CHANGE_VERSION para la ejecución de la siguiente canalización.

Si no tiene una suscripción de Azure, cree una cuenta free antes de comenzar.

Requisitos previos

Azure PowerShell. Instale los módulos de Azure PowerShell más recientes siguiendo las instrucciones de Cómo instalar y configurar Azure PowerShell.
Azure SQL Database. La base de datos se usa como almacén de datos de origen. Si no tiene una base de datos en Azure SQL Database, consulte el artículo Crear una base de datos en Azure SQL Database para ver los pasos para crear una.
cuenta de almacenamiento de Azure. Blob Storage se usa como almacén de datos receptor. Si no tiene una cuenta de almacenamiento de Azure, consulte el artículo Crear una cuenta de almacenamiento para ver los pasos para crear una. Cree un contenedor denominado adftutorial.

Creación de una tabla de origen de datos en la base de datos

Inicie SQL Server Management Studio y conéctese a SQL Database.
En el Explorador de servidores, haga clic con el botón derecho en la base de datos y elija la Nueva consulta.

Ejecute el siguiente comando SQL en su base de datos para crear una tabla denominada data_source_table como almacén de origen de datos.

create table data_source_table
(
    PersonID int NOT NULL,
    Name varchar(255),
    Age int
    PRIMARY KEY (PersonID)
);

INSERT INTO data_source_table
    (PersonID, Name, Age)
VALUES
    (1, 'aaaa', 21),
    (2, 'bbbb', 24),
    (3, 'cccc', 20),
    (4, 'dddd', 26),
    (5, 'eeee', 22);

Habilite el mecanismo de control de cambios en la base de datos y la tabla de origen (data_source_table) ejecutando la siguiente consulta SQL:
Nota:
- Reemplaza <el nombre de tu base de datos> por el nombre de la base de datos que tiene la tabla data_source_table.
- Los datos modificados se mantienen durante dos días en el ejemplo actual. Si carga los datos cambiados cada tres días o con mayor frecuencia, no se incluirán algunos datos cambiados. Tiene que cambiar el valor de CHANGE_RETENTION por un número mayor. También puede asegurarse de que el período para cargar los datos cambiados es dentro de dos días. Para más información, vea Habilitar el control de cambios para una base de datos
```
ALTER DATABASE <your database name>
SET CHANGE_TRACKING = ON  
(CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON)  

ALTER TABLE data_source_table
ENABLE CHANGE_TRACKING  
WITH (TRACK_COLUMNS_UPDATED = ON)
```

Cree una tabla y almacene el valor predeterminado ChangeTracking_version ejecutando la consulta siguiente:

create table table_store_ChangeTracking_version
(
    TableName varchar(255),
    SYS_CHANGE_VERSION BIGINT,
);

DECLARE @ChangeTracking_version BIGINT
SET @ChangeTracking_version = CHANGE_TRACKING_CURRENT_VERSION();  

INSERT INTO table_store_ChangeTracking_version
VALUES ('data_source_table', @ChangeTracking_version)

Nota:

Si los datos no cambian después de haber habilitado el control de cambios para SQL Database, el valor de la versión de control de cambios es 0.

Ejecute la siguiente consulta para crear un procedimiento almacenado en su base de datos. La canalización invoca este procedimiento almacenado para actualizar la versión de control de cambios en la tabla que creó en el paso anterior.

CREATE PROCEDURE Update_ChangeTracking_Version @CurrentTrackingVersion BIGINT, @TableName varchar(50)
AS

BEGIN

UPDATE table_store_ChangeTracking_version
SET [SYS_CHANGE_VERSION] = @CurrentTrackingVersion
WHERE [TableName] = @TableName

END

Azure PowerShell

Instale los módulos de Azure PowerShell más recientes siguiendo las instrucciones de Cómo instalar y configurar Azure PowerShell.

Crear una factoría de datos

Defina una variable para el nombre del grupo de recursos que usa en los comandos de PowerShell más adelante. Copie el siguiente texto de comando en PowerShell, especifique un nombre para el grupo de recursos Azure entre comillas dobles y, a continuación, ejecute el comando . Por ejemplo: "adfrg".
```
$resourceGroupName = "ADFTutorialResourceGroup";
```
Si el grupo de recursos ya existe, puede que no desee sobrescribirlo. Asigne otro valor a la variable $resourceGroupName y vuelva a ejecutar el comando
Defina una variable para la ubicación de la factoría de datos:
```
$location = "East US"
```
Para crear el grupo de recursos Azure, ejecute el siguiente comando:
```
New-AzResourceGroup $resourceGroupName $location
```
Si el grupo de recursos ya existe, puede que no desee sobrescribirlo. Asigne otro valor a la variable $resourceGroupName y ejecute el comando de nuevo.
Defina una variable para el nombre de la factoría de datos.

Importante

Actualice el nombre de la factoría de datos para que sea globalmente único.
```
$dataFactoryName = "IncCopyChgTrackingDF";
```

Para crear la factoría de datos, ejecute el siguiente cmdlet, Set-AzDataFactoryV2:

Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location $location -Name $dataFactoryName

Tenga en cuenta los siguientes puntos:

El nombre de la factoría de datos de Azure debe ser único globalmente. Si recibe el siguiente error, cambie el nombre y vuelva a intentarlo.
```
The specified Data Factory name 'ADFIncCopyChangeTrackingTestFactory' is already in use. Data Factory names must be globally unique.
```
Para crear instancias de Data Factory, la cuenta de usuario que usa para iniciar sesión en Azure debe ser miembro de contributor o owner o un administrator de la suscripción de Azure.
Para obtener una lista de las regiones de Azure en las que Data Factory está disponible actualmente, seleccione las regiones que le interesan en la página siguiente y, a continuación, expanda Analytics para buscar Data Factory: Productos disponibles por región. Los almacenes de datos (Azure Storage, Azure SQL Database, etc.) y los procesos (HDInsight, etc.) usados por la factoría de datos pueden estar en otras regiones.

Crear servicios vinculados

Los servicios vinculados se crean en una factoría de datos para vincular los almacenes de datos y los servicios de proceso con la factoría de datos. En esta sección, creará servicios vinculados a la cuenta de Azure Storage y a la base de datos en Azure SQL Database.

Cree un servicio vinculado de Azure Storage.

En este paso, vinculará la cuenta de Azure Storage a la factoría de datos.

Cree un archivo JSON llamado AzureStorageLinkedService.json en la carpeta C:\ADFTutorials\IncCopyChangeTrackingTutorial con el siguiente contenido: (cree la carpeta si aún no existe). Reemplace <accountName>, <accountKey> por el nombre y la clave de la cuenta de almacenamiento de Azure antes de guardar el archivo.
```
{
    "name": "AzureStorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountName>;AccountKey=<accountKey>"
        }
    }
}
```
En Azure PowerShell, cambie a la carpeta C:\ADFTutorials\IncCopyChangeTrackingTutorial.

Ejecute el cmdlet Set-AzDataFactoryV2LinkedService para crear el servicio vinculado: AzureStorageLinkedService. En el ejemplo siguiente, debe pasar los valores de los parámetros ResourceGroupName y DataFactoryName.

Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureStorageLinkedService" -File ".\AzureStorageLinkedService.json"

Este es la salida de ejemplo:

LinkedServiceName : AzureStorageLinkedService
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureStorageLinkedService

Crear un servicio vinculado de Azure SQL Database.

En este paso, vinculará la base de datos a la factoría de datos.

Cree un archivo JSON denominado AzureSQLDatabaseLinkedService.json en la carpeta C:\ADFTutorials\IncCopyChangeTrackingTutorial con el siguiente contenido: Reemplace <your-server-name> y <your-database-name> por el nombre del servidor y la base de datos antes de guardar el archivo. También debe configurar Azure SQL Server para otorgar el acceso a la identidad administrada de la fábrica de datos.
```
{
"name": "AzureSqlDatabaseLinkedService",
"properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "connectionString": "Server=tcp:<your-server-name>.database.windows.net,1433;Database=<your-database-name>;"
        },
        "authenticationType": "ManagedIdentity",
        "annotations": []
    }
}
```

En Azure PowerShell, ejecute el cmdlet Set-AzDataFactoryV2LinkedService para crear el servicio vinculado: AzureSQLDatabaseLinkedService.

Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSQLDatabaseLinkedService" -File ".\AzureSQLDatabaseLinkedService.json"

Este es la salida de ejemplo:

LinkedServiceName : AzureSQLDatabaseLinkedService
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDatabaseLinkedService

Creación de conjuntos de datos

En este paso, creará conjuntos de datos para representar el origen de datos, el destino de datos. y el lugar donde almacenar el valor SYS_CHANGE_VERSION.

Creación de un conjunto de datos de origen

En este paso, creará conjuntos de datos para representar el origen de datos.

Cree un archivo JSON llamado SourceDataset.json en la misma carpeta con el siguiente contenido:

{
    "name": "SourceDataset",
    "properties": {
        "type": "AzureSqlTable",
        "typeProperties": {
            "tableName": "data_source_table"
        },
        "linkedServiceName": {
            "referenceName": "AzureSQLDatabaseLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

Ejecute el cmdlet Set-AzDataFactoryV2Dataset para crear el conjunto de datos: SourceDataset

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SourceDataset" -File ".\SourceDataset.json"

Esta es la salida de ejemplo del cmdlet:

DatasetName       : SourceDataset
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

Creación de un conjunto de datos receptor

En este paso, creará un conjunto de datos para representar los datos que se copian desde el almacén de datos de origen.

Cree un archivo JSON llamado SinkDataset.json en la misma carpeta con el siguiente contenido:

{
    "name": "SinkDataset",
    "properties": {
        "type": "AzureBlob",
        "typeProperties": {
            "folderPath": "adftutorial/incchgtracking",
            "fileName": "@CONCAT('Incremental-', pipeline().RunId, '.txt')",
            "format": {
                "type": "TextFormat"
            }
        },
        "linkedServiceName": {
            "referenceName": "AzureStorageLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

Creas el contenedor adftutorial en Azure Blob Storage como parte de los prerrequisitos. Cree el contenedor si no existe (o) asígnele el nombre de uno existente. En este tutorial, el nombre del archivo de salida se genera dinámicamente con la expresión:@CONCAT('Incremental-', pipeline().RunId, '.txt').

Ejecute el cmdlet Set-AzDataFactoryV2Dataset para crear el conjunto de datos: SinkDataset

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SinkDataset" -File ".\SinkDataset.json"

Esta es la salida de ejemplo del cmdlet:

DatasetName       : SinkDataset
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureBlobDataset

Creación de un conjunto de datos de control de cambios

En este paso, creará un conjunto de datos para almacenar la versión de control de cambios.

Cree un archivo JSON llamado ChangeTrackingDataset.json en la misma carpeta con el siguiente contenido:

{
    "name": " ChangeTrackingDataset",
    "properties": {
        "type": "AzureSqlTable",
        "typeProperties": {
            "tableName": "table_store_ChangeTracking_version"
        },
        "linkedServiceName": {
            "referenceName": "AzureSQLDatabaseLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

Crea la tabla table_store_ChangeTracking_version como parte de los requisitos previos.

Ejecute el cmdlet Set-AzDataFactoryV2Dataset para crear el conjunto de datos: ChangeTrackingDataset

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "ChangeTrackingDataset" -File ".\ChangeTrackingDataset.json"

Esta es la salida de ejemplo del cmdlet:

DatasetName       : ChangeTrackingDataset
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

Cree una canalización para la copia completa

En este paso, creará una canalización con una actividad de copia que copia todos los datos del almacén de datos de origen (Azure SQL Database) en el almacén de datos de destino (Azure Blob Storage).

Cree un archivo JSON llamado FullCopyPipeline.json en la misma carpeta con el siguiente contenido:

{
    "name": "FullCopyPipeline",
    "properties": {
        "activities": [{
            "name": "FullCopyActivity",
            "type": "Copy",
            "typeProperties": {
                "source": {
                    "type": "SqlSource"
                },
                "sink": {
                    "type": "BlobSink"
                }
            },

            "inputs": [{
                "referenceName": "SourceDataset",
                "type": "DatasetReference"
            }],
            "outputs": [{
                "referenceName": "SinkDataset",
                "type": "DatasetReference"
            }]
        }]
    }
}

Ejecute el cmdlet Set-AzDataFactoryV2Pipeline para crear la canalización: FullCopyPipeline.

 Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "FullCopyPipeline" -File ".\FullCopyPipeline.json"

Este es la salida de ejemplo:

 PipelineName      : FullCopyPipeline
 ResourceGroupName : ADFTutorialResourceGroup
 DataFactoryName   : IncCopyChgTrackingDF
 Activities        : {FullCopyActivity}
 Parameters        :

Ejecución de la canalización de copia completa

Ejecución de la canalización FullCopyPipeline mediante el cmdlet Invoke-AzDataFactoryV2Pipeline.

Invoke-AzDataFactoryV2Pipeline -PipelineName "FullCopyPipeline" -ResourceGroup $resourceGroupName -dataFactoryName $dataFactoryName

Supervisión de la canalización de copia completa

Inicie sesión en Azure portal.
Haga clic en Todos los servicios, busque con la palabra clave data factories y seleccione Factorías de datos.
Busque su factoría de datos en la lista y selecciónela para iniciar la página Factoría de datos.
En la página Factoría de datos, haga clic en el icono Supervisión y administración.
La aplicación de integración de datos se inicia en otra pestaña. Puede ver todas las ejecuciones de canalización y sus estados. Tenga en cuenta que, en el ejemplo siguiente, el estado de ejecución de la canalización es Completado con éxito. Puede comprobar los parámetros pasados a la canalización si hace clic en la columna Parámetros. Si se ha producido un error, verá un vínculo en la columna Error. Haga clic en el vínculo de la columna Acciones.
Al hacer clic en el vínculo de la columna Acciones, verá la página siguiente que muestra todas las ejecuciones de actividad de la canalización.
Para volver a la vista de ejecuciones de canalización, haga clic en Canalizaciones como se muestra en la imagen.

Revisión del resultado

Verá un archivo denominado incremental-<GUID>.txt en la carpeta incchgtracking del contenedor adftutorial.

Archivo de salida de una copia completa

El archivo debe tener los datos de la base de datos:

1,aaaa,21
2,bbbb,24
3,cccc,20
4,dddd,26
5,eeee,22

Adición de más datos a la tabla de origen

Ejecute la siguiente consulta en la base de datos para agregar una fila y actualizarla.

INSERT INTO data_source_table
(PersonID, Name, Age)
VALUES
(6, 'new','50');


UPDATE data_source_table
SET [Age] = '10', [name]='update' where [PersonID] = 1

Creación de una canalización para la copia delta

En este paso, creará una canalización con las siguientes actividades y la ejecutará de forma periódica. Las actividades lookup obtienen la SYS_CHANGE_VERSION antigua y nueva desde Azure SQL Database y se la pasan a la actividad de copia. La actividad copy copia los datos insertados, actualizados o eliminados entre los dos valores de SYS_CHANGE_VERSION de Azure SQL Database a Azure Blob Storage. La actividad de procedimiento almacenado actualiza el valor SYS_CHANGE_VERSION para la ejecución de la siguiente canalización.

Cree un archivo JSON llamado IncrementalCopyPipeline.json en la misma carpeta con el siguiente contenido:

{
    "name": "IncrementalCopyPipeline",
    "properties": {
        "activities": [
            {
                "name": "LookupLastChangeTrackingVersionActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select * from table_store_ChangeTracking_version"
                    },
                    "dataset": {
                        "referenceName": "ChangeTrackingDataset",
                        "type": "DatasetReference"
                    }
                }
            },
            {
                "name": "LookupCurrentChangeTrackingVersionActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion"
                    },
                    "dataset": {
                        "referenceName": "SourceDataset",
                        "type": "DatasetReference"
                    }
                }
            },
            {
                "name": "IncrementalCopyActivity",
                "type": "Copy",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select data_source_table.PersonID,data_source_table.Name,data_source_table.Age, CT.SYS_CHANGE_VERSION, SYS_CHANGE_OPERATION from data_source_table RIGHT OUTER JOIN CHANGETABLE(CHANGES data_source_table, @{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.SYS_CHANGE_VERSION}) as CT on data_source_table.PersonID = CT.PersonID where CT.SYS_CHANGE_VERSION <= @{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}"
                    },
                    "sink": {
                        "type": "BlobSink"
                    }
                },
                "dependsOn": [
                    {
                        "activity": "LookupLastChangeTrackingVersionActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    },
                    {
                        "activity": "LookupCurrentChangeTrackingVersionActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "inputs": [
                    {
                        "referenceName": "SourceDataset",
                        "type": "DatasetReference"
                    }
                ],
                "outputs": [
                    {
                        "referenceName": "SinkDataset",
                        "type": "DatasetReference"
                    }
                ]
            },
            {
                "name": "StoredProceduretoUpdateChangeTrackingActivity",
                "type": "SqlServerStoredProcedure",
                "typeProperties": {
                    "storedProcedureName": "Update_ChangeTracking_Version",
                    "storedProcedureParameters": {
                        "CurrentTrackingVersion": {
                            "value": "@{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}",
                            "type": "INT64"
                        },
                        "TableName": {
                            "value": "@{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.TableName}",
                            "type": "String"
                        }
                    }
                },
                "linkedServiceName": {
                    "referenceName": "AzureSQLDatabaseLinkedService",
                    "type": "LinkedServiceReference"
                },
                "dependsOn": [
                    {
                        "activity": "IncrementalCopyActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ]
            }
        ]
    }
}

Ejecute el cmdlet Set-AzDataFactoryV2Pipeline para crear la canalización: FullCopyPipeline.

 Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "IncrementalCopyPipeline" -File ".\IncrementalCopyPipeline.json"

Este es la salida de ejemplo:

 PipelineName      : IncrementalCopyPipeline
 ResourceGroupName : ADFTutorialResourceGroup
 DataFactoryName   : IncCopyChgTrackingDF
 Activities        : {LookupLastChangeTrackingVersionActivity, LookupCurrentChangeTrackingVersionActivity, IncrementalCopyActivity, StoredProceduretoUpdateChangeTrackingActivity}
 Parameters        :

Ejecución de la canalización de la copia incremental

Ejecución de la canalización IncrementalCopyPipeline mediante el cmdlet Invoke-AzDataFactoryV2Pipeline.

Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroup $resourceGroupName -dataFactoryName $dataFactoryName

Monitorear el flujo de copia incremental

En la aplicación de integración de datos, actualice la vista de ejecuciones de canalización. Confirme que ve IncrementalCopyPipeline en la lista. Haga clic en el vínculo de la columna Acciones.
Al hacer clic en el vínculo de la columna Acciones, verá la página siguiente que muestra todas las ejecuciones de actividad de la canalización.
Para volver a la vista de ejecuciones de canalización, haga clic en Canalizaciones como se muestra en la imagen.

Revisión del resultado

Verá el segundo archivo incchgtracking en la carpeta adftutorial del contenedor.

Archivo de salida de la copia incremental

El archivo debe tener solo los datos diferenciales de la base de datos. El registro con U es la fila actualizada en la base de datos y I es la fila que se agrega.

1,update,10,2,U
6,new,50,1,I

Las tres primeras columnas son datos cambiados de data_source_table. Las dos últimas columnas son los metadatos de la tabla del sistema de control de cambios. La cuarta columna es el valor SYS_CHANGE_VERSION de cada fila modificada. La quinta columna es la operación: U = actualización, I = inserción. Para obtener más información acerca de la información de control de cambios, consulte CHANGETABLE.

==================================================================
PersonID Name    Age    SYS_CHANGE_VERSION    SYS_CHANGE_OPERATION
==================================================================
1        update  10            2                                 U
6        new     50            1                                 I

Pase al tutorial siguiente para obtener información acerca de cómo copiar archivos nuevos y modificados solo según el valor de LastModifiedDate:

Copia de archivos nuevos por LastModifiedDate

Comentarios

¿Le resultó útil esta página?

Last updated on 2026-04-07

Compartir vía

Carga incremental de datos de Azure SQL Database a Azure Blob Storage mediante la información de seguimiento de cambios mediante PowerShell

Información general

Flujo de trabajo de un extremo a otro

Solución de alto nivel

Requisitos previos

Creación de una tabla de origen de datos en la base de datos

Azure PowerShell

Crear una factoría de datos

Crear servicios vinculados

Cree un servicio vinculado de Azure Storage.

Crear un servicio vinculado de Azure SQL Database.

Creación de conjuntos de datos

Creación de un conjunto de datos de origen

Creación de un conjunto de datos receptor

Creación de un conjunto de datos de control de cambios

Cree una canalización para la copia completa

Ejecución de la canalización de copia completa

Supervisión de la canalización de copia completa

Revisión del resultado

Adición de más datos a la tabla de origen

Creación de una canalización para la copia delta

Ejecución de la canalización de la copia incremental

Monitorear el flujo de copia incremental

Revisión del resultado

Contenido relacionado

Comentarios

Recursos adicionales