Not
Bu sayfaya erişim yetkilendirme gerektiriyor. Oturum açmayı veya dizinleri değiştirmeyi deneyebilirsiniz.
Bu sayfaya erişim yetkilendirme gerektiriyor. Dizinleri değiştirmeyi deneyebilirsiniz.
ŞUNLARA UYGULANIR:
Azure Data Factory Azure Synapse Analytics
İpucu
Microsoft Fabric'daki
Bu öğreticide, SQL Server veritabanındaki birden çok tablodan delta verilerini Azure SQL Veritabanı yükleyen bir işlem hattına sahip bir Azure Data Factory oluşturacaksınız.
Bu öğreticide aşağıdaki adımları gerçekleştireceksiniz:
- Kaynak ve hedef veri depolarını hazırlayın.
- Veri fabrikası oluşturma.
- Kendi kendine barındırılan tümleştirme çalışma zamanı oluşturma.
- Entegrasyon çalışma zamanını yükleyin.
- Bağlı hizmet oluşturma.
- Kaynak, havuz ve eşik veri kümeleri oluşturun.
- Bir işlem hattı oluşturun, çalıştırın ve izleyin.
- Sonuçları inceleyin.
- Kaynak tablolarına veri ekleyin veya bu verileri güncelleştirin.
- İşlem hattını tekrar çalıştırın ve takip edin.
- Son sonuçları gözden geçirin.
Overview
Bu çözümü oluşturmak için önemli adımlar şunlardır:
Filigran sütununu seçin.
Kaynak veri deposundaki her tablo için bir sütun seçin. Bu sütun, her çalıştırma için yeni veya güncelleştirilmiş kayıtları tanımlayabilir. Normalde, satırlar oluşturulduğunda veya güncelleştirildiğinde seçilen bu sütundaki veriler (örneğin, last_modify_time veya kimlik) artmaya devam eder. Bu sütundaki en büyük değer eşik olarak kullanılır.
Filigran değerini depolamak için bir veri deposu hazırlayın.
Bu öğreticide, filigran değerini bir SQL veritabanında saklayacaksınız.
Aşağıdaki eylemler ile bir işlem hattı oluşturun:
İşlem hattına parametre olarak geçen kaynak tablosu adlarının bir listesi üzerinden yinelenen bir ForEach eylemi oluşturun. Her kaynak tablosu için, o tabloya yönelik delta yüklemesini gerçekleştirmek amacıyla aşağıdaki faaliyetleri başlatır.
İki arama etkinliği oluşturun. Son eşik değerini almak için ilk Arama etkinliğini kullanın. Yeni eşik değerini almak için ikinci Arama etkinliğini kullanın. Filigran değerleri Kopyalama etkinliğine geçirilir.
Kaynak veri deposundaki satırları, filigran sütununun değeri eski filigran değerinden büyük ve yeni filigran değerine eşit veya bundan küçük olacak şekilde kopyalamak için bir kopyalama etkinliği oluşturun. Ardından, delta verilerini kaynak veri deposundan yeni bir dosya olarak Azure Blob depolamaya kopyalar.
Sonraki seferde çalışan işlem hattı için eşik değerini güncelleştiren bir StoredProcedure etkinliği oluşturun.
Yüksek düzeyli çözüm diyagramı aşağıdaki gibidir:
Azure aboneliğiniz yoksa başlamadan önce bir free hesabı oluşturun.
Prerequisites
- SQL Server. Bu öğreticide kaynak veri deposu olarak bir SQL Server veritabanı kullanırsınız.
- Azure SQL Veritabanı. havuz veri deposu olarak Azure SQL Veritabanı'de bir veritabanı kullanırsınız. SQL veritabanınız yoksa, oluşturma adımları için bkz. Azure SQL Veritabanı'da veritabanı oluşturma.
SQL Server veritabanınızda kaynak tablolar oluşturma
SQL Server Management Studio (SSMS) veya Visual Studio Code açın ve SQL Server veritabanınıza bağlanın.
Server Gezgini'nde (SSMS) veya Connections bölmesinde (Visual Studio Code) veritabanına sağ tıklayın ve Yeni Sorgu'yi seçin.
customer_tableveproject_tableadlı tabloları oluşturmak için aşağıdaki SQL komutunu veritabanınızda çalıştırın: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 Veritabanı'nizde hedef tablolar oluşturma
SQL Server Management Studio (SSMS) veya Visual Studio Code açın ve SQL Server veritabanınıza bağlanın.
Server Gezgini'nde (SSMS) veya Connections bölmesinde (Visual Studio Code) veritabanına sağ tıklayın ve Yeni Sorgu'yi seçin.
customer_tableveproject_tableadlı tabloları oluşturmak için aşağıdaki SQL komutunu veritabanınızda çalıştırın:create table customer_table ( PersonID int, Name varchar(255), LastModifytime datetime ); create table project_table ( Project varchar(255), Creationtime datetime );
Yüksek filigran değerini depolamak için Azure SQL Veritabanı'de başka bir tablo oluşturma
Filigran değerini depolamak için adlı
watermarktablebir tablo oluşturmak için veritabanınızda aşağıdaki SQL komutunu çalıştırın:create table watermarktable ( TableName varchar(255), WatermarkValue datetime, );Her iki kaynak tablonun ilk eşik değerlerini eşik tablosuna ekleyin.
INSERT INTO watermarktable VALUES ('customer_table','1/1/2010 12:00:00 AM'), ('project_table','1/1/2010 12:00:00 AM');
Azure SQL Veritabanı'de bir saklı yordam oluşturun
Veritabanınızda bir saklı yordam oluşturmak için aşağıdaki komutu çalıştırın. Bu saklı yordam, her işlem hattı çalıştırılması sonrasında filigran değerini günceller.
CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS
BEGIN
UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime
WHERE [TableName] = @TableName
END
Azure SQL Veritabanı'de veri türleri ve ek saklı yordamlar oluşturma
Veritabanınızda iki saklı yordam ve iki veri türü oluşturmak için aşağıdaki sorguyu çalıştırın. Bunlar, kaynak tablodaki verileri hedef tablolarla birleştirmek için kullanılır.
Başlangıçtaki süreci kolaylaştırmak için, delta verilerini bir tablo değişkeni kullanarak içeri aktaran bu Saklı Yordamları doğrudan kullanırız ve ardından bunları hedef veri deposuna entegre ederiz. Tablo değişkeninde "büyük" sayıda delta satırının (100'den fazla) depolanmasını beklemiyorum, bu yüzden dikkatli olun.
Çok sayıda delta satırını hedef depoyla birleştirmeniz gerekiyorsa, kopyalama etkinliğini kullanarak tüm delta verilerini önce hedef depodaki geçici bir "hazırlama" tablosuna kopyalamanızı ve ardından bunları "hazırlama" tablosundan "son" tabloya birleştirmek için tablo değişkeni kullanmadan kendi saklı yordamınızı oluşturmanızı öneririz.
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'daki yönergeleri izleyerek en son Azure PowerShell modüllerini yükleyin.
Veri fabrikası oluşturma
Daha sonra PowerShell komutlarında kullanacağınız kaynak grubu adı için bir değişken tanımlayın. Aşağıdaki komut metnini PowerShell'e kopyalayın, çift tırnak içinde Azure kaynak grubu için bir ad belirtin ve komutunu çalıştırın.
"adfrg"bunun bir örneğidir.$resourceGroupName = "ADFTutorialResourceGroup";Kaynak grubu zaten varsa, bunun üzerine yazmak istemeyebilirsiniz.
$resourceGroupNamedeğişkenine farklı bir değer atayın ve komutu yeniden çalıştırın.Veri fabrikasının konumu için bir değişken tanımlayın.
$location = "East US"Azure kaynak grubunu oluşturmak için aşağıdaki komutu çalıştırın:
New-AzResourceGroup $resourceGroupName $locationKaynak grubu zaten varsa, bunun üzerine yazmak istemeyebilirsiniz.
$resourceGroupNamedeğişkenine farklı bir değer atayın ve komutu yeniden çalıştırın.Veri fabrikasının adı için bir değişken tanımlayın.
Önemli
Veri fabrikasının adını genel olarak benzersiz olacak şekilde güncelleştirin. ADFIncMultiCopyTutorialFactorySP1127 buna bir örnektir.
$dataFactoryName = "ADFIncMultiCopyTutorialFactory";Veri fabrikasını oluşturmak için aşağıdaki Set-AzDataFactoryV2 cmdlet'ini çalıştırın:
Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location $location -Name $dataFactoryName
Aşağıdaki noktaları not edin:
Veri fabrikasının adı genel olarak benzersiz olmalıdır. Aşağıdaki hata iletisini alırsanız adı değiştirip yeniden deneyin:
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 örnekleri oluşturmak için, Azure oturum açmak için kullandığınız kullanıcı hesabının katkıda bulunan veya sahip rollerinin bir üyesi ya da Azure aboneliğinin yöneticisi olması gerekir.
Data Factory'nin şu anda kullanılabilir olduğu Azure bölgelerin listesi için aşağıdaki sayfada ilginizi çekebilecek bölgeleri seçin ve ardından Analytics'yi genişleterek Data Factory: Products by region öğesini bulun. Veri fabrikası tarafından kullanılan veri depoları (Azure Depolama, SQL Veritabanı, SQL Yönetilen Örneği vb.) ve işlem (Azure HDInsight vb.) başka bölgelerde olabilir.
Kendi kendine barındırılan tümleştirme çalışma zamanı oluşturmak
Bu bölümde, kendi kendine barındırılan bir tümleştirme çalışma zamanı oluşturur ve bunu SQL Server veritabanını içeren şirket içi bir makineyle ilişkilendirirsiniz. Makinenizdeki SQL Server'dan Azure SQL Veritabanı'e veri kopyalayan bir bileşen olan yerel barındırılan tümleştirme çalışma zamanı.
Tümleştirme çalışma zamanının adı için bir değişken oluşturun. Benzersiz bir ad kullanın ve not edin. Daha sonra bu öğreticide kullanacaksınız.
$integrationRuntimeName = "ADFTutorialIR"Kendi kendine barındırılan tümleştirme çalışma zamanı oluşturma.
Set-AzDataFactoryV2IntegrationRuntime -Name $integrationRuntimeName -Type SelfHosted -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupNameÖrnek çıktı aşağıdaki gibidir:
Name : <Integration Runtime name> Type : SelfHosted ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Description : Id : /subscriptions/<subscription ID>/resourceGroups/<ResourceGroupName>/providers/Microsoft.DataFactory/factories/<DataFactoryName>/integrationruntimes/ADFTutorialIROluşturulan tümleştirme çalışma zamanının durumunu almak için aşağıdaki komutu çalıştırın. State özelliğinin değerinin NeedRegistration olarak ayarlandığını onaylayın.
Get-AzDataFactoryV2IntegrationRuntime -name $integrationRuntimeName -ResourceGroupName $resourceGroupName -DataFactoryName $dataFactoryName -StatusÖrnek çıktı aşağıdaki gibidir:
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>Kendi kendine barındırılan tümleştirme çalışma zamanını bulutta Azure Data Factory hizmetiyle kaydetmek için kullanılan kimlik doğrulama anahtarlarını almak üzere aşağıdaki komutu çalıştırın:
Get-AzDataFactoryV2IntegrationRuntimeKey -Name $integrationRuntimeName -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName | ConvertTo-JsonÖrnek çıktı aşağıdaki gibidir:
{ "AuthKey1": "IR@0000000000-0000-0000-0000-000000000000@xy0@xy@xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx=", "AuthKey2": "IR@0000000000-0000-0000-0000-000000000000@xy0@xy@yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy=" }Kendi kendine barındırılan tümleştirme çalışma zamanını makinenize yüklerken, kaydetmek için kullanılan anahtarlardan birini (çift tırnak işaretleri olmadan) kopyalayın.
Tümleştirme çalışma zamanı aracını yükleme
Makinenizde tümleştirme çalışma zamanı zaten varsa, Program Ekle veya Kaldır'ı kullanarak kaldırın.
İndir yerinde barındırılan entegrasyon çalışma zamanını yerel bir Windows makinesine. Yüklemeyi çalıştırın.
Microsoft Integration Runtime Kurulum Karşılama sayfasında İleri öğesini seçin.
End-User Lisans Sözleşmesi sayfasında koşulları ve lisans sözleşmesini kabul edin ve İleri'yi seçin.
Hedef Klasör sayfasında İleri'yi seçin.
Microsoft Integration Runtime yüklemek için
Ready sayfasında Install öğesini seçin.Microsoft Integration Runtime Kurulumu tamamlandı sayfasında Finish öğesini seçin.
Register Integration Runtime (Yerel) sayfasında, önceki bölümde kaydettiğiniz anahtarı yapıştırın ve Register öğesini seçin.
Yeni Integration Runtime (Şirket içinde barındırılan) Node sayfasında Finish öğesini seçin.
Yerel barındırılan tümleştirme çalışma zamanı başarıyla kaydedildiğinde aşağıdaki iletiyi görürsünüz:
Register Integration Runtime (Şirket içinde barındırılan) sayfasında Yapılandırma Yöneticisi'ı Başlat seçeneğini seçin.
Düğüm bulut hizmetine bağlandığında aşağıdaki sayfayı görürsünüz:
Şimdi SQL Server veritabanınıza bağlantıyı test edin.
a. Yapılandırma Yöneticisi sayfasında Diagnostics sekmesine gidin.
b. Veri kaynağı türü için SqlServer'ı seçin.
c. Sunucu adını girin.
d. Veritabanı adını girin.
e. Kimlik doğrulama modunu seçin.
f. Kullanıcı adını girin.
g. Kullanıcı adı için ilişkili parolayı girin.
h. Tümleştirme çalışma zamanının SQL Server bağlanabildiğini onaylamak için Test öğesini seçin. Bağlantı başarılı olursa yeşil bir onay işareti görürsünüz. Bağlantı başarılı olmazsa bir hata iletisi görürsünüz. Tüm sorunları düzeltin ve tümleştirme çalışma zamanının SQL Server bağlanadığından emin olun.
Note
Kimlik doğrulama türü, sunucu, veritabanı, kullanıcı ve parola değerlerini not edin. Bunları bu öğreticinin ilerleyen bölümlerinde kullanacaksınız.
Bağlı hizmetler oluşturma
Veri depolarınızı ve işlem hizmetlerinizi veri fabrikasına bağlamak için veri fabrikasında bağlı hizmetler oluşturursunuz. Bu bölümde, Azure SQL Veritabanı'da SQL Server veritabanınıza ve veritabanınıza bağlı hizmetler oluşturacaksınız.
SQL Server bağlantılı hizmet oluşturma
Bu adımda, SQL Server veritabanınızı veri fabrikasına bağlarsınız.
C:\ADFTutorials\IncCopyMultiTableTutorial klasöründe (henüz yoksa yerel klasörleri oluşturun) aşağıdaki içeriğe sahip SqlServerLinkedService.json adlı bir JSON dosyası oluşturun. SQL Server bağlanmak için kullandığınız kimlik doğrulamasına göre doğru bölümü seçin.
Önemli
SQL Server bağlanmak için kullandığınız kimlik doğrulamasına göre doğru bölümü seçin.
SQL kimlik doğrulaması kullanıyorsanız aşağıdaki JSON tanımını kopyalayın:
{ "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 kullanıyorsanız aşağıdaki JSON tanımını kopyalayın:
{ "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" } } }Önemli
- SQL Server bağlanmak için kullandığınız kimlik doğrulamasına göre doğru bölümü seçin.
- Tümleştirme çalışma zamanı adını< tümleştirme çalışma zamanınızın adıyla değiştirin>.
- replace <servername>, dosyayı kaydetmeden önce SQL Server veritabanınızın değerleriyle <databasename>, <username> ve <password>.
- Kullanıcı hesabınızda veya sunucu adında eğik çizgi karakteri (
\) kullanmanız gerekirse kaçış karakterini (\) kullanın.mydomain\\myuserbunun bir örneğidir.
PowerShell'de, C:\ADFTutorials\IncCopyMultiTableTutorial klasörüne geçmek için aşağıdaki cmdlet'i çalıştırın.
Set-Location 'C:\ADFTutorials\IncCopyMultiTableTutorial'AzureStorageLinkedService bağlı hizmetini oluşturmak için Set-AzDataFactoryV2LinkedService cmdlet'ini çalıştırın. Aşağıdaki örnekte, ResourceGroupName ve DataFactoryName parametrelerinin değerlerini geçirirsiniz:
Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SqlServerLinkedService" -File ".\SqlServerLinkedService.json"Örnek çıktı aşağıdaki gibidir:
LinkedServiceName : SqlServerLinkedService ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Properties : Microsoft.Azure.Management.DataFactory.Models.SqlServerLinkedService
SQL Veritabanı bağlı hizmetini oluştur
C:\ADFTutorials\IncCopyMultiTableTutorial klasöründe aşağıdaki içeriğe sahip AzureSQLDatabaseLinkedService.json adlı bir JSON dosyası oluşturun. (Henüz yoksa ADF klasörünü oluşturun.) <sunucu adı>, <SQL Server adı>, <kullanıcı adı> ve <parola> öğelerini, dosyayı kaydetmeden önce kendi SQL Server veritabanı adınız, veritabanınızın adı, kullanıcı adınız ve parolanız ile değiştirin.
{ "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'de Set-AzDataFactoryV2LinkedService cmdlet'ini çalıştırarak AzureSQLDatabaseLinkedService bağlı hizmetini oluşturun.
Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSQLDatabaseLinkedService" -File ".\AzureSQLDatabaseLinkedService.json"Örnek çıktı aşağıdaki gibidir:
LinkedServiceName : AzureSQLDatabaseLinkedService ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Properties : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDatabaseLinkedService
Veri kümeleri oluşturma
Bu adımda veri kaynağı, veri hedefi ve eşiğin depolanacağı yeri temsil eden veri kümeleri oluşturacaksınız.
Kaynak veri kümesi oluşturma
Aynı klasörde aşağıdaki içeriğe sahip SourceDataset.json adlı bir JSON dosyası oluşturun:
{ "name":"SourceDataset", "properties":{ "linkedServiceName":{ "referenceName":"SqlServerLinkedService", "type":"LinkedServiceReference" }, "annotations":[ ], "type":"SqlServerTable", "schema":[ ] } }İşlem hattındaki Kopyalama etkinliği, tablonun tamamını yüklemek yerine verileri yüklemek için bir SQL sorgusu kullanır.
SourceDataset veri kümesini oluşturmak için Set-AzDataFactoryV2Dataset cmdlet'ini çalıştırın.
Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SourceDataset" -File ".\SourceDataset.json"Cmdlet’in örnek çıktısı aşağıdaki gibidir:
DatasetName : SourceDataset ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Structure : Properties : Microsoft.Azure.Management.DataFactory.Models.SqlServerTableDataset
Lavabo veri kümesi oluşturma
Aşağıdaki içeriğe sahip aynı klasörde SinkDataset.json adlı bir JSON dosyası oluşturun. tableName öğesi, işlem hattı tarafından çalışma zamanında dinamik olarak ayarlanır. İşlem hattındaki ForEach etkinliği, tablo adlarının bir listesi üzerinden yinelenir ve her yinelemede tablo adını bu veri kümesine geçirir.
{ "name":"SinkDataset", "properties":{ "linkedServiceName":{ "referenceName":"AzureSQLDatabaseLinkedService", "type":"LinkedServiceReference" }, "parameters":{ "SinkTableName":{ "type":"String" } }, "annotations":[ ], "type":"AzureSqlTable", "typeProperties":{ "tableName":{ "value":"@dataset().SinkTableName", "type":"Expression" } } } }SinkDataset veri kümesini oluşturmak için Set-AzDataFactoryV2Dataset cmdlet'ini çalıştırın.
Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SinkDataset" -File ".\SinkDataset.json"Cmdlet’in örnek çıktısı aşağıdaki gibidir:
DatasetName : SinkDataset ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Structure : Properties : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset
Eşik için veri kümesi oluşturma
Bu adımda üst eşik değerini depolamak için bir veri kümesi oluşturacaksınız.
Aynı klasörde aşağıdaki içeriğe sahip WatermarkDataset.json adlı bir JSON dosyası oluşturun:
{ "name": " WatermarkDataset ", "properties": { "type": "AzureSqlTable", "typeProperties": { "tableName": "watermarktable" }, "linkedServiceName": { "referenceName": "AzureSQLDatabaseLinkedService", "type": "LinkedServiceReference" } } }WatermarkDataset veri kümesini oluşturmak için Set-AzDataFactoryV2Dataset cmdlet'ini çalıştırın.
Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "WatermarkDataset" -File ".\WatermarkDataset.json"Cmdlet’in örnek çıktısı aşağıdaki gibidir:
DatasetName : WatermarkDataset ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Structure : Properties : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset
Boru hattı oluştur
Bu işlem hattı parametre olarak tablo adları listesini alır. ForEach etkinliği, tablo adları listesinde yinelenir ve aşağıdaki işlemleri gerçekleştirir:
Eski filigran değerini (ilk değer veya son yinelemede kullanılan değer) almak için Arama etkinliğini kullanın.
Yeni filigran değerini (kaynak tablodaki filigran sütununun en büyük değeri) almak için Arama etkinliğini kullanın.
Bu iki filigran değeri arasındaki verileri kaynak veritabanından hedef veritabanına kopyalamak için Kopyalama etkinliği kullanın.
Bir sonraki yinelemenin ilk adımında kullanılacak eski filigran değerini güncelleştirmek için StoredProcedure etkinliğini kullanın.
İşlem hattını oluştur
Aynı klasörde aşağıdaki içeriğe sahip IncrementalCopyPipeline.json adlı bir JSON dosyası oluşturun:
{ "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":[ ] } }IncrementalCopyPipeline işlem hattını oluşturmak için Set-AzDataFactoryV2Pipeline cmdlet'ini çalıştırın.
Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "IncrementalCopyPipeline" -File ".\IncrementalCopyPipeline.json"Örnek çıktı aşağıdaki gibidir:
PipelineName : IncrementalCopyPipeline ResourceGroupName : <ResourceGroupName> DataFactoryName : <DataFactoryName> Activities : {IterateSQLTables} Parameters : {[tableList, Microsoft.Azure.Management.DataFactory.Models.ParameterSpecification]}
İşlem hattını çalıştırma
Aynı klasörde aşağıdaki içeriğe sahip Parameters.json adlı bir parametre dosyası oluşturun:
{ "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" } ] }Invoke-AzDataFactoryV2Pipeline cmdlet'ini kullanarak IncrementalCopyPipeline işlem hattını çalıştırın. Yer tutucuları kendi kaynak grubu ve veri fabrikası adınızla değiştirin.
$RunId = Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroup $resourceGroupName -dataFactoryName $dataFactoryName -ParameterFile ".\Parameters.json"
İşlem hattını izleme
Azure portalında oturum açın.
Tüm hizmetler'i seçin, Veri fabrikaları anahtar sözcüğüyle arama yapın ve Veri fabrikaları'nı seçin.
Veri fabrikaları listesinde veri fabrikanızı arayın ve Data factory sayfasını açmak için bu fabrikayı seçin.
Data factory sayfasında, Azure Data Factory Studio'yu Aç kutucuğundan Aç seçeneğini seçerek Azure Data Factory'yi ayrı bir sekmede başlatın.
Azure Data Factory giriş sayfasında sol taraftaki Monitor seçin.
Ekran görüntüsü Azure Data Factory'nin giriş sayfasını gösteriyor. Tüm işlem hattı çalıştırmalarını ve bunların durumunu görebilirsiniz. Aşağıdaki örnekte işlem hattı çalıştırmasının durumunun Başarılı olarak belirtildiğini görebilirsiniz. İşlem hattına geçirilen parametreleri denetlemek için Parametreler sütunundaki bağlantıyı seçin. Bir hata oluştuysa, Hata sütununda bir bağlantı görürsünüz.
Eylemler sütununda bağlantıyı seçtiğinizde işlem hattı için tüm aktivite yürütmelerini görürsünüz.
İşlem Hattı Çalıştırmaları görünümüne dönmek için Tüm İşlem Hattı Çalıştırmaları'nı seçin.
Sonuçları gözden geçirin
SQL Server Management Studio'da, verilerin kaynak tablolardan hedef tablolara kopyalandığını doğrulamak için hedef SQL veritabanında aşağıdaki sorguları çalıştırın:
Query
select * from customer_table
Output
===========================================
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
Query
select * from project_table
Output
===================================
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
Query
select * from watermarktable
Output
======================================
TableName WatermarkValue
======================================
customer_table 2017-09-05 08:06:00.000
project_table 2017-03-04 05:16:00.000
Her iki tablonun da eşik değerlerinin güncelleştirildiğine dikkat edin.
Kaynak tablolara daha fazla veri ekleme
customer_table'da var olan bir satırı güncelleştirmek için kaynak SQL Server veritabanında aşağıdaki sorguyu çalıştırın. Project_table içine yeni bir satır ekleyin.
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');
İşlem hattını yeniden çalıştırma
Şimdi aşağıdaki PowerShell komutunu yürüterek işlem hattını yeniden çalıştırın:
$RunId = Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroup $resourceGroupname -dataFactoryName $dataFactoryName -ParameterFile ".\Parameters.json"İşlem hattını izleme bölümündeki yönergeleri izleyerek işlem hattı çalıştırmalarını izleyin . İşlem hattı durumu Devam Ediyor olduğunda, İşlem hattı çalıştırmasını iptal etmek için Eylemler altında başka bir eylem bağlantısı görürsünüz.
İşlem hattı çalıştırması başarılı olana kadar listeyi yenilemek için Yenile'yi seçin.
İsteğe bağlı olarak, bu işlem hattı çalıştırmasıyla ilişkili tüm etkinlik çalıştırmalarını görmek için Eylemler altındaki Etkinlik Çalıştırmalarını Görüntüle bağlantısını seçin.
Son sonuçları gözden geçirme
SQL Server Management Studio'da, güncelleştirilmiş/yeni verilerin kaynak tablolardan hedef tablolara kopyalandığını doğrulamak için hedef veritabanında aşağıdaki sorguları çalıştırın.
Query
select * from customer_table
Output
===========================================
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
3 numaraya ait PersonID için yeni Name ve LastModifytime değerlerine dikkat edin.
Query
select * from project_table
Output
===================================
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
NewProject girişinin project_table tablosuna eklendiğine dikkat edin.
Query
select * from watermarktable
Output
======================================
TableName WatermarkValue
======================================
customer_table 2017-09-08 00:00:00.000
project_table 2017-10-01 00:00:00.000
Her iki tablonun da eşik değerlerinin güncelleştirildiğine dikkat edin.
İlgili içerik
Bu öğreticide aşağıdaki adımları gerçekleştirdiniz:
- Kaynak ve hedef veri depolarını hazırlayın.
- Veri fabrikası oluşturma.
- Kendi kendine barındırılan tümleştirme çalışma zamanı (IR) oluşturun.
- Entegrasyon çalışma zamanını yükleyin.
- Bağlı hizmet oluşturma.
- Kaynak, havuz ve eşik veri kümeleri oluşturun.
- Bir işlem hattı oluşturun, çalıştırın ve izleyin.
- Sonuçları inceleyin.
- Kaynak tablolarına veri ekleyin veya bu verileri güncelleştirin.
- İşlem hattını tekrar çalıştırın ve takip edin.
- Son sonuçları gözden geçirin.
Azure bir Spark kümesi kullanarak verileri dönüştürme hakkında bilgi edinmek için aşağıdaki öğreticiye ilerleyin: