إشعار
يتطلب الوصول إلى هذه الصفحة تخويلاً. يمكنك محاولة تسجيل الدخول أو تغيير الدلائل.
يتطلب الوصول إلى هذه الصفحة تخويلاً. يمكنك محاولة تغيير الدلائل.
ينطبق على:
Azure Data Factory Azure Synapse Analytics
تلميح
Data Factory في Microsoft Fabric هو الجيل القادم من Azure Data Factory، مع بنية أبسط، وذكاء اصطناعي مدمج، وميزات جديدة. إذا كنت جديدا في تكامل البيانات، ابدأ مع Fabric Data Factory. يمكن لأعباء عمل ADF الحالية الترقية إلى Fabric للوصول إلى قدرات جديدة في علوم البيانات، والتحليلات اللحظية، والتقارير.
في هذا الدرس، تنشئ Azure Data Factory مع خط أنابيب يحمل بيانات دلتا من جداول متعددة في قاعدة بيانات SQL Server إلى قاعدة بيانات Azure SQL.
نفذ الخطوات التالية في هذا البرنامج التعليمي:
- إعداد مخازن بيانات المصدر والوجهة.
- إنشاء data factory.
- إنشاء وقت تشغيل تكامل الاستضافة الذاتية.
- ثبت أداة وقت تشغيل التكامل.
- أنشئ الخدمة ذات الصلة.
- أنشئ مجموعات بيانات المورد والمصدر والعلامة المائية.
- إنشاء خط أنابيب وتشغيله ومراقبة.
- راجع النتائج.
- إضافة بيانات أو تحديثها في جداول المصدر.
- مراقبة تشغيل المسار.
- راجع النتائج التالية.
نظرة عامة
فيما يلي الخطوات الهامة لإنشاء هذا الحل:
حدد عمود العلامة المائية.
حدد عمودا واحدا لكل جدول في مخزن البيانات المصدر، والذي يمكنك تحديد السجلات الجديدة أو المحدثة لكل تشغيل. عادةً ما تستمر البيانات الموجودة في هذا العمود المحدد في الزيادة (على سبيل المثال، last_modify_time أو معرف) عند إنشاء صفوف أو تحديثها. تستخدم أعلى قيمة في هذا العمود كعلامة مائية.
أعد مخزن بيانات لتخزين قيمة العلامة المائية.
في هذا البرنامج التعليمي، يمكنك تخزين قيمة العلامة المائية في قاعدة بيانات SQL.
إنشاء خط أنابيب مع الأنشطة التالية:
إنشاء نشاط ForEach الذي يبتكر من خلال قائمة أسماء الجداول المصدر التي يتم تمريرها كمعلمة إلى خط الأنابيب. لكل جدول مصدر استدعاء الأنشطة التالية لتنفيذ تحميل دلتا لهذا الجدول.
إنشاء نشاطين من أنشطة البحث. استخدم نشاط البحث الأول لاسترداد قيمة العلامة المائية الأخيرة. استخدم نشاط البحث الثاني لاسترداد قيمة العلامة المائية الجديدة. يتم تمرير هذه القيم المائية إلى Copy activity.
أنشئ Copy activity ينسخ الصفوف من مخزن البيانات المصدر بحيث تكون قيمة عمود العلامة المائية أكبر من قيمة العلامة المائية القديمة وأقل من أو تساوي قيمة العلامة المائية الجديدة. ثم ينسخ بيانات دلتا من مخزن البيانات المصدر إلى تخزين Azure Blob كملف جديد.
أنشئ نشاط StoredProcedure الذي يحدث قيمة العلامة المائية للمسار الذي سيتم تشغيله في المرة القادمة.
فيما يلي رسم تخطيطي لحل رفيع المستوى:
إذا لم يكن لديك اشتراك Azure، أنشئ حسابا free قبل أن تبدأ.
المتطلبات الأساسية
- SQL Server. تستخدم قاعدة بيانات SQL Server كمخزن بيانات المصدر في هذا الدرس.
- قاعدة بيانات Azure SQL. تستخدم قاعدة بيانات في قاعدة بيانات Azure SQL كمخزن بيانات المصرف. إذا لم يكن لديك قاعدة بيانات SQL، راجع إنشاء قاعدة بيانات في قاعدة بيانات Azure SQL لخطوات إنشاء واحدة.
أنشئ جداول المصدر في قاعدة بيانات SQL Server الخاصة بك
افتح SQL Server Management Studio (SSMS) أو تعليمة Visual Studio برمجية، واتصل بقاعدة بيانات SQL Server الخاصة بك.
في Server Explorer (SSMS) أو في لوحة Connections (تعليمة Visual Studio برمجية)، انقر بزر الفأرة الأيمن على قاعدة البيانات واختر New Query.
تشغيل الأمر 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');
إنشاء جداول الوجهة في قاعدة بيانات Azure SQL الخاص بك
افتح SQL Server Management Studio (SSMS) أو تعليمة Visual Studio برمجية، واتصل بقاعدة بيانات SQL Server الخاصة بك.
في Server Explorer (SSMS) أو في لوحة Connections (تعليمة Visual Studio برمجية)، انقر بزر الفأرة الأيمن على قاعدة البيانات واختر New Query.
تشغيل الأمر 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 );
إنشاء جدول آخر في قاعدة بيانات Azure SQL لتخزين قيمة العلامة المائية العالية
شغل أمر SQL التالي مقابل قاعدة بياناتك في SQL لإنشاء جدول باسم
watermarktableلتخزين قيمة العلامة المائية:create table watermarktable ( TableName varchar(255), WatermarkValue datetime, );إدراج قيم العلامة المائية الأولية لكلا الجدولين المصدرين في جدول العلامة المائية.
INSERT INTO watermarktable VALUES ('customer_table','1/1/2010 12:00:00 AM'), ('project_table','1/1/2010 12:00:00 AM');
إنشاء إجراء مخزن في قاعدة بيانات Azure SQL
تشغيل الأمر التالي لإنشاء إجراء مخزن في قاعدة البيانات الخاصة بك. هذا الإجراء المخزن بتحديث قيمة العلامة المائية بعد تشغيل كل خط أنابيب.
CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS
BEGIN
UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime
WHERE [TableName] = @TableName
END
إنشاء أنواع بيانات وإجراءات مخزنة إضافية في قاعدة بيانات Azure SQL
تشغيل الاستعلام التالي لإنشاء اثنين من الإجراءات المخزنة ونوعي بيانات في قاعدة البيانات الخاصة بك. يتم استخدامها لدمج البيانات من جداول المصدر في جداول الوجهة.
من أجل جعل الرحلة سهلة لتبدأ، ونحن نستخدم مباشرة هذه الإجراءات المخزنة تمرير البيانات دلتا في عبر متغير الجدول ومن ثم دمج لهم في مخزن الوجهة. كن حذرا فإنه لا تتوقع عدد "كبير" من صفوف دلتا (أكثر من 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
Azure PowerShell
قم بتثبيت أحدث وحدات Azure PowerShell باتباع التعليمات في تثبيت وتكوين Azure PowerShell.
إنشاء مصدرًا للبيانات
حدد متغيراً لاسم مجموعة الموارد الذي ستستخدمه لاحقاً في أوامر PowerShell. انسخ نص الأمر التالي إلى PowerShell، وحدد اسما لمجموعة الموارد Azure بعلامات اقتباس مزدوجة، ثم شغل الأمر. مثال على ذلك
"adfrg".$resourceGroupName = "ADFTutorialResourceGroup";إذا كانت مجموعة الموارد موجودة بالفعل، فقد لا ترغب في الكتابة فوقها. عين قيمة مختلفة
$resourceGroupNameللمتغير وشغل الأمر مرة أخرى.حدد متغير لموقع بيانات المصنع.
$location = "East US"لإنشاء مجموعة موارد Azure، قم بتشغيل الأمر التالي:
New-AzResourceGroup $resourceGroupName $locationإذا كانت مجموعة الموارد موجودة بالفعل، فقد لا ترغب في الكتابة فوقها. عين قيمة مختلفة
$resourceGroupNameللمتغير وشغل الأمر مرة أخرى.حدد متغير لموقع بيانات المصنع.
هام
حدث اسم مصنع البيانات باسم عمومي فريد. مثال على ذلك هو ADFIncMultiCopyTutorialFactorySP1127.
$dataFactoryName = "ADFIncMultiCopyTutorialFactory";لإنشاء بيانات المصنع شغل التالي Set-AzDataFactoryV2 cmdlet:
Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location $location -Name $dataFactoryName
لاحظ النقاط التالية:
يجب أن يكون اسم مصنع البيانات مميزًا وعامًا. إذا استلمت الخطأ التالي، فغير الاسم وحاول مرة أخرى:
Set-AzDataFactoryV2 : HTTP Status Code: Conflict Error Code: DataFactoryNameInUse Error Message: The specified resource name 'ADFIncMultiCopyTutorialFactory' is already in use. Resource names must be globally unique.لإنشاء نسخ Data Factory، يجب أن يكون حساب المستخدم الذي تستخدمه لتسجيل الدخول إلى Azure عضوا في أدوار مساهم أو مالك، أو مسؤولا في اشتراك Azure.
للحصول على قائمة بالمناطق Azure التي يتوفر فيها Data Factory حاليا، اختر المناطق التي تهمك في الصفحة التالية، ثم قم بتوسيع Analytics لتحديد موقع Data Factory: المنتجات المتاحة حسب المنطقة. مخازن البيانات (تخزين Azure، SQL Database، مثيل SQL المدار، وهكذا) والحسابات (Azure HDInsight، إلخ) التي يستخدمها مصنع البيانات يمكن أن تكون في مناطق أخرى.
إنشاء وقت تشغيل تكامل مستضاف ذاتيا
في هذا القسم، تنشئ وقت تشغيل تكامل مستضاف ذاتيا وتربطه بجهاز محلي مع قاعدة بيانات SQL Server. وقت تشغيل التكامل المستضاف ذاتيا هو المكون الذي ينسخ البيانات من SQL Server على جهازك إلى قاعدة بيانات Azure SQL.
إنشاء متغير لاسم وقت تشغيل التكامل. استخدم اسمًا فريدًا، ودوّن ملاحظة خاصة به. يمكن استخدامه في وقت لاحق في هذا البرنامج التعليمي.
$integrationRuntimeName = "ADFTutorialIR"إنشاء وقت تشغيل تكامل الاستضافة الذاتية.
Set-AzDataFactoryV2IntegrationRuntime -Name $integrationRuntimeName -Type SelfHosted -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupNameفيما يلي ناتج العينة:
Name : <Integration Runtime name> Type : SelfHosted ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Description : Id : /subscriptions/<subscription ID>/resourceGroups/<ResourceGroupName>/providers/Microsoft.DataFactory/factories/<DataFactoryName>/integrationruntimes/ADFTutorialIRلاسترداد حالة وقت تشغيل التكامل الذي تم إنشاؤه، شغّل الأمر التالي. تأكد من تعيين قيمة خاصية State إلى NeedRegistration.
Get-AzDataFactoryV2IntegrationRuntime -name $integrationRuntimeName -ResourceGroupName $resourceGroupName -DataFactoryName $dataFactoryName -Statusفيما يلي ناتج العينة:
State : NeedRegistration Version : CreateTime : 9/24/2019 6:00:00 AM AutoUpdate : On ScheduledUpdateDate : UpdateDelayOffset : LocalTimeZoneOffset : InternalChannelEncryption : Capabilities : {} ServiceUrls : {eu.frontend.clouddatahub.net} Nodes : {} Links : {} Name : ADFTutorialIR Type : SelfHosted ResourceGroupName : <ResourceGroup name> DataFactoryName : <DataFactory name> Description : Id : /subscriptions/<subscription ID>/resourceGroups/<ResourceGroup name>/providers/Microsoft.DataFactory/factories/<DataFactory name>/integrationruntimes/<Integration Runtime name>لاسترجاع مفاتيح المصادقة المستخدمة لتسجيل وقت تشغيل التكامل المستضاف ذاتيا مع خدمة Azure Data Factory في السحابة، قم بتشغيل الأمر التالي:
Get-AzDataFactoryV2IntegrationRuntimeKey -Name $integrationRuntimeName -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName | ConvertTo-Jsonفيما يلي ناتج العينة:
{ "AuthKey1": "IR@0000000000-0000-0000-0000-000000000000@xy0@xy@xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx=", "AuthKey2": "IR@0000000000-0000-0000-0000-000000000000@xy0@xy@yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy=" }انسخ أحد المفاتيح (باستثناء علامات الاقتباس المزدوجة) المُستخدمة لتسجيل وقت تشغيل تكامل الاستضافة الذاتية الذي قمت بتثبيته على جهازك في الخطوات التالية.
ثبت أداة وقت تشغيل التكامل
إذا كان لديك بالفعل وقت تشغيل التكامل على جهازك، فقم بإلغاء تثبيته باستخدام إضافة برامج أو إزالتها.
Download وقت تشغيل التكامل المستضاف ذاتيا على جهاز Windows محلي. شغّل التثبيت.
في صفحة مرحبا بكم في Microsoft Integration Runtime Setup، اختر Next.
في صفحة اتفاقية ترخيص المستخدم النهائي، اقبل الشروط واتفاقية الترخيص، وحدد التالي.
في صفحة مجلد الوجهة، حدد التالي.
في صفحة جاهز للتثبيت Microsoft Integration Runtime، اختر Install.
في صفحة Completeded the Microsoft Integration Runtime Setup، اختر Finish.
في صفحة Register Integration Runtime (مستضاف ذاتيا)، الصق المفتاح الذي حفظته في القسم السابق، واختر Register.
في صفحة New Integration Runtime (المستضافة ذاتيا)، اختر Finish.
ستظهر الرسالة التالية عند تسجيل وقت تشغيل تكامل الاستضافة الذاتية بنجاح:
في صفحة Register Integration Runtime (مستضاف ذاتيا)، اختر Launch Configuration Manager.
عند توصيل العقدة بخدمة مجموعة النظراء، سترى الصفحة التالية:
الآن، اختبر الاتصال بقاعدة بيانات SQL Server الخاصة بك.
أ. في صفحة Configuration Manager، انتقل إلى تبويب Diagnostics.
ب. حدد SqlServer لنوع مصدر البيانات.
جـ. أدخل اسم الخادم.
د. أدخل اسم قاعدة البيانات.
هـ. حدد وضع المصادقة.
و. أدخل اسم المستخدم.
ز. أدخل كلمة المرور المرتبطة باسم المستخدم.
ح. اختر Test للتأكد من أن وقت تشغيل التكامل يمكنه الاتصال ب SQL Server. إذا لم ينجح الاتصال، فسترى علامة تحديد خضراء. إذا لم ينجح الاتصال، فسترى رسالة وجود خطأ. أصلح أي مشاكل، وتأكد من أن وقت تشغيل التكامل يمكن الاتصال ب SQL Server.
إشعار
دوّن القيم لنوع المصادقة والخادم وقاعدة البيانات والمستخدم وكلمة المرور. يمكنك استخدام ذلك لاحقًا في هذا البرنامج التعليمي.
إنشاء linked services
إنشاء خدمات مرتبطة في مصنع بيانات لربط مخازن بياناتك وحساب الخدمات إلى مصنع البيانات. في هذا القسم، تنشئ خدمات مرتبطة لقاعدة بيانات SQL Server الخاصة بك وقاعدة بياناتك في قاعدة بيانات Azure SQL.
إنشاء خدمة SQL Server المرتبطة
في هذه الخطوة، تربط قاعدة بيانات SQL Server بمصنع البيانات.
إنشاء ملف JSON المسمى SqlServerLinkedService.js في المجلد C:\ADFTutorials\IncCopyMultiTableTutorial (إنشاء المجلدات المحلية إذا لم تكن موجودة مسبقا) مع المحتوى التالي. اختر القسم الصحيح بناء على المصادقة التي تستخدمها للاتصال ب SQL Server.
هام
اختر القسم الصحيح بناء على المصادقة التي تستخدمها للاتصال ب SQL Server.
إذا كنت تستخدم SQL المصادقة، فانسخ تعريف JSON التالي:
{ "name":"SqlServerLinkedService", "properties":{ "annotations":[ ], "type":"SqlServer", "typeProperties":{ "connectionString":"integrated security=False;data source=<servername>;initial catalog=<database name>;user id=<username>;Password=<password>" }, "connectVia":{ "referenceName":"<integration runtime name>", "type":"IntegrationRuntimeReference" } } }إذا كنت تستخدم Windows authentication، انسخ تعريف JSON التالي:
{ "name":"SqlServerLinkedService", "properties":{ "annotations":[ ], "type":"SqlServer", "typeProperties":{ "connectionString":"integrated security=True;data source=<servername>;initial catalog=<database name>", "userName":"<username> or <domain>\\<username>", "password":{ "type":"SecureString", "value":"<password>" } }, "connectVia":{ "referenceName":"<integration runtime name>", "type":"IntegrationRuntimeReference" } } }هام
- اختر القسم الصحيح بناء على المصادقة التي تستخدمها للاتصال ب SQL Server.
- استبدل < اسم وقت تشغيل التكامل > باسم وقت تشغيل التكامل.
- استبدل <servername>، <databasename>، <username>، و<password> بقيم قاعدة البيانات SQL Server الخاصة بك قبل حفظ الملف.
- إذا كنت بحاجة إلى استخدام حرف مائل (
\) في حساب المستخدم أو اسم الخادم، فاستخدم حرف الهروب (\). مثال على ذلكmydomain\\myuser.
في PowerShell، قم بتشغيل cmdlet التالية للتبديل إلى المجلد C:\ADFTutorials\IncCopyMultiTableTutorial.
Set-Location 'C:\ADFTutorials\IncCopyMultiTableTutorial'شغل cmdlet Set-AzDataFactoryV2LinkedService لإنشاء الخدمة المرتبطة: AzureStorageLinkedService. في المثال التالي، يمكنك تمرير قيم ResourceGroupName واسم معلمات DataFactory:
Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SqlServerLinkedService" -File ".\SqlServerLinkedService.json"فيما يلي ناتج العينة:
LinkedServiceName : SqlServerLinkedService ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Properties : Microsoft.Azure.Management.DataFactory.Models.SqlServerLinkedService
إنشاء خدمة مرتبطة بقاعدة بيانات SQL
إنشاء ملف JSON المسمى AzureSQLDatabaseLinkedService.js في C:\ADFTutorials\IncCopyMultiTableTutorial المجلد مع المحتوى التالي. (أنشئ المجلد ADF إذا لم يكن موجودا بالفعل.) استبدل <servername>، <قاعدة البيانات>، <user name>، و <password> باسم قاعدة البيانات SQL Server الخاصة بك، واسم قاعدة البيانات، واسم المستخدم، وكلمة المرور قبل حفظ الملف.
{ "name":"AzureSQLDatabaseLinkedService", "properties":{ "annotations":[ ], "type":"AzureSqlDatabase", "typeProperties":{ "connectionString":"integrated security=False;encrypt=True;connection timeout=30;data source=<servername>.database.windows.net;initial catalog=<database name>;user id=<user name>;Password=<password>;" } } }في PowerShell، شغل cmdlet Set-AzDataFactoryV2LinkedService لإنشاء الخدمة المرتبطة: AzureSqlDatabaseLinkedService.
Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSQLDatabaseLinkedService" -File ".\AzureSQLDatabaseLinkedService.json"فيما يلي ناتج العينة:
LinkedServiceName : AzureSQLDatabaseLinkedService ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Properties : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDatabaseLinkedService
إنشاء datasets
في هذه الخطوة، يمكنك إنشاء مجموعات البيانات لتمثيل مصدر البيانات ووجهة البيانات والمكان لتخزين العلامة المائية.
قم بإنشاء مجموعة بيانات المصدر
أنشئ ملف JSON باسم SourceDataset.json في نفس الملف بالمحتويات التالية:
{ "name":"SourceDataset", "properties":{ "linkedServiceName":{ "referenceName":"SqlServerLinkedService", "type":"LinkedServiceReference" }, "annotations":[ ], "type":"SqlServerTable", "schema":[ ] } }يستخدم Copy activity في خط الأنابيب استعلام SQL لتحميل البيانات بدلا من تحميل الجدول بالكامل.
شغل Set-AzDataFactoryV2Dataset cmdlet لإنشاء مجموعة بيانات SourceDataset.
Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SourceDataset" -File ".\SourceDataset.json"إليك ناتج تشغيل عينة cmdlet:
DatasetName : SourceDataset ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Structure : Properties : Microsoft.Azure.Management.DataFactory.Models.SqlServerTableDataset
أنشئ مجموعة بيانات المورد
أنشئ ملف JSON باسم SinkDataset.json في نفس الملف بالمحتويات التالية. يتم تعيين عنصر tableName بواسطة خط أنابيب بشكل حيوي في وقت التشغيل. النشاط ForEach في خط الأنابيب تكرار خلال قائمة أسماء الجداول ويقوم بتمرير اسم الجدول إلى مجموعة البيانات هذه في كل تكرار.
{ "name":"SinkDataset", "properties":{ "linkedServiceName":{ "referenceName":"AzureSQLDatabaseLinkedService", "type":"LinkedServiceReference" }, "parameters":{ "SinkTableName":{ "type":"String" } }, "annotations":[ ], "type":"AzureSqlTable", "typeProperties":{ "tableName":{ "value":"@dataset().SinkTableName", "type":"Expression" } } } }شغل Set-AzDataFactoryV2Dataset cmdlet لإنشاء مجموعة بيانات SinkDataset.
Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SinkDataset" -File ".\SinkDataset.json"إليك ناتج تشغيل عينة cmdlet:
DatasetName : SinkDataset ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Structure : Properties : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset
أنشئ مجموعة بيانات لهذه العلامة المائية
في هذه الخطوة، يمكنك إنشاء مجموعة بيانات لتخزين قيمة علامة مائية عالية.
أنشئ ملف JSON باسم WatermarkDataset.json في نفس الملف بالمحتويات التالية:
{ "name": " WatermarkDataset ", "properties": { "type": "AzureSqlTable", "typeProperties": { "tableName": "watermarktable" }, "linkedServiceName": { "referenceName": "AzureSQLDatabaseLinkedService", "type": "LinkedServiceReference" } } }شغل Set-AzDataFactoryV2Dataset cmdlet لإنشاء مجموعة بيانات WatermarkDataset.
Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "WatermarkDataset" -File ".\WatermarkDataset.json"إليك ناتج تشغيل عينة cmdlet:
DatasetName : WatermarkDataset ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Structure : Properties : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset
إنشاء البنية الأساسية لبرنامج ربط العمليات التجارية
يأخذ خط أنابيب قائمة أسماء الجداول كمعلمة. النشاط ForEach يكرر من خلال قائمة أسماء الجداول وتنفيذ العمليات التالية:
استخدم نشاط البحث لاسترداد قيمة العلامة المائية القديمة (القيمة الأولية أو التي تم استخدامها في التكرار الأخير).
استخدم نشاط البحث لاسترداد قيمة العلامة المائية الجديدة (القيمة القصوى لعمود العلامة المائية في الجدول المصدر).
استخدم Copy activity لنسخ البيانات بين هاتين القيمتين للعلامة المائية من قاعدة البيانات المصدر إلى قاعدة بيانات الوجهة.
استخدم نشاط StoredProcedure لتحديث قيمة العلامة المائية القديمة لاستخدامها في الخطوة الأولى من التكرار التالي.
إنشاء البنية الأساسية لبرنامج ربط العمليات التجارية
أنشئ ملف JSON باسم ADFTutorialPipeline.json في مجلد C:\ADFGetStartedPSH بالمحتوى التالي:
{ "name":"IncrementalCopyPipeline", "properties":{ "activities":[ { "name":"IterateSQLTables", "type":"ForEach", "dependsOn":[ ], "userProperties":[ ], "typeProperties":{ "items":{ "value":"@pipeline().parameters.tableList", "type":"Expression" }, "isSequential":false, "activities":[ { "name":"LookupOldWaterMarkActivity", "type":"Lookup", "dependsOn":[ ], "policy":{ "timeout":"7.00:00:00", "retry":0, "retryIntervalInSeconds":30, "secureOutput":false, "secureInput":false }, "userProperties":[ ], "typeProperties":{ "source":{ "type":"AzureSqlSource", "sqlReaderQuery":{ "value":"select * from watermarktable where TableName = '@{item().TABLE_NAME}'", "type":"Expression" } }, "dataset":{ "referenceName":"WatermarkDataset", "type":"DatasetReference" } } }, { "name":"LookupNewWaterMarkActivity", "type":"Lookup", "dependsOn":[ ], "policy":{ "timeout":"7.00:00:00", "retry":0, "retryIntervalInSeconds":30, "secureOutput":false, "secureInput":false }, "userProperties":[ ], "typeProperties":{ "source":{ "type":"SqlServerSource", "sqlReaderQuery":{ "value":"select MAX(@{item().WaterMark_Column}) as NewWatermarkvalue from @{item().TABLE_NAME}", "type":"Expression" } }, "dataset":{ "referenceName":"SourceDataset", "type":"DatasetReference" }, "firstRowOnly":true } }, { "name":"IncrementalCopyActivity", "type":"Copy", "dependsOn":[ { "activity":"LookupOldWaterMarkActivity", "dependencyConditions":[ "Succeeded" ] }, { "activity":"LookupNewWaterMarkActivity", "dependencyConditions":[ "Succeeded" ] } ], "policy":{ "timeout":"7.00:00:00", "retry":0, "retryIntervalInSeconds":30, "secureOutput":false, "secureInput":false }, "userProperties":[ ], "typeProperties":{ "source":{ "type":"SqlServerSource", "sqlReaderQuery":{ "value":"select * from @{item().TABLE_NAME} where @{item().WaterMark_Column} > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and @{item().WaterMark_Column} <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'", "type":"Expression" } }, "sink":{ "type":"AzureSqlSink", "sqlWriterStoredProcedureName":{ "value":"@{item().StoredProcedureNameForMergeOperation}", "type":"Expression" }, "sqlWriterTableType":{ "value":"@{item().TableType}", "type":"Expression" }, "storedProcedureTableTypeParameterName":{ "value":"@{item().TABLE_NAME}", "type":"Expression" }, "disableMetricsCollection":false }, "enableStaging":false }, "inputs":[ { "referenceName":"SourceDataset", "type":"DatasetReference" } ], "outputs":[ { "referenceName":"SinkDataset", "type":"DatasetReference", "parameters":{ "SinkTableName":{ "value":"@{item().TABLE_NAME}", "type":"Expression" } } } ] }, { "name":"StoredProceduretoWriteWatermarkActivity", "type":"SqlServerStoredProcedure", "dependsOn":[ { "activity":"IncrementalCopyActivity", "dependencyConditions":[ "Succeeded" ] } ], "policy":{ "timeout":"7.00:00:00", "retry":0, "retryIntervalInSeconds":30, "secureOutput":false, "secureInput":false }, "userProperties":[ ], "typeProperties":{ "storedProcedureName":"[dbo].[usp_write_watermark]", "storedProcedureParameters":{ "LastModifiedtime":{ "value":{ "value":"@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}", "type":"Expression" }, "type":"DateTime" }, "TableName":{ "value":{ "value":"@{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}", "type":"Expression" }, "type":"String" } } }, "linkedServiceName":{ "referenceName":"AzureSQLDatabaseLinkedService", "type":"LinkedServiceReference" } } ] } } ], "parameters":{ "tableList":{ "type":"array" } }, "annotations":[ ] } }شغل Set-AzDataFactoryV2Pipeline cmdlet لإنشاء البنية الأساسية IncrementalCopyPipeline.
Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "IncrementalCopyPipeline" -File ".\IncrementalCopyPipeline.json"فيما يلي ناتج العينة:
PipelineName : IncrementalCopyPipeline ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Activities : {IterateSQLTables} Parameters : {[tableList, Microsoft.Azure.Management.DataFactory.Models.ParameterSpecification]}
قم بتشغيل البنية الأساسية
إنشاء ملف معلمة المسمى Parameters.js في نفس المجلد مع المحتوى التالي:
{ "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" } ] }شغل المسار IncrementalCopyPipeline باستخدام Cmdlet Invoke-AzDataFactoryV2Pipeline. استبدل placeholders بمجموعة الموارد الخاصة بك واسم بيانات المصنع.
$RunId = Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroup $resourceGroupName -dataFactoryName $dataFactoryName -ParameterFile ".\Parameters.json"
مراقبة المسار
سجل الدخول إلى بوابة Azure.
حدد جميع الخدمات، ابحث باستخدام الكلمة الرئيسية مصانع البيانات، وحدد مصانع البيانات.
ابحث عن مصنع البيانات في قائمة مصانع البيانات، وحدده لفتح صفحة مصنع البيانات.
في صفحة Data factory، اختر Open في بلاطة Open Azure Data Factory Studio لتشغيل Azure Data Factory في تبويب منفصل.
في الصفحة الرئيسية Azure Data Factory، اختر Monitor على الجانب الأيسر.
يمكنك أن ترى كل خطوط الأنابيب تعمل ووضعها. لاحظ أن في المثال التالي، يتم بنجاححالة تشغيل خط أنابيب . للتحقق من المعلمات التي تم تمريرها إلى خط الأنابيب، حدد الارتباط في العمود معلمات. إذا حدث خطأ، فسترى ارتباطاً في العمود خطأ.
عند تحديد الارتباط في العمود الإجراءات، سترى جميع الأنشطة التي يتم تشغيلها لخط الأنابيب.
للعودة إلى طريقة عرض "تشغيل خطوط الأنابيب"، حدد جميع تشغيلات خطوط الأنابيب.
مراجعة النتائج
في 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');
أعد تشغيل التدفق
الآن، أعد تشغيل خط الأنابيب بتنفيذ أمر PowerShell التالي:
$RunId = Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroup $resourceGroupname -dataFactoryName $dataFactoryName -ParameterFile ".\Parameters.json"مراقبة تشغيل "تدفقات" باتباع الإرشادات الموجودة في قسم مراقبة "تدفقات" . عندما تكون حالة خط الأنابيب قيد التقدم،سترى ارتباط إجراء آخر ضمن إجراءات لإلغاء تشغيل خط الأنابيب.
حدد تحديث لتحديث القائمة حتى ينجح تشغيل خط الأنابيب.
اختياريا، حدد الارتباط عرض تشغيل النشاط ضمن إجراءات لمشاهدة جميع عمليات تشغيل النشاط المقترنة بتشغيل خط الأنابيب هذا.
راجع النتائج التالية
في SQL Server Management Studio، قم بتشغيل الاستعلامات التالية على قاعدة البيانات المستهدفة للتحقق من أن البيانات المحدثة/الجديدة تم نسخها من جداول المصدر إلى جداول الوجهة.
استفسار
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: