إشعار
يتطلب الوصول إلى هذه الصفحة تخويلاً. يمكنك محاولة تسجيل الدخول أو تغيير الدلائل.
يتطلب الوصول إلى هذه الصفحة تخويلاً. يمكنك محاولة تغيير الدلائل.
ينطبق على:
Azure Data Factory
Azure Synapse Analytics
تلميح
جرب Data Factory في Microsoft Fabric، وهو حل تحليلي متكامل للمؤسسات. يغطي Microsoft Fabric كل شيء بدءا من حركة البيانات إلى علم البيانات والتحليلات في الوقت الحقيقي والمعلومات المهنية وإعداد التقارير. تعرف على كيفية بدء إصدار تجريبي جديد مجانا!
في هذا البرنامج التعليمي، يمكنك إنشاء Azure Data Factory باستخدام "تدفقات" لتحميل بيانات دلتا من جدول في قاعدة بيانات Azure SQL إلى مخزن Azure Blob.
نفذ الخطوات التالية في هذا البرنامج التعليمي:
- أعد مخزن بيانات لتخزين قيمة العلامة المائية.
- إنشاء data factory.
- أنشئ الخدمة ذات الصلة.
- أنشئ مجموعات بيانات المورد والمصدر والعلامة المائية.
- إنشاء البنية الأساسية لبرنامج ربط العمليات التجارية.
- تشغيل البنية الأساسية لبرنامج ربط العمليات التجارية.
- مراقبة تشغيل المسار.
- مراجعة النتائج
- أضف المزيد من البيانات إلى المصدر.
- تشغيل "تدفقات" مرة أخرى.
- راقب تشغيل "تدفقات" الثاني
- مراجعة النتائج من الجولة الثانية
نظرة عامة
فيما يلي رسم تخطيطي لحل رفيع المستوى:
فيما يلي الخطوات الهامة لإنشاء هذا الحل:
حدد عمود العلامة المائية. حدد عموداً واحداً في مخزن بيانات المصدر، والذي يمكن استخدامه لتقطيع السجلات الجديدة أو المحدثة لكل تشغيل. عادةً ما تستمر البيانات الموجودة في هذا العمود المحدد في الزيادة (على سبيل المثال، last_modify_time أو معرف) عند إنشاء صفوف أو تحديثها. تستخدم أعلى قيمة في هذا العمود كعلامة مائية.
أعد مخزن بيانات لتخزين قيمة العلامة المائية. في هذا البرنامج التعليمي، يمكنك تخزين قيمة العلامة المائية في قاعدة بيانات SQL.
أنشئ بنية أساسية مع سير العمل التالي:
تحتوي البنية الأساسية لهذا الحل على الأنشطة التالية:
- إنشاء نشاطين من أنشطة البحث. استخدم نشاط البحث الأول لاسترداد قيمة العلامة المائية الأخيرة. استخدم نشاط البحث الثاني لاسترداد قيمة العلامة المائية الجديدة. يتم تمرير قيم هذه العلامات المائية إلى نشاط النسخ.
- أنشئ نشاط نسخ ينسخ صفوف من مخزن بيانات المصدر مع إدخال قيمة لعمود العلامة المائية أكبر من قيمة العلامة المائية القديمة وأقل من قيمة العلامة المائية الجديدة. ثم، ينسخ بيانات دلتا من مخزن بيانات المصدر إلى Blob storage كملف جديد.
- أنشئ نشاط StoredProcedure الذي يحدث قيمة العلامة المائية للمسار الذي سيتم تشغيله في المرة القادمة.
في حال لم يكن لديك اشتراك Azure، فأنشئ حساباً مجانيّاً قبل البدء.
المتطلبات الأساسية
- قاعدة بيانات Azure SQL. يمكنك استخدام قاعدة البيانات كمخزن لبيانات المصدر. في حالة عدم امتلاك قاعدة بيانات فيAzure SQL Database، يرجى الرجوع إلى إنشاء قاعدة بيانات في Azure SQL Database للاطلاع على خطواتإنشاء واحدة.
- Azure Storage. يمكنك استخدام blob storage كمخزن لمصدر البيانات. إذا لم يكن لديك حساب تخزين، فيمكنك مراجعة إنشاء حساب تخزينللاطلاع على خطوات إنشاء الحساب. أنشئ حاوية باسم adftutorial.
أنشئ جدول مصدر بيانات في قاعدة بيانات SQL
فتح SQL Server Management Studio. في مستكشف العناصرانقر بزر الماوس الأيمن فوق database وحدد New Query.
شغل أمرSQL التالي مقابل قاعدة بيانات Azure لإنشاء جدول باسم
data_source_tableمخزن مصدر البيانات:create table data_source_table ( PersonID int, Name varchar(255), LastModifytime datetime ); INSERT INTO data_source_table (PersonID, Name, LastModifytime) VALUES (1, 'aaaa','9/1/2017 12:56:00 AM'), (2, 'bbbb','9/2/2017 5:23:00 AM'), (3, 'cccc','9/3/2017 2:36:00 AM'), (4, 'dddd','9/4/2017 3:21:00 AM'), (5, 'eeee','9/5/2017 8:06:00 AM');في هذا البرنامج التعليمي، يمكنك استخدام LastModifytime كعمود علامة مائية. تظهر البيانات الموجودة في مخزن مصدر البيانات في الجدول التالي:
PersonID | Name | LastModifytime -------- | ---- | -------------- 1 | aaaa | 2017-09-01 00:56:00.000 2 | bbbb | 2017-09-02 05:23:00.000 3 | cccc | 2017-09-03 02:36:00.000 4 | dddd | 2017-09-04 03:21:00.000 5 | eeee | 2017-09-05 08:06:00.000
أنشئ جدول أخر في قاعدة بيانات SQL لتخزين أعلى قيمة للعلامة المائية
شغل أمر SQL التالي مقابل قاعدة بياناتك في SQL لإنشاء جدول باسم
watermarktableلتخزين قيمة العلامة المائية:create table watermarktable ( TableName varchar(255), WatermarkValue datetime, );عين القيمة الافتراضية للعلامة المائية العالية مع اسم جدول مخزن بيانات المصدر. في هذا البرنامج التعليمي، اسم الجدول هو data_source_table.
INSERT INTO watermarktable VALUES ('data_source_table','1/1/2010 12:00:00 AM')راجع البيانات الموجودة في الجدول
watermarktable.Select * from watermarktableإخراج:
TableName | WatermarkValue ---------- | -------------- data_source_table | 2010-01-01 00:00:00.000
أنشئ إجراء مخزن في قاعدة بياناتك بـSQL
شغل الأمر التالي لإنشاء إجراء مخزن في قاعدة بياناتك في SQL:
CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS
BEGIN
UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime
WHERE [TableName] = @TableName
END
إنشاء مصدرًا للبيانات
شغِّل إما مستعرض الويب Microsoft Edge أو Google Chrome. يتم حاليًّا دعم واجهة مستخدم Data Factory فقط في مستعرضي الويب Microsoft Edge وGoogle Chrome.
في القائمة العلوية، حدد Create a resource>Analytics>Data Factory :
في صفحة مصنع بيانات جديد، أدخل ADFIncCopyTutorialDFللاسم.
يجب أن يكون اسم Azure data factory مميزاً بصفة عمومية. إذا رأيت علامة تعجب حمراء مع الخطأ التالي تغيير اسم مصنع البيانات (على سبيل المثال، yournameADFIncCopyTutorialDF) ثم حاول إنشاء مرة أخرى. راجع المقالةData Factory - Naming Rules لمعرفة قواعد تسمية أدوات مصنع البيانات.
اسم مصنع بيانات "ADFIncCopyTutorialDF" غير متوفر
حدد اشتراك Azure الذي تريد إنشاء مصنع البيانات فيه.
بالنسبة Resource Group، نفِّذ إحدى الخطوات التالية:
حدد Use existing واختر مجموعة موارد موجودة من القائمة المنسدلة.
حدد Create new وأدخل اسم مجموعة الموارد.
للتعرف على مجموعات الموارد، يُرجى الرجوع إلى استخدام مجموعات الموارد لإدارة موارد Azure.
حدد "V2" لحقل"version".
حدد موقع، مصِّنع البيانات. لن تظهر القائمة المنسدلة إلا على المواقع المعتمدة فقط. يمكن أن تكون مخازن البيانات (تخزين Azure، وقاعدة بيانات azure SQL، ومثيل Azure SQL المدار، وما إلى ذلك) والحوسبة (HDInsight، وما إلى ذلك) المستخدمة من قبل مصنع البيانات في مناطق أخرى.
انقر فوق Create.
بعد الانتهاء من الإنشاء، ستظهر صفحة مصنع البيانات كما هو موضح في الصورة.
حدد Open على تجانب Open Azure Data Factory Studio لبدء تشغيل تطبيق واجهة مستخدم Azure Data Factory الموجود في علامة تبويب منفصلة.
إنشاء البنية الأساسية لبرنامج ربط العمليات التجارية
في هذا البرنامج التعليمي، يمكنك إنشاء البنية الأساسية مع اثنين من أنشطة البحث، نشاط نسخ واحد، ونشاط StoredProcedure واحد مسلسلاً في البنية الأساسية.
في الصفحة الرئيسية من واجهة مستخدم Data Factory، انقر فوق إطار Orchestrate.
في اللوحة العامة ضمن خصائص، حدد IncrementalCopyPipelineللاسم. ثم قم بطي اللوحة بالنقر فوق رمز الخصائص في الزاوية العلوية اليمنى.
دعونا نضِف أول نشاط بحث للحصول على قيمة العلامة المائية القديمة. في مربع أدوات الأنشطة، قم بتوسيع عام، ثم قم بسحب وإفلات نشاط البحث إلى سطح مصمم التدفقات. تغيير اسم النشاط إلى LookupOldWaterMarkActivity.
قم بالتبديل إلى علامة التبويب الإعدادات، وانقر فوق + جديد لمجموعة البيانات المصدر. في هذه الخطوة، يمكنك إنشاء مجموعة بيانات لتمثيل البيانات في watermarktable. يحتوي هذا الجدول على العلامة المائية القديمة التي تم استخدامها في عملية النسخ السابقة.
في إطار مجموعة البيانات الجديدة، حدد قاعدة بيانات azure SQL، وانقر فوق متابعة. ترى نافذة جديدة مفتوحة لمجموعة البيانات.
في إطار تعيين خصائص لمجموعة البيانات، أدخل WatermarkDatasetللاسم.
بالنسبة إلى الخدمة المرتبطة، حدد جديد، ثم قم بالخطوات التالية:
سجل "AzureSqlDatabaseLinkedService" في مربع "Name".
حدد خادمك في مربع "Server name".
حدد اسم قاعدة البيانات من القائمة المنسدلة.
أدخل اسم المستخدم وكلمة المرور.
لاختبار الاتصال بقاعدة بيانات SQL، انقر فوق اختبار الاتصال.
انقر فوق إنهاء.
تأكد من تحديد AzureSqlDatabaseLinkedServiceللخدمة ذات الصلة.
حدد إنهاء.
في علامة التبويب "Connection"، حدد [dbo].[watermarktable] من أجل الجدول. إذا كنت ترغب في معاينة البيانات في الجدول، فانقر فوق معاينة البيانات.
التبديل إلى محرر خط أنابيب بالنقر فوق علامة التبويب "تدفقات" في الجزء العلوي أو بالنقر فوق اسم خط الأنابيب في طريقة العرض شجرة على اليسار. في نافذة خصائص نشاط البحث، تأكد من تحديد WatermarkDataset لحقل مجموعة البيانات المصدر.
في مربع أدوات الأنشطة، قم بتوسيع عام، ثم قم بسحب وإفلات نشاط بحث آخر إلى سطح مصمم "تدفقات"، وقم بتعيين الاسم إلى LookupNewWaterMarkActivity في علامة التبويب عامة من إطار الخصائص. يحصل نشاط البحث هذا على قيمة العلامة المائية الجديدة من الجدول مع البيانات المصدر المراد نسخها إلى الوجهة.
في نافذة خصائص نشاط البحث الثاني، قم بالتبديل إلى علامة التبويب الإعدادات، وانقر فوق جديد. إنشاء مجموعة بيانات للإشارة إلى الجدول المصدر الذي يحتوي على قيمة العلامة المائية الجديدة (القيمة القصوى لـLastModifyTime).
في إطار مجموعة البيانات الجديدة، حدد قاعدة بيانات azure SQL، وانقر فوق متابعة.
في إطار تعيين خصائص، أدخل SourceDatasetللاسم. حدد AzureSqlDatabaseLinkedServiceللخدمة المرتبطة.
حدد [dbo].[ data_source_table] للجدول. يمكنك تحديد استعلام على مجموعة البيانات هذه لاحقا في البرنامج التعليمي. يأخذ الاستعلام الأسبقية على الجدول الذي تحدده في هذه الخطوة.
حدد إنهاء.
التبديل إلى محرر خط أنابيب بالنقر فوق علامة التبويب "تدفقات" في الجزء العلوي أو بالنقر فوق اسم خط الأنابيب في طريقة العرض شجرة على اليسار. في نافذة خصائص نشاط البحث، تأكد من تحديد SourceDataset للحقل "مجموعة البيانات المصدر".
حدد استعلام للحقل "استخدام استعلام" وأدخل الاستعلام التالي: أنت تقوم بتحديد القيمة القصوى لـLastModifytime فقط من data_source_table. يُرجى التأكد من أنك قد راجعت أيضاً الصف الأول فقط.
select MAX(LastModifytime) as NewWatermarkvalue from data_source_table
في مربع أدوات الأنشطة، قم بتوسيع نقل وتحويل، ثم قم بسحب وإفلات نشاط النسخ من مربع أدوات الأنشطة، وقم بتعيين الاسم إلى IncrementalCopyActivity.
قم بتوصيل أنشطة البحث بنشاط النسخ عن طريق سحب الزر الأخضر المرفق بأنشطة البحث إلى نشاط النسخ. تحرير زر الماوس عندما ترى لون حد نشاط النسخ يتغير إلى اللون الأزرق.
حدد نشاط النسخ وتأكد من رؤية خصائص النشاط في نافذة "Properties".
يمكنك التبديل إلى علامة التبويب "Source" في نافذة "Properties" وأكمل هذه الخطوات:
حدد SourceDataset للحقل "مجموعة البيانات المصدر".
حدد استعلام للحقل "استخدام استعلام".
أدخل استعلام SQL التالي للحقل استعلام.
select * from data_source_table where LastModifytime > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and LastModifytime <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'
قم بالتبديل إلى علامة التبويب Sink، وانقر فوق + جديد لحقل "مجموعة بيانات الحوض".
في هذا البرنامج التعليمي مخزن البيانات "sink" هو من نوع تخزين Azure Blob. لذلك، حدد تخزين النقطة Azure، ثم انقر فوق متابعة في نافذة مجموعة بيانات جديدة.
في نافذة تحديد تنسيق، حدد نوع تنسيق البيانات، وانقر فوق متابعة.
في نافذة تعيين خصائص أدخل SinkDatasetللاسم. تحت "Linked service"، حدد "New". في هذه الخطوة، يمكنك إنشاء اتصال (خدمة مرتبطة) إلى تخزين Azure Blob.
في نافذة "New Linked Service (Azure Blob Storage)"، نفذ الخطوات التالية:
- سجل AzureSqlDatabaseLinkedService في مربع "Name".
- بالنسبة لاسم حساب التخزين، حدد اسم حساب تخزين Azure الخاص بك.
- اختبار الاتصال ثم انقر فوق إنهاء.
في نافذة "Set Properties"، تأكد من تحديد AzureStorageLinkedService لـ "Linked service". ثم حدد إنهاء.
انتقل إلى علامة التبويب اتصال لـ SinkDataset ثم قم بالخطوات التالية:
- لحقل مسار الملف، أدخل adftutorial/incrementalcopy. adftutorial هو اسم حاوية blob وincrementalcopy هو اسم المجلد. يفترض هذا المقتطف أن لديك حاوية blob المسماة adftutorial في تخزين blob. أنشئ الحاوية إذا لم تكن موجودة، أو استخدم اسم حاوية موجودة. تقوم Azure Data Factory بإنشاء مجلد الإخراج incrementalcopy إذا لم يكن موجوداً. يمكنك استخدام زر "Browse" لدى مسار الملف للتنقل إلى مجلد الحاوية الثنائية كبيرة الحجم.
- بالنسبة إلى جزء الملف من الحقل مسار الملف، حدد إضافة محتوى ديناميكي [Alt+P]ثم أدخل
@CONCAT('Incremental-', pipeline().RunId, '.txt')في النافذة المفتوحة. ثم حدد إنهاء. يتم إنشاء اسم الملف بشكل حيوي باستخدام التعبير. كل مسار له معرّف فريد. يستخدم نشاط النسخ معرف التشغيل لإنشاء اسم الملف.
التبديل إلى محرر خط أنابيب بالنقر فوق علامة التبويب "تدفقات" في الجزء العلوي أو بالنقر فوق اسم خط الأنابيب في طريقة العرض شجرة على اليسار.
في مربع أدوات الأنشطة، قم بتوسيع عام، ثم قم بسحب-إسقاط نشاط الإجراء المخزن من مربع أدوات الأنشطة إلى سطح مصمم خط الأنابيب. توصيل المخرج الأخضر (ناجح) لنشاط النسخ بنشاط الإجراء المخزن.
حدد نشاط الإجراء المخزن في مصمم "تدفقات"، قم بتغيير اسمه إلى StoredProceduretoWriteWatermarkActivity.
قم بالتبديل إلى علامة التبويب حساب SQL، وحدد AzureSqlDatabaseLinkedServiceللخدمة ذات الصلة.
التبديل إلى علامة التبويب إجراء مخزن ثم قم بالخطوات التالية:
بالنسبة لاسم الإجراء المخزن، حدد usp_write_watermark.
لتحديد قيم معلمات الإجراء المخزن، انقر فوق استيراد المعلمة، وأدخل القيم التالية للمعلمات:
الاسم نوع القيمة LastModifiedtime التاريخ والوقت @{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue} اسم الجدول السلسلة @{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}
للتحقق من صحة إعدادات خط الأنابيب، انقر فوق التحقق من الصحة على شريط الأدوات. تأكد من عدم وجود أخطاء التحقق من الصحة. لإغلاق إطار تقرير التحقق من صحة "تدفقات" انقر فوق >>.
انشر الكيانات (الخدمات المرتبطة، ومجموعات البيانات، والمسارات) في خدمة Azure Data Factory عن طريق تحديد الزر "Publish All". انتظر حتى ترى رسالة نجاح النشر.
تتبع تشغيل البنية الأساسية
انقر فوق إضافة مشغل على شريط الأدوات، وانقر فوق المشغل الآن.
في نافذة تشغيل "تدفقات"، حدد إنهاء.
راقب عملية تشغيل البنية الأساسية لبرنامج ربط العمليات التجارية
يمكنك التبديل إلى علامة التبويب "Monitor" في الجهة اليسرى. ترى حالة تشغيل "تدفقات" تشغيل بواسطة مشغل يدوي. يمكنك استخدام الارتباطات ضمن العمود PIPELINE NAME لعرض تفاصيل التشغيل وإعادة تشغيل خط الأنابيب.
لمشاهدة تشغيل النشاط المقترنة بتشغيل خط أنابيب حدد الارتباط ضمن عمود NAME PIPELINE. للحصول على تفاصيل حول تشغيل النشاط، حدد الارتباط تفاصيل (رمز النظارات) ضمن العمود اسم النشاط. حدد جميع "تدفقات" قيد التشغيل في الأعلى للعودة إلى طريقة عرض "تشغيل خطوط الأنابيب". لإعادة تنشيط طريقة العرض، حدد Refresh.
مراجعة النتائج
الاتصال بحساب تخزين Azure باستخدام أدوات مثل مستكشف تخزين Azure. تحقق من إنشاء ملف إخراج في مجلد incrementalcopy للحاوية adftutorial.
افتح ملف الإخراج ولاحظ أن يتم نسخ جميع البيانات من data_source_table إلى ملف blob.
1,aaaa,2017-09-01 00:56:00.0000000 2,bbbb,2017-09-02 05:23:00.0000000 3,cccc,2017-09-03 02:36:00.0000000 4,dddd,2017-09-04 03:21:00.0000000 5,eeee,2017-09-05 08:06:00.0000000تحقق من أحدث قيمة من
watermarktable. يمكنك ملاحظة تحديث قيمة العلامة المائية.Select * from watermarktableوفيما يلي الناتج:
| TableName | WatermarkValue | | --------- | -------------- | | data_source_table | 2017-09-05 8:06:00.000 |
إضافة المزيد من البيانات إلى المصدر
إدراج بيانات جديدة في قاعدة البيانات (مخزن مصدر البيانات).
INSERT INTO data_source_table
VALUES (6, 'newdata','9/6/2017 2:23:00 AM')
INSERT INTO data_source_table
VALUES (7, 'newdata','9/7/2017 9:01:00 AM')
البيانات المحدثة في قاعدة البيانات الخاصة بك هي:
PersonID | Name | LastModifytime
-------- | ---- | --------------
1 | aaaa | 2017-09-01 00:56:00.000
2 | bbbb | 2017-09-02 05:23:00.000
3 | cccc | 2017-09-03 02:36:00.000
4 | dddd | 2017-09-04 03:21:00.000
5 | eeee | 2017-09-05 08:06:00.000
6 | newdata | 2017-09-06 02:23:00.000
7 | newdata | 2017-09-07 09:01:00.000
تشغيل "تدفقات" آخر
التبديل إلى علامة التبويب تحرير . انقر فوق "تدفقات" في طريقة عرض الشجرة إذا لم يتم فتحه في المصمم.
انقر فوق إضافة مشغل على شريط الأدوات، وانقر فوق المشغل الآن.
راقب تشغيل "تدفقات" الثاني
يمكنك التبديل إلى علامة التبويب "Monitor" في الجهة اليسرى. ترى حالة تشغيل "تدفقات" تشغيل بواسطة مشغل يدوي. يمكنك استخدام الارتباطات ضمن العمود PIPELINE NAME لعرض تفاصيل النشاط وإعادة تشغيل "تدفقات".
لمشاهدة تشغيل النشاط المقترنة بتشغيل خط أنابيب حدد الارتباط ضمن عمود NAME PIPELINE. للحصول على تفاصيل حول تشغيل النشاط، حدد الارتباط تفاصيل (رمز النظارات) ضمن العمود اسم النشاط. حدد جميع "تدفقات" قيد التشغيل في الأعلى للعودة إلى طريقة عرض "تشغيل خطوط الأنابيب". لإعادة تنشيط طريقة العرض، حدد Refresh.
تحقق من الناتج الثاني
في blob storage، تستطيع رؤية إنشاء ملف أخر. في هذا البرنامج التعليمي، اسم الملف الجديد هو
Incremental-<GUID>.txt. افتح هذا الملف، وسترى صفين من السجلات.6,newdata,2017-09-06 02:23:00.0000000 7,newdata,2017-09-07 09:01:00.0000000تحقق من أحدث قيمة من
watermarktable. يمكنك ملاحظة تحديث قيمة العلامة المائية.Select * from watermarktableعينة الإخراج:
| TableName | WatermarkValue | | --------- | -------------- | | data_source_table | 2017-09-07 09:01:00.000 |
المحتوى ذو الصلة
نفّذت الخطوات التالية في هذا البرنامج التعليمي:
- أعد مخزن بيانات لتخزين قيمة العلامة المائية.
- إنشاء data factory.
- أنشئ الخدمة ذات الصلة.
- أنشئ مجموعات بيانات المورد والمصدر والعلامة المائية.
- إنشاء البنية الأساسية لبرنامج ربط العمليات التجارية.
- تشغيل البنية الأساسية لبرنامج ربط العمليات التجارية.
- مراقبة تشغيل المسار.
- مراجعة النتائج
- أضف المزيد من البيانات إلى المصدر.
- تشغيل "تدفقات" مرة أخرى.
- راقب تشغيل "تدفقات" الثاني
- مراجعة النتائج من الجولة الثانية
في هذا البرنامج التعليمي، مسار البيانات المنسوخة من جدول واحد في قاعدة بيانات Azure SQL إلى Blob storage. انتقل إلى البرنامج التعليمي التالي لمعرفة كيفية نسخ البيانات من جداول متعددة في قاعدة بيانات SQL Server إلى قاعدة بيانات SQL.