مشاركة عبر

تحميل البيانات بشكل متزايد من جداول متعددة في SQL Server إلى قاعدة بيانات في قاعدة بيانات azure SQL باستخدام مدخل Microsoft Azure

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

تلميح

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

في هذا البرنامج التعليمي، يمكنك إنشاء Azure Data Factory مع مسار يقوم بتحميل بيانات دلتا من جداول متعددة في قاعدة بيانات SQL Server إلى قاعدة بيانات في قاعدة بيانات Azure SQL.

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

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

نظرة عامة

فيما يلي الخطوات الهامة لإنشاء هذا الحل:

  1. حدد عمود العلامة المائية.

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

  2. أعد مخزن بيانات لتخزين قيمة العلامة المائية.

    في هذا البرنامج التعليمي، يمكنك تخزين قيمة العلامة المائية في قاعدة بيانات SQL.

  3. إنشاء خط أنابيب مع الأنشطة التالية:

    أ. إنشاء نشاط ForEach الذي يبتكر من خلال قائمة أسماء الجداول المصدر التي يتم تمريرها كمعلمة إلى خط الأنابيب. لكل جدول مصدر استدعاء الأنشطة التالية لتنفيذ تحميل دلتا لهذا الجدول.

    ب. إنشاء نشاطين من أنشطة البحث. استخدم نشاط البحث الأول لاسترداد قيمة العلامة المائية الأخيرة. استخدم نشاط البحث الثاني لاسترداد قيمة العلامة المائية الجديدة. يتم تمرير قيم هذه العلامات المائية إلى نشاط النسخ.

    جـ. أنشئ نشاط نسخ ينسخ صفوف من مخزن بيانات المصدر مع إدخال قيمة لعمود العلامة المائية أكبر من قيمة العلامة المائية القديمة وأقل من قيمة العلامة المائية الجديدة. بعد ذلك، ينسخ بيانات "دلتا" من مخزن بيانات المصدر إلى تخزين Azure Blob كملف جديد.

    د. أنشئ نشاط StoredProcedure الذي يحدث قيمة العلامة المائية للمسار الذي سيتم تشغيله في المرة القادمة.

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

    تحميل البيانات على نحو تدريجي

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

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

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

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

  1. افتح SQL Server Management Studio واتصل بالخادم الأول.

  2. في مستكشف العناصرانقر بزر الماوس الأيمن فوق database وحدد New Query.

  3. تشغيل الأمر SQL التالية مقابل قاعدة البيانات لإنشاء جداول customer_table المسماة project_table و:

    create table customer_table
(
    PersonID int,
    Name varchar(255),
    LastModifytime datetime
);

create table project_table
(
    Project varchar(255),
    Creationtime datetime
);

INSERT INTO customer_table
(PersonID, Name, LastModifytime)
VALUES
(1, 'John','9/1/2017 12:56:00 AM'),
(2, 'Mike','9/2/2017 5:23:00 AM'),
(3, 'Alice','9/3/2017 2:36:00 AM'),
(4, 'Andy','9/4/2017 3:21:00 AM'),
(5, 'Anny','9/5/2017 8:06:00 AM');

INSERT INTO project_table
(Project, Creationtime)
VALUES
('project1','1/1/2015 0:00:00 AM'),
('project2','2/2/2016 1:23:00 AM'),
('project3','3/4/2017 5:16:00 AM');

إنشاء جداول الوجهة في قاعدة بياناتك

  1. افتح SQL Server Management Studio، واتصل بقاعدة البيانات الخاصة بك في Azure SQL Database.

  2. في مستكشف العناصرانقر بزر الماوس الأيمن فوق database وحدد New Query.

  3. تشغيل الأمر SQL التالية مقابل قاعدة البيانات لإنشاء جداول customer_table المسماة project_table و:

    create table customer_table
(
    PersonID int,
    Name varchar(255),
    LastModifytime datetime
);

create table project_table
(
    Project varchar(255),
    Creationtime datetime
);

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

  1. شغل أمر SQL التالي مقابل قاعدة بياناتك في SQL لإنشاء جدول باسم watermarktableلتخزين قيمة العلامة المائية:

    create table watermarktable
(

    TableName varchar(255),
    WatermarkValue datetime,
);

  2. إدراج قيم العلامة المائية الأولية لكلا الجدولين المصدرين في جدول العلامة المائية.

    
INSERT INTO watermarktable
VALUES 
('customer_table','1/1/2010 12:00:00 AM'),
('project_table','1/1/2010 12:00:00 AM');

أنشئ إجراء مخزن في قاعدة بياناتك

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

CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS

BEGIN

UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime 
WHERE [TableName] = @TableName

END

إنشاء أنواع البيانات والإجراءات المخزنة الإضافية في قاعدة بياناتك

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

من أجل جعل الرحلة سهلة لتبدأ، ونحن نستخدم مباشرة هذه الإجراءات المخزنة تمرير البيانات دلتا في عبر متغير الجدول ومن ثم دمج لهم في مخزن الوجهة. كن حذرا فإنه لا تتوقع عدد "كبير" من صفوف دلتا (أكثر من 100) ليتم تخزينها في متغير الجدول.

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

CREATE TYPE DataTypeforCustomerTable AS TABLE(
    PersonID int,
    Name varchar(255),
    LastModifytime datetime
);

GO

CREATE PROCEDURE usp_upsert_customer_table @customer_table DataTypeforCustomerTable READONLY
AS

BEGIN
  MERGE customer_table AS target
  USING @customer_table AS source
  ON (target.PersonID = source.PersonID)
  WHEN MATCHED THEN
      UPDATE SET Name = source.Name,LastModifytime = source.LastModifytime
  WHEN NOT MATCHED THEN
      INSERT (PersonID, Name, LastModifytime)
      VALUES (source.PersonID, source.Name, source.LastModifytime);
END

GO

CREATE TYPE DataTypeforProjectTable AS TABLE(
    Project varchar(255),
    Creationtime datetime
);

GO

CREATE PROCEDURE usp_upsert_project_table @project_table DataTypeforProjectTable READONLY
AS

BEGIN
  MERGE project_table AS target
  USING @project_table AS source
  ON (target.Project = source.Project)
  WHEN MATCHED THEN
      UPDATE SET Creationtime = source.Creationtime
  WHEN NOT MATCHED THEN
      INSERT (Project, Creationtime)
      VALUES (source.Project, source.Creationtime);
END

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

  1. شغِّل إما مستعرض الويب Microsoft Edge أو Google Chrome. يتم حاليًّا دعم واجهة مستخدم Data Factory فقط في مستعرضي الويب Microsoft Edge وGoogle Chrome.

  2. في القائمة العلوية، حدد Create a resource>Analytics>Data Factory :

    تحديد Data Factory في الاقتباس; &اقتباس جديد; جزء

  3. في صفحة مصنع بيانات جديد، أدخل ADFMultiIncCopyTutorialDFللاسم.

    يجب أن يكون اسم Azure data factory مميزاً بصفة عمومية. إذا رأيت علامة تعجب حمراء مع الخطأ التالي تغيير اسم مصنع البيانات (على سبيل المثال، yournameADFIncCopyTutorialDF) ثم حاول إنشاء مرة أخرى. راجع المقالةData Factory - Naming Rules لمعرفة قواعد تسمية أدوات مصنع البيانات.

    Data factory name "ADFIncCopyTutorialDF" is not available

  4. حدد اشتراك Azure الذي تريد إنشاء مصنع البيانات فيه.

  5. بالنسبة Resource Group، نفِّذ إحدى الخطوات التالية:

  6. حدد "V2" لحقل"version".

  7. حدد موقع، مصِّنع البيانات. لن تظهر القائمة المنسدلة إلا على المواقع المعتمدة فقط. تخزن البيانات (Azure Storage، وAzure SQL Database، وما إلى ذلك) وتحسب (HDInsight، وما إلى ذلك) التي يستخدمها مصنع البيانات في مناطق أخرى.

  8. انقر فوق Create.

  9. بعد الانتهاء من الإنشاء، ستظهر صفحة مصنع البيانات كما هو موضح في الصورة.

    الصفحة الرئيسية لـ zure Data Factory بالإضافة إلى تجانب Open Azure Data Factory Studio

  10. حدد Open على تجانب Open Azure Data Factory Studio لبدء تشغيل تطبيق واجهة مستخدم Azure Data Factory الموجود في علامة تبويب منفصلة.

إنشاء وقت تشغيل التكامل المستضاف ذاتيًا

أثناء نقل البيانات من مخزن بيانات في شبكة اتصال خاصة (محلية) إلى مخزن بيانات Azure، قم بتثبيت وقت تشغيل تكامل ذاتي الاستضافة (IR) في البيئة الداخلية. ينقل IR المستضاف ذاتيا البيانات بين الشبكة الخاصة وAzure.

  1. في الصفحة الرئيسية لواجهة مستخدم Azure Data Factory، حدد علامة التبويب إدارة من الجزء الموجود في أقصى اليسار.

    زر

  2. حدد أوقات تشغيل التكامل على الجزء الأيسر، ثم حدد +جديد.

    إنشاء وقت تشغيل تكامل

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

  4. حدد ذاتي الاستضافة، وانقر فوق متابعة.

  5. أدخل MySelfHostedIRللاسم، وانقر فوق إنشاء.

  6. انقر فوق هنا لبدء الإعداد السريع لهذا الكمبيوتر في قسم الخيار 1: إعداد Express .

    انقر فوق ارتباط الإعداد السريع

  7. في نافذة تشغيل التكامل لـ Express Setup (ذاتي الاستضافة)، انقر فوق إغلاق.

    إعداد وقت تشغيل التكامل - ناجح

  8. في مستعرض الويب، في نافذة إعداد وقت التشغيل التكامل، انقر فوق إنهاء.

  9. تأكد من أنك ترى MySelfHostedIR في قائمة أوقات تشغيل التكامل.

إنشاء linked services

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

إنشاء الخدمة المرتبطة SQL Server

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

  1. في إطار الاتصالات، قم بالتبديل من علامة التبويب أوقات تشغيل التكامل إلى علامة التبويب الخدمات المرتبطة، وانقر فوق + جديد.

    خدمة مرتبطة جديدة.

  2. في إطار "الخدمة المرتبطة الجديدة"، حدد SQL Server، وانقر فوق متابعة.

  3. في نافذة New Linked Service، قم بالخطوات التالية:

    1. أدخل SqlServerLinkedServiceللاسم.
    2. حدد MySelfHostedIRالاتصال عبر وقت تشغيل التكامل. وهذه خطوة هامة. لا يمكن الاتصال وقت تشغيل التكامل الافتراضي إلى مخزن بيانات محلي. استخدم وقت تشغيل التكامل المستضاف ذاتيا الذي قمت بإنشائه سابقا.
    3. بالنسبة إلى اسم الملقم، أدخل اسم الكمبيوتر الذي يحتوي على قاعدة بيانات SQL Server.
    4. بالنسبة إلى اسم قاعدة البيانات، أدخل اسم قاعدة البيانات في SQL Server الذي يحتوي على بيانات المصدر. إنشاء جدول وإدراج البيانات في قاعدة البيانات هذه كجزء من المتطلبات الأساسية.
    5. بالنسبة لنوع المصادقة، حدد نوع المصادقة التي تريد استخدامها للاتصال بقاعدة البيانات.
    6. بالنسبة إلى اسم المستخدم، أدخل اسم المستخدم الذي لديه حق الوصول إلى قاعدة بيانات SQL Server. إذا كنت بحاجة إلى استخدام حرف مائل ( \ ) في حساب المستخدم أو اسم الخادم، فاستخدم حرف الهروب ( \ ). مثال على ذلك mydomain\\myuser .
    7. وبالنسبة لكلمة المرور، أدخل كلمة مرور المستخدم.
    8. لاختبار ما إذا كان مصنع البيانات يمكنه الاتصال بقاعدة بيانات SQL Server، فانقر فوق اختبار الاتصال. إصلاح أي أخطاء حتى ينجح الاتصال.
    9. لحفظ الخدمة المرتبطة، انقر فوق إنهاء.

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

في الخطوة الأخيرة، يمكنك إنشاء خدمة مرتبطة لربط قاعدة بيانات المصدر SQL Server بمصنع البيانات. في هذه الخطوة، يمكنك ربط قاعدة بيانات الوجهة/الحوض بمصنع البيانات.

  1. في إطار الاتصالات، قم بالتبديل من علامة التبويب أوقات تشغيل التكامل إلى علامة التبويب الخدمات المرتبطة، وانقر فوق + جديد.

  2. في نافذة "New Linked Service"، حدد "Azure SQL Database"، ثم حدد "Continue".

  3. في نافذة New Linked Service، قم بالخطوات التالية:

    1. سجل "AzureSqlDatabaseLinkedService" في مربع "Name".
    2. بالنسبة لاسم الخادم، حدد اسم الخادم من القائمة المنسدلة.
    3. بالنسبة إلى اسم قاعدة البيانات، حدد قاعدة البيانات التي قمت بإنشائها customer_table project_table كجزء من المتطلبات الأساسية.
    4. بالنسبة إلى اسم المستخدم، أدخل اسم المستخدم الذي لديه حق الوصول إلى قاعدة البيانات.
    5. وبالنسبة لكلمة المرور، أدخل كلمة مرور المستخدم.
    6. لاختبار ما إذا كان مصنع البيانات يمكنه الاتصال بقاعدة بيانات SQL Server، فانقر فوق اختبار الاتصال. إصلاح أي أخطاء حتى ينجح الاتصال.
    7. لحفظ الخدمة المرتبطة، انقر فوق إنهاء.

  4. تأكد من مشاهدة اثنين من الخدمات المرتبطة في القائمة.

    خدمتان مرتبطتان

إنشاء datasets

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

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

  1. أنقر فوق "+ (plus)" في اللوحة اليسرى، ثم أنقر فوق "مجموعة البيانات".

  2. في نافذة مجموعة البيانات الجديدة، حدد SQL Server، انقر فوق متابعة.

  3. تظهر علامة تبويب جديدة مفتوحة في مستعرض ويب لتكوين مجموعة البيانات. كما ترى مجموعة بيانات في طريقة عرض الشجرة. في علامة التبويب عام من إطار خصائص في الأسفل، أدخل SourceDatasetللاسم.

  4. التبديل إلى علامة التبويب اتصال في إطار خصائص وحدد SqlServerLinkedServiceللخدمة المرتبطة. لا تحدد جدولاً هنا. يستخدم نشاط النسخ في خط الأنابيب استعلام SQL لتحميل البيانات بدلاً من تحميل الجدول بأكمله.

    مجموعة البيانات المصدر - اتصال

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

  1. أنقر فوق "+ (plus)" في اللوحة اليسرى، ثم أنقر فوق "مجموعة البيانات".

  2. في إطار مجموعة البيانات الجديدة، حدد قاعدة بيانات azure SQL، وانقر فوق متابعة.

  3. تظهر علامة تبويب جديدة مفتوحة في مستعرض ويب لتكوين مجموعة البيانات. كما ترى مجموعة بيانات في طريقة عرض الشجرة. في علامة التبويب عام من إطار خصائص في الأسفل، أدخل SinkDatasetللاسم.

  4. التبديل إلى علامة التبويب معلمات في نافذة "خصائص"، ثم قم بالخطوات التالية:

    1. انقر فوق جديد في قسم إنشاء/تحديث المعلمات .

    2. أدخل SinkTableNameللاسم، وسلسلةللنوع. تأخذ مجموعة البيانات هذه SinkTableName كمعلمة. يتم تعيين المعلمة SinkTableName بواسطة خط أنابيب بشكل حيوي في وقت التشغيل. النشاط ForEach في خط الأنابيب تكرار خلال قائمة أسماء الجداول ويقوم بتمرير اسم الجدول إلى مجموعة البيانات هذه في كل تكرار.

      مجموعة بيانات

  5. التبديل إلى علامة التبويب اتصال في نافذة "خصائص"، وحدد AzureSqlDatabaseLinkedService لـ الخدمة المرتبطة. بالنسبة إلى خاصية الجدول، انقر فوق إضافة محتوى ديناميكي.

  6. في نافذة إضافة محتوى حيوي حدد SinkTableName في قسم معلمات.

  7. بعد النقر فوق إنهاء، سترى "@dataset().SinkTableName" كاسم الجدول.

    مجموعة بيانات التخزين - الاتصال

أنشئ مجموعة بيانات لهذه العلامة المائية

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

  1. أنقر فوق "+ (plus)" في اللوحة اليسرى، ثم أنقر فوق "مجموعة البيانات".

  2. في إطار مجموعة البيانات الجديدة، حدد قاعدة بيانات azure SQL، وانقر فوق متابعة.

  3. في علامة التبويب عام من نافذة "خصائص" في الأسفل، أدخل مجموعة بيانات مائيةللاسم.

  4. انتقل إلى علامة التبويب Connection ثم قم بالخطوات التالية:

    1. حدد AzureSqlDatabaseLinkedServiceللخدمة المرتبطة.

    2. حدد [dbo].[watermarktable] لـ الجدول.

      مجموعة بيانات العلامة المائية - اتصال

إنشاء البنية الأساسية لبرنامج ربط العمليات التجارية

يأخذ خط أنابيب قائمة أسماء الجداول كمعلمة. نشاط ForEach يكرر من خلال قائمة أسماء الجداول وتنفيذ العمليات التالية:

  1. استخدم "نشاط البحث" لاسترداد قيمة العلامة المائية القديمة (القيمة الأولية أو التي تم استخدامها في التكرار الأخير).

  2. استخدم "نشاط البحث" لاسترداد قيمة العلامة المائية الجديدة (القيمة القصوى لعمود العلامة المائية في الجدول المصدر).

  3. استخدم "نشاط النسخ" لنسخ البيانات بين قيمتي العلامة المائية من قاعدة البيانات المصدر إلى قاعدة البيانات الوجهة.

  4. استخدم "نشاط StoredProcedure" لتحديث قيمة العلامة المائية القديمة لاستخدامها في الخطوة الأولى من التكرار التالي.

إنشاء البنية الأساسية لبرنامج ربط العمليات التجارية

  1. أنقر فوق "+ (plus)" في اللوحة اليسرى، ثم أنقر فوق "Pipeline".

  2. في اللوحة العامة ضمن خصائص، حدد IncrementalCopyPipelineللاسم. ثم قم بطي اللوحة بالنقر فوق رمز الخصائص في الزاوية العلوية اليمنى.

  3. في علامة التبويب معلمات، قم بالخطوات التالية:

    1. انقر على + New.
    2. أدخل "tableList" في مربع "name"المعلمة.
    3. حدد صفيف لنوع المعلمة.

  4. في مربع أدوات "Activities"، قم بتوسيع "Iteration & Conditions"، ثم أسحب وأفلت نشاط ForEach إلى سطح مصمم البنية الأساسية. في علامة التبويب عام من نافذة خصائص أدخل IterateSQLTables.

  5. قم بالتبديل إلى علامة التبويب الإعدادات، وأدخل @pipeline().parameters.tableListلـ العناصر. النشاط ForEach يكرر من خلال قائمة من الجداول وتنفيذ عملية النسخ التزايدي.

    نشاط ForEach - الإعدادات

  6. حدد نشاط ForEach قيد الإعداد إذا لم يكن محدداً مسبقاً. انقر فوق زر تحرير (رمز قلم رصاص) .

  7. في مربع أدوات الأنشطة، قم بتوسيع عام، وسحب-إفلات نشاط البحث إلى سطح مصمم خط الأنابيب، وأدخل LookupOldWaterMarkActivityللاسم.

  8. التبديل إلى علامة التبويب الإعدادات من إطار خصائص ثم قم بالخطوات التالية:

    1. حدد مجموعة بيانات العلامة المائية لـ مجموعة بيانات المصدر.

    2. حدد "Query" تحت "Use query".

    3. أدخل استعلام SQL التالي في Query.

      select * from watermarktable where TableName  =  '@{item().TABLE_NAME}'

      نشاط البحث الأول - الإعدادات

  9. سحب-إفلات نشاط البحث من مربع أدوات الأنشطة، وأدخل بحثNewWaterMarkActivityللاسم.

  10. الانتقال إلى علامة تبويب الإعدادات.

    1. حدد مجموعة بيانات المصدر لـ مجموعة البيانات المصدر.

    2. حدد "Query" تحت "Use query".

    3. أدخل استعلام SQL التالي في Query.

      select MAX(@{item().WaterMark_Column}) as NewWatermarkvalue from @{item().TABLE_NAME}

      نشاط البحث الثاني - الإعدادات

  11. سحب-إفلات نشاط النسخ من مربع أدوات الأنشطة، وأدخل IncrementalCopyActivityللاسم.

  12. الاتصال أنشطة البحث إلى نشاط النسخ واحدا تلو الآخر. للاتصال، ابدأ بالسحب في المربع الأخضر المرفق بنشاط البحث وأسقطه على نشاط النسخ. حرر زر الماوس عند تغيير لون حد نشاط النسخ إلى اللون الأزرق.

    الاتصال بأنشطة بحث لنسخ النشاط

  13. حدد نشاط النسخ في "pipeline". التبديل إلى علامة التبويب المصدر في نافذة خصائص.

    1. حدد مجموعة بيانات المصدر لـ مجموعة البيانات المصدر.

    2. حدد "Query" تحت "Use query".

    3. أدخل استعلام SQL التالي في Query.

      select * from @{item().TABLE_NAME} where @{item().WaterMark_Column} > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and @{item().WaterMark_Column} <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'

      Copy Activity - إعدادات المصدر

  14. التبديل إلى علامة التبويب "بالوعة" ثم حدد SinkDataset لمجموعة بيانات مصدر.

  15. قم بالخطوات التالية:

    1. في خصائص مجموعة البيانات، للمعلمة SinkTableName، وأدخل @{item().TABLE_NAME} .

    2. بالنسبة إلى خاصية اسم الإجراء المخزن، أدخل @{item().StoredProcedureNameForMergeOperation}.

    3. بالنسبة إلى خاصية نوع الجدول، أدخل @{item().TableType} .

    4. بالنسبة إلى اسم معلمة نوع الجدول، أدخل @{item().TABLE_NAME} .

      نسخ النشاط - المعلمات

  16. سحب وإفلات نشاط الإجراء المخزن من مربع أدوات الأنشطة إلى سطح مصمم خط أنابيب. الاتصال نشاط النسخ إلى نشاط الإجراء المخزن .

  17. حدد نشاط الإجراء المخزن في خط الأنابيب، ثم أدخل StoredProceduretoWriteWatermarkActivityللاسم في علامة التبويب عام من نافذة خصائص.

  18. التبديل إلى علامة التبويب حساب SQL، ثم حدد AzureSqlDatabaseLinkedServiceللخدمة المرتبطة.

    نشاط الإجراء المخزن - حساب SQL

  19. التبديل إلى علامة التبويب إجراء مخزن ثم قم بالخطوات التالية:

    1. بالنسبة إلى اسم الإجراء المخزن، حدد [dbo].[usp_write_watermark] .

    2. حدد استيراد المعلمة.

    3. حدد القيم التالية المعلمات:

      الاسم نوع القيمة‬
      LastModifiedtime التاريخ والوقت @{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}
      اسم الجدول السلسلة‬ @{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}

      نشاط الإجراء المخزن - إعدادات الإجراء المخزن

  20. حدد نشر الكل لنشر الكيانات التي أنشأتها لخدمة مصنع البيانات.

  21. انتظر حتى ترى رسالة تم النشر بنجاح. للاطلاع على الإشعارات، انقر على ارتباط إظهار الإشعارات. أغلق نافذة الإعلامات بالنقر فوق X.

قم بتشغيل البنية الأساسية

  1. على شريط الأدوات لخط الأنابيب، انقر فوق إضافة تشغيل، وانقر فوق المشغل الآن.

  2. في نافذة "تشغيل Pipeline"، أدخل القيمة التالية للمعلمة tableList، وانقر فوق إنهاء.

    [
    {
        "TABLE_NAME": "customer_table",
        "WaterMark_Column": "LastModifytime",
        "TableType": "DataTypeforCustomerTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_customer_table"
    },
    {
        "TABLE_NAME": "project_table",
        "WaterMark_Column": "Creationtime",
        "TableType": "DataTypeforProjectTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_project_table"
    }
]

    وسيطات تشغيل

مراقبة المسار

  1. يمكنك التبديل إلى علامة التبويب "Monitor" في الجهة اليسرى. ترى تشغيل "Pipeline" تم تشغيله بواسطة التشغيل اليدوي. يمكنك استخدام الارتباطات ضمن العمود PIPELINE NAME لعرض تفاصيل النشاط وإعادة تشغيل "تدفقات".

  2. لمشاهدة تشغيل النشاط المقترنة بتشغيل خط أنابيب حدد الارتباط ضمن عمود NAME PIPELINE. للحصول على تفاصيل حول تشغيل النشاط، حدد الارتباط تفاصيل (رمز النظارات) ضمن العمود اسم النشاط.

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

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

في SQL Server Management Studio تشغيل الاستعلامات التالية مقابل قاعدة بيانات SQL الهدف للتحقق من أن تم نسخ البيانات من جداول المصدر إلى جداول الوجهة:

استفسار

select * from customer_table

الناتج

===========================================
PersonID	Name	LastModifytime
===========================================
1	        John	2017-09-01 00:56:00.000
2	        Mike	2017-09-02 05:23:00.000
3	        Alice	2017-09-03 02:36:00.000
4	        Andy	2017-09-04 03:21:00.000
5	        Anny	2017-09-05 08:06:00.000

استفسار

select * from project_table

الناتج

===================================
Project	    Creationtime
===================================
project1	2015-01-01 00:00:00.000
project2	2016-02-02 01:23:00.000
project3	2017-03-04 05:16:00.000

استفسار

select * from watermarktable

الناتج

======================================
TableName	    WatermarkValue
======================================
customer_table	2017-09-05 08:06:00.000
project_table	2017-03-04 05:16:00.000

لاحظ أنه تم تحديث قيم العلامة المائية لكلا الجدولين.

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

تشغيل الاستعلام التالي مقابل قاعدة بيانات المصدر SQL Server لتحديث صف موجود في customer_table. إدراج صف جديد في project_table.

UPDATE customer_table
SET [LastModifytime] = '2017-09-08T00:00:00Z', [name]='NewName' where [PersonID] = 3

INSERT INTO project_table
(Project, Creationtime)
VALUES
('NewProject','10/1/2017 0:00:00 AM');

أعد تشغيل التدفق

  1. في نافذة مستعرض ويب، انتقل إلى علامة التبويب تحرير على اليسار.

  2. على شريط الأدوات لخط الأنابيب، انقر فوق إضافة تشغيل، وانقر فوق المشغل الآن.

  3. في نافذة "تشغيل Pipeline"، أدخل القيمة التالية للمعلمة tableList، وانقر فوق إنهاء.

    [
    {
        "TABLE_NAME": "customer_table",
        "WaterMark_Column": "LastModifytime",
        "TableType": "DataTypeforCustomerTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_customer_table"
    },
    {
        "TABLE_NAME": "project_table",
        "WaterMark_Column": "Creationtime",
        "TableType": "DataTypeforProjectTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_project_table"
    }
]

مراقبة تشغيل "pipeline"

  1. يمكنك التبديل إلى علامة التبويب "Monitor" في الجهة اليسرى. ترى تشغيل "Pipeline" تم تشغيله بواسطة التشغيل اليدوي. يمكنك استخدام الارتباطات ضمن العمود PIPELINE NAME لعرض تفاصيل النشاط وإعادة تشغيل "تدفقات".

  2. لمشاهدة تشغيل النشاط المقترنة بتشغيل خط أنابيب حدد الارتباط ضمن عمود NAME PIPELINE. للحصول على تفاصيل حول تشغيل النشاط، حدد الارتباط تفاصيل (رمز النظارات) ضمن العمود اسم النشاط.

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

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

في SQL Server Management Studio تشغيل الاستعلامات التالية مقابل قاعدة بيانات SQL الهدف للتحقق من أن تم نسخ البيانات المحدثة/الجديدة من جداول المصدر إلى جداول الوجهة.

استفسار

select * from customer_table

الناتج

===========================================
PersonID	Name	LastModifytime
===========================================
1	        John	2017-09-01 00:56:00.000
2	        Mike	2017-09-02 05:23:00.000
3	        NewName	2017-09-08 00:00:00.000
4	        Andy	2017-09-04 03:21:00.000
5	        Anny	2017-09-05 08:06:00.000

لاحظ القيم الجديدة من الاسم وLastModifytime لـPersonID رقم 3.

استفسار

select * from project_table

الناتج

===================================
Project	    Creationtime
===================================
project1	2015-01-01 00:00:00.000
project2	2016-02-02 01:23:00.000
project3	2017-03-04 05:16:00.000
NewProject	2017-10-01 00:00:00.000

لاحظ أنه تمت إضافة إدخال مشروع جديد إلى project_table.

استفسار

select * from watermarktable

الناتج

======================================
TableName	    WatermarkValue
======================================
customer_table	2017-09-08 00:00:00.000
project_table	2017-10-01 00:00:00.000

لاحظ أنه تم تحديث قيم العلامة المائية لكلا الجدولين.

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

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

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

انتقل إلى البرنامج التعليمي التالي لمعرفة المزيد حول تحويل البيانات باستخدام مجموعة Spark على Azure:

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

  • Last updated on