مشاركة عبر

تحميل البيانات بشكل متزايد من Azure SQL Database إلى تخزين Azure Blob باستخدام معلومات تعقب التغييرات باستخدام PowerShell

  • مقالة

ينطبق على: Azure Data Factory Azure Synapse Analytics

تلميح

جرب Data Factory في Microsoft Fabric، وهو حل تحليلي متكامل للمؤسسات. يغطي Microsoft Fabric كل شيء بدءا من حركة البيانات إلى علم البيانات والتحليلات في الوقت الحقيقي والمعلومات المهنية وإعداد التقارير. تعرف على كيفية بدء إصدار تجريبي جديد مجانا!

في هذا البرنامج التعليمي، يمكنك إنشاء مصنع بيانات Azure مع خط أنابيب يقوم بتحميل بيانات دلتا استناداً إلى معلومات تعقب التغيير في قاعدة البيانات المصدر في Azure SQL Database إلى تخزين النقطة Azure.

نفذ الخطوات التالية في هذا البرنامج التعليمي:

  • قم بإعداد مخزن البيانات المصدر
  • إنشاء data factory.
  • أنشئ الخدمة ذات الصلة.
  • إنشاء مجموعات بيانات التعقب المصدر والغرق وتغييرها.
  • إنشاء "pipeline" للنسخ الكامل وتشغيله ومراقبة
  • إضافة بيانات أو تحديثها في الجدول المصدر
  • إنشاء تشغيل ومراقبة "pipeline" للنسخ التزايدي

إشعار

نوصي باستخدام الوحدة النمطية Azure Az PowerShell للتفاعل مع Azure. للبدء، راجع تثبيت Azure PowerShell. لمعرفة كيفية الترحيل إلى الوحدة النمطية Az PowerShell، راجع ترحيل Azure PowerShell من AzureRM إلى Az.

نظرة عامة

في حل تكامل البيانات، تحميل البيانات بشكل متزايد بعد تحميل البيانات الأولية سيناريو مستخدم على نطاق واسع. في بعض الحالات، يمكن تقسيم البيانات التي تم تغييرها خلال فترة في مخزن البيانات المصدر بسهولة إلى شرائح (على سبيل المثال، LastModifyTime وCreationTime). في بعض الحالات، لا توجد طريقة صريحة لتعريف بيانات دلتا من آخر مرة قمت بمعالجة البيانات. يمكن استخدام تقنية تعقب التغيير المعتمدة من قبل مخازن البيانات مثل Azure SQL Database وSQL Server لتحديد بيانات دلتا. يصف هذا البرنامج التعليمي كيفية استخدام Azure Data Factory مع تقنية تعقب التغيير SQL لتحميل بيانات دلتا بشكل متزايد من قاعدة بيانات Azure SQL إلى تخزين Azure Blob. لمزيد من المعلومات الملموسة حول تقنية تعقب التغييرات SQL، راجع تعقب التغيير في SQL Server.

سير العمل من البداية إلى النهاية

فيما يلي خطوات سير العمل النموذجية من البداية إلى النهاية لتحميل البيانات بشكل متزايد باستخدام تقنية "تعقب التغيير".

إشعار

تدعم كل من قاعدة بيانات SQL Azure SQL Server تقنية تعقب التغيير. يستخدم هذا البرنامج التعليمي قاعدة بيانات azure SQL كمخزن بيانات المصدر. يمكنك أيضاً استخدام مثيل SQL Server.

  1. التحميل الأولي للبيانات التاريخية (تشغيل مرة واحدة):
    1. تمكين تقنية تعقب التغيير في قاعدة البيانات المصدر في Azure SQL Database.
    2. الحصول على القيمة الأولية SYS_CHANGE_VERSION في قاعدة البيانات كأساس لالتقاط البيانات التي تم تغييرها.
    3. تحميل البيانات الكاملة من قاعدة البيانات المصدر في تخزين Azure blob.
  2. التحميل التزايدي لبيانات دلتا على جدول زمني (يتم تشغيله بشكل دوري بعد التحميل الأولي للبيانات):
    1. احصل على قيم SYS_CHANGE_VERSION القديمة والجديدة.
    2. قم بتحميل بيانات دلتا عن طريق ربط المفاتيح الأساسية للصفوف المتغيرة (بين قيمتين SYS_CHANGE_VERSION) من sys.change_tracking_tables مع البيانات في الجدول المصدر، ثم قم بنقل بيانات "دلتا" إلى الوجهة.
    3. تحديث SYS_CHANGE_VERSION لتحميل "دلتا" في المرة القادمة.

حل رفيع المستوى

في هذا البرنامج التعليمي، يمكنك إنشاء خطي أنابيب تنفيذ العمليات التالية اثنين:

  1. التحميل الأولي: إنشاء "pipeline" مع نشاط نسخ البيانات بالكامل من مخزن البيانات المصدر (Azure SQL قاعدة البيانات) إلى مخزن بيانات الوجهة (تخزين Azure Blob).

    التحميل الكامل للبيانات

  2. تحميل تزايدي: إنشاء خط أنابيب مع الأنشطة التالية وتشغيله بشكل دوري.

    1. إنشاء نشاطين للبحث للحصول على SYS_CHANGE_VERSION القديمة والجديدة من Azure SQL Database وتمريرها لنسخ النشاط.
    2. إنشاء نشاط نسخة واحدة لنسخ البيانات المدرجة/المحدثة/المحذوفة بين قيمتي SYS_CHANGE_VERSION من Azure SQL Database إلى تخزين Azure Blob.
    3. إنشاء نشاط إجراء مخزن لتحديث قيمة SYS_CHANGE_VERSION لتشغيل خط الأنابيب التالي.

    رسم تخطيطي لتدفق الحمل بالزيادة

في حال لم يكن لديك اشتراك Azure، فأنشئ حساباً مجانيّاً قبل البدء.

المتطلبات الأساسية

أنشئ جدول مصدر بيانات في قاعدة بياناتك

  1. قم بتشغيل SQL Server Management Studio، والاتصال بقاعدة بيانات SQL.

  2. في Server Explorer، انقر بزر الماوس الأيمن فوق database وحدد New Query.

  3. شغل أمر SQL التالي مقابل قاعدة بيانات Azure لإنشاء جدول باسم data_source_tableمخزن مصدر البيانات.

    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);

  4. تمكين آلية تعقب التغيير على قاعدة البيانات والجدول المصدر (data_source_table) عن طريق تشغيل الاستعلام SQL التالي:

    إشعار

    • استبدال اسم قاعدة البيانات <باسم قاعدة البيانات> التي تحتوي على data_source_table.
    • يتم الاحتفاظ البيانات المتغيرة لمدة يومين في المثال الحالي. إذا قمت بتحميل البيانات التي تم تغييرها كل ثلاثة أيام أو أكثر، فلن يتم تضمين بعض البيانات التي تم تغييرها. تحتاج إلى تغيير قيمة CHANGE_RETENTION إلى رقم أكبر. وبدلاً من ذلك، تأكد من أن الفترة الزمنية لتحميل البيانات التي تم تغييرها في غضون يومين. للمزيد من المعلومات، راجع تمكين تعقب التغييرات لقاعدة بيانات
    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)

  5. إنشاء جدول جديد وتخزين ChangeTracking_version بقيمة افتراضية عن طريق تشغيل الاستعلام التالي:

    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)

    إشعار

    إذا لم يتم تغيير البيانات بعد تمكين تعقب التغيير SQL قاعدة البيانات، قيمة إصدار تعقب التغيير هو 0.

  6. تشغيل الاستعلام التالي لإنشاء إجراء مخزن في قاعدة البيانات الخاصة بك. استدعاء "pipeline" هذا الإجراء المخزن لتحديث إصدار تعقب التغيير في الجدول الذي قمت بإنشائه في الخطوة السابقة.

    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

ثبّت أحدث وحدات نمطية Azure PowerShell باتباع الإرشادات الواردة في كيفية تثبيتAzure PowerShell وتكوينه.

إنشاء مصدرًا للبيانات

  1. حدد متغيراً لاسم مجموعة الموارد الذي ستستخدمه لاحقاً في أوامر PowerShell. انسخ الأمر النصي التالي إلى PowerShell، حدد اسماً لمجموعة موارد Azure وأحطها بعلامات اقتباس مزدوجة، ثم شغل الأمر. على سبيل المثال: "adfrg".

    $resourceGroupName = "ADFTutorialResourceGroup";

    إذا كانت مجموعة الموارد موجودة بالفعل، فقد لا ترغب في الكتابة فوقها. عين قيمة مختلفة $resourceGroupName للمتغير وشغل الأمر مرة أخرى

  2. تحديد متغير لموقع مصنع البيانات:

    $location = "East US"

  3. لإنشاء مجموعة موارد Azure، شغل الأمر التالي:

    New-AzResourceGroup $resourceGroupName $location

    إذا كانت مجموعة الموارد موجودة بالفعل، فقد لا ترغب في الكتابة فوقها. تعيين قيمة مختلفة $resourceGroupName للمتغير وتشغيل الأمر مرة أخرى.

  4. حدد متغير لموقع بيانات المصنع.

    هام

    قم بتحديث اسم مصنع البيانات ليكون فريداً عالمياً.

    $dataFactoryName = "IncCopyChgTrackingDF";

  5. لإنشاء بيانات المصنع شغل التالي Set-AzDataFactoryV2 cmdlet:

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

لاحظ النقاط التالية:

  • يجب أن يكون اسم Azure Data Factory فريداً بصفة عمومية. في حال تلقيت الخطأ التالي، تغيير الاسم، ثم حاول مرة أخرى.

    The specified Data Factory name 'ADFIncCopyChangeTrackingTestFactory' is already in use. Data Factory names must be globally unique.

  • لإنشاء مثيلات Data Factory، يجب أن يكون حساب المستخدم الذي تستخدمه لتسجيل الدخول إلى Azure عضواً في أدوار المساهم أو المالك ، أو مسؤول في اشتراك Azure.

  • للحصول على قائمة بمناطق Azure التي يتوفر فيها حالياً Data Factory، حدد المناطق التي تهمك في الصفحة التالية، ثم قم بتوسيع "Analytics" لتحديد موقع Data Factory: "Products available by region". تخزن البيانات (Azure Storage، وAzure SQL Database، وما إلى ذلك) وتحسب (HDInsight، وما إلى ذلك) التي يستخدمها مصنع البيانات في مناطق أخرى.

إنشاء linked services

إنشاء خدمات مرتبطة في مصنع بيانات لربط مخازن بياناتك وحساب الخدمات إلى مصنع البيانات. في هذا القسم، يمكنك إنشاء خدمات مرتبطة بحساب تخزين Azure وقاعدة البيانات الخاصة بك في Azure SQL Database.

إنشاء خدمة مرتبطة لتخزين Azure.

يمكنك إنشاء خدمة مرتبطة لربط حساب Azure Storage بمصنع البيانات في هذه الخطوة.

  1. إنشاء ملف JSON المسمى AzureStorageLinkedService.js في C:\ADFTutorials\IncCopyChangeTrackingTutorial المجلد مع المحتوى التالي: (إنشاء المجلد إذا كان غير موجود بالفعل.). استبدل <accountName>، <accountKey> باسم ومفتاح حساب تخزين Azure قبل حفظ الملف.

    {
    "name": "AzureStorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountName>;AccountKey=<accountKey>"
        }
    }
}

  2. في Azure PowerShell، قم بالتبديل إلى مجلد C:\ADFTutorials\IncCopyChangeTrackingTutorial.

  3. شغل cmdlet Set-AzDataFactoryV2LinkedService لإنشاء الخدمة المرتبطة: AzureStorageLinkedService. في المثال التالي، يمكنك تمرير قيم ResourceGroupName واسم معلمات DataFactory.

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

    فيما يلي ناتج العينة:

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

إنشاء خدمة مرتبطة بقاعدة بيانات Azure SQL.

في هذه الخطوة، يمكنك ربط قاعدة البيانات بمصنع البيانات.

  1. أنشئ ملف JSON باسم AzureSQLDatabaseLinkedService.json في المجلد C:\ADFTutorials\IncCopyChangeTrackingTutorial بالمحتوى التالي: استبدال <your-server-name> و <your-database-name> باسم الخادم وقاعدة البيانات قبل حفظ الملف. يجب عليك أيضا تكوين Azure SQL Server لمنح حق الوصول إلى الهوية المدارة لمصنع البيانات.

    {
"name": "AzureSqlDatabaseLinkedService",
"properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "connectionString": "Server=tcp:<your-server-name>.database.windows.net,1433;Database=<your-database-name>;"
        },
        "authenticationType": "ManagedIdentity",
        "annotations": []
    }
}

  2. في Azure PowerShell، شغل Set-AzDataFactoryV2LinkedService cmdlet لإنشاء الخدمة المرتبطة: AzureSQLDatabaseLinkedService.

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

    فيما يلي ناتج العينة:

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

إنشاء datasets

في هذه الخطوة، يمكنك إنشاء مجموعات البيانات لتمثيل مصدر البيانات ووجهة البيانات. والمكان لتخزين SYS_CHANGE_VERSION.

قم بإنشاء مجموعة بيانات المصدر

في هذه الخطوة، يمكنك إنشاء مجموعات البيانات لتمثيل مصدر البيانات ووجهة البيانات.

  1. أنشئ ملف JSON باسم SourceDataset.json في نفس الملف بالمحتويات التالية:

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

  2. شغل Set-AzDataFactoryV2Dataset cmdlet لإنشاء مجموعة البيانات: SourceDataset

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

    إليك ناتج تشغيل عينة cmdlet:

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

أنشئ مجموعة بيانات المورد

في هذه الخطوة، يمكنك إنشاء مجموعة بيانات لتمثيل البيانات التي يتم نسخها من مخزن البيانات المصدر.

  1. أنشئ ملف JSON باسم SinkDataset.json في نفس الملف بالمحتويات التالية:

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

    إنشاء حاوية adftutorial في تخزين النقطة Azure كجزء من المتطلبات الأساسية. قم بإنشاء الحاوية إذا لم تكن موجودة، أو استخدم اسم حاوية موجودة. في هذا البرنامج التعليمي، يتم إنشاء اسم ملف الإخراج بشكل حيوي باستخدام التعبير: @CONCAT( ('تزايدي-'، خط أنابيب(). RunId، '.txt').

  2. شغل Set-AzDataFactoryV2Dataset cmdlet لإنشاء مجموعة البيانات: SinkDataset

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

    إليك ناتج تشغيل عينة cmdlet:

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

إنشاء مجموعة بيانات لتعقب التغييرات

في هذه الخطوة، يمكنك إنشاء مجموعة بيانات لتخزين إصدار تعقب التغيير.

  1. أنشئ ملف JSON باسم SourceDataset.json في نفس الجلد بالمحتويات التالية:

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

    إنشاء الجدول table_store_ChangeTracking_version كجزء من المتطلبات الأساسية.

  2. شغل Set-AzDataFactoryV2Dataset cmdlet لإنشاء مجموعة بيانات WatermarkDataset

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

    إليك ناتج تشغيل عينة cmdlet:

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

إنشاء "pipeline" للنسخة الكاملة

في هذه الخطوة، يمكنك إنشاء خط أنابيب مع نشاط نسخ البيانات بأكملها من مخزن البيانات المصدر (Azure SQL Database) إلى مخزن البيانات الوجهة (تخزين Azure Blob).

  1. أنشئ ملف JSON: FullCopyPipeline.json في نفس الملف بالمحتويات التالية:

    {
    "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"
            }]
        }]
    }
}

  2. شغل Set-AzDataFactoryV2Pipeline cmdlet لإنشاء البنية الأساسية IncrementalCopyPipeline.

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

    فيما يلي ناتج العينة:

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

تشغيل pipeline النسخ الكامل

تشغيل خط الأنابيب: FullCopyPipeline باستخدام Invoke-AzDataFactoryV2Pipeline cmdlet.

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

مراقبة خط أنابيب النسخ الكامل

  1. تسجيل الدخول إلى مدخل Microsoft Azure.

  2. انقر فوق جميع الخدمات، ابحث باستخدام الكلمة الرئيسية، وحدد data factories مصانع البيانات.

    قائمة مصانع البيانات

  3. ابحث عن مصنع البيانات في قائمة مصانع البيانات، وحدده لإطلاق صفحة مصنع البيانات.

    البحث عن مصنع البيانات

  4. في صفحة مصنع البيانات، انقر فوق مراقبة & إدارة الإطار المتجانب.

    مراقبة وإدارة التجانب

  5. يتم تشغيل تطبيق تكامل البيانات في علامة تبويب منفصلة. يمكنك أن ترى جميع خطوط الأنابيب تعمل وحالاتها. لاحظ أن في المثال التالي، يتم بنجاححالة تشغيل خط أنابيب . يمكنك التحقق من المعلمات التي تم تمريرها إلى خط الأنابيب بالنقر فوق الارتباط في العمود معلمات. إذا كان هناك خطأ، فسترى ارتباطاً في العمود خطأ. انقر فوق الارتباط في العمود الإجراءات.

    تظهر لقطة الشاشة تشغيل خط أنابيب لمصنع بيانات.

  6. عند النقر فوق الارتباط في العمود الإجراءات، تشاهد الصفحة التالية التي تعرض جميع الأنشطة التي يتم تشغيلها لـ "pipeline".

    تظهر لقطة الشاشة نشاطا يعمل لمصنع بيانات مع استدعاء رابط خطوط الأنابيب.

  7. للتبديل مرة أخرى إلى طريقة عرض تشغيل خطوط الأنابيب، انقر فوق "Pipelines" كما هو موضح في الصورة.

مراجعة النتائج

تشاهد ملف اسمه incremental-<GUID>.txt في incchgtracking مجلد adftutorial الحاوية.

ملف الإخراج من النسخة الكاملة

يجب أن يكون الملف البيانات من قاعدة البيانات الخاصة بك:

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

إضافة المزيد من البيانات إلى الجدول المصدر

تشغيل الاستعلام التالي مقابل قاعدة البيانات الخاصة بك لإضافة صف وتحديث صف.

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


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

إنشاء خط أنابيب لنسخة "دلتا"

في هذه الخطوة، يمكنك إنشاء "pipeline" مع الأنشطة التالية، وتشغيله بشكل دوري. الحصول على أنشطة البحث SYS_CHANGE_VERSION القديمة والجديدة من Azure SQL Database وتمريره لنسخ النشاط. نشاط النسخ نسخ البيانات المدرجة/المحدثة/المحذوفة بين قيم SYS_CHANGE_VERSION اثنين من Azure SQL Database إلى تخزين النقطة Azure. نشاط الإجراء المخزن بتحديث قيمة SYS_CHANGE_VERSION لتشغيل خط أنابيب التالي.

  1. أنشئ ملف JSON باسم IncrementalCopyPipeline.json في نفس الملف بالمحتويات التالية:

    {
    "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"
                        ]
                    }
                ]
            }
        ]
    }
}

  2. شغل Set-AzDataFactoryV2Pipeline cmdlet لإنشاء البنية الأساسية IncrementalCopyPipeline.

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

    فيما يلي ناتج العينة:

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

تشغيل البنية الأساسية المنسوخة والمتزايدة

تشغيل pipeline الخاص بـ IncrementalCopyPipeline باستخدام Invoke-AzDataFactoryV2Pipeline cmdlet.

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

مراقبة البنية الأساسية المنسوخة والمتزايدة

  1. في تطبيق تكامل البيانات، قم بتحديث طريقة عرض تشغيل خطوط الأنابيب. تأكد من أن تشاهد IncrementalCopyPipeline في القائمة. انقر فوق الارتباط في العمود الإجراءات.

    تظهر لقطة الشاشة تشغيل خط أنابيب لمصنع بيانات بما في ذلك خط الأنابيب الخاص بك.

  2. عند النقر فوق الارتباط في العمود الإجراءات، تشاهد الصفحة التالية التي تعرض جميع الأنشطة التي يتم تشغيلها لـ "pipeline".

    تظهر لقطة الشاشة تشغيل

  3. للتبديل مرة أخرى إلى طريقة عرض تشغيل خطوط الأنابيب، انقر فوق "Pipelines" كما هو موضح في الصورة.

مراجعة النتائج

راجع الملف الثاني في incchgtracking مجلد adftutorial الحاوية.

ملف الإخراج من نسخة تزايدية

يجب أن يكون الملف بيانات "دلتا" فقط من قاعدة البيانات الخاصة بك. السجل المزود بـ U هو الصف المحدث في قاعدة البيانات وهو الصف I المضاف.

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

يتم تغيير الأعمدة الثلاثة الأولى البيانات من data_source_table. العمودان الأخيران هما بيانات التعريف من جدول نظام تعقب التغييرات. العمود الرابع هو SYS_CHANGE_VERSION لكل صف تم تغييره. العمود الخامس هو العملية: U = update، I = insert. للحصول على تفاصيل حول معلومات تعقب التغيير، راجع CHANGETABLE.

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

المحتوى ذو الصلة

تقدم إلى البرنامج التعليمي التالي لمعرفة المزيد عن نسخ الملفات الجديدة والمتغيرة فقط استناداً إلى LastModifiedDate:

نسخ ملفات جديدة حسب lastmodifieddate