PowerShell kullanarak verileri Azure SQL Veritabanı'den Azure Blob depolamaya artımlı olarak yükleme

ŞUNLARA UYGULANIR: Azure Data Factory Azure Synapse Analytics

İpucu

Microsoft Fabric'daki Data Factory, daha basit bir mimariye, yerleşik yapay zekaya ve yeni özelliklere sahip yeni nesil Azure Data Factory. Veri tümleştirmeyi yeni kullanmaya başladıysanız Fabric Data Factory ile başlayın. Mevcut ADF iş yükleri veri bilimi, gerçek zamanlı analiz ve raporlama genelinde yeni özelliklere erişmek için Fabric yükseltebilir.

Bu öğreticide Azure Data Factory kullanarak Azure SQL Veritabanı'daki bir tablodan Azure Blob depolamaya delta verileri yükleyen bir işlem hattı oluşturacaksınız.

Bu öğreticide aşağıdaki adımları gerçekleştireceksiniz:

  • Filigran değerini depolamak için veri deposunu hazırlayın.
  • Veri fabrikası oluşturma.
  • Bağlı hizmet oluşturma.
  • Kaynak, havuz ve eşik veri kümeleri oluşturun.
  • Bir işlem hattı oluşturun.
  • Pipeline'ı çalıştır.
  • İşlem hattı çalışmasını izleyin.

Genel bakış

Yüksek düzeyli çözüm diyagramı aşağıdaki gibidir:

Artımlı olarak veri yükleme

Bu çözümü oluşturmak için önemli adımlar şunlardır:

  1. Filigran sütununu seçin. Kaynak veri deposunda her çalıştırma için yeni veya güncelleştirilmiş kayıtları dilimlemek için kullanılabilen bir sütun seçin. 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.

  2. Filigran değerini depolamak için veri deposunu hazırlayın.
    Bu öğreticide, filigran değerini bir SQL veritabanında depolayacaksınız.

  3. Aşağıdaki iş akışı ile bir işlem hattı oluşturun:

    Bu çözümdeki işlem hattı aşağıdaki etkinlikleri içerir:

    • İki Arama etkinliği oluşturun. Son işaret değerini almak için ilk Arama etkinliğini kullanın. Yeni watermark değerini almak için ikinci Lookup 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 veriler kaynak veri deposundan Blob depolama alanına yeni bir dosya olarak kopyalanır.
    • Sonraki seferde çalışan işlem hattı için eşik değerini güncelleştiren bir StoredProcedure etkinliği oluşturun.

Azure aboneliğiniz yoksa başlamadan önce bir free hesabı oluşturun.

Önkoşullar

Not

Azure ile etkileşime geçmek için Azure Az PowerShell modülünü kullanmanızı öneririz. Başlamak için bkz. Azure PowerShell yükleme. Az PowerShell modülüne nasıl geçiş yapılacağını öğrenmek için bkz. AzureRM'den Az Azure PowerShell dağıtma.

  • Azure SQL Veritabanı. Veritabanını kaynak veri deposu olarak kullanabilirsiniz. Azure SQL Veritabanı'da veritabanınız yoksa, oluşturma adımları için bkz. Azure SQL Veritabanı'da veritabanı oluşturma.
  • Azure Depolama. Blob depolamayı havuz veri deposu olarak kullanabilirsiniz. Depolama hesabınız yoksa, oluşturma adımları için bkz. Depolama hesabı oluşturma. adftutorial adlı bir kapsayıcı oluşturun.
  • Azure PowerShell. yükle ve Azure PowerShell yapılandırma başlığındaki yönergeleri izleyin.

SQL veritabanınızda bir veri kaynağı tablosu oluşturma

  1. SQL Server Management Studio'i açın. Sunucu Gezgini’nde veritabanına sağ tıklayın ve Yeni Sorgu’yu seçin.

  2. SQL veritabanınızda aşağıdaki SQL komutunu çalıştırarak veri kaynağı deponuz olarak data_source_table adlı bir tablo oluşturun:

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

    Bu öğreticide, filigran sütunu olarak LastModifytime kullanılır. Veri kaynağı deposundaki veriler aşağıdaki tabloda gösterilmiştir:

    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

Üst eşik değerini depolamak için SQL veritabanınızda başka bir tablo oluşturma

  1. SQL veritabanınızda aşağıdaki SQL komutunu çalıştırarak eşik değerini depolamak için watermarktable adlı bir tablo oluşturun:

    create table watermarktable
(

TableName varchar(255),
WatermarkValue datetime,
);

  2. Üst eşiğin varsayılan değerini kaynak veri deposunun tablo adıyla ayarlayın. Bu öğreticide tablo adı data_source_table şeklindedir.

    INSERT INTO watermarktable
VALUES ('data_source_table','1/1/2010 12:00:00 AM')

  3. watermarktable tablosundaki verileri gözden geçirin.

    Select * from watermarktable

    Çıktı:

    TableName  | WatermarkValue
----------  | --------------
data_source_table | 2010-01-01 00:00:00.000

SQL veritabanınızda bir saklı yordam oluşturma

SQL veritabanınızda bir saklı yordam oluşturmak için aşağıdaki komutu çalıştırın:

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

BEGIN

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

END

Veri fabrikası oluşturma

  1. 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, üzerine yazılmasını istemeyebilirsiniz. $resourceGroupName değişkenine farklı bir değer atayın ve komutu yeniden çalıştırın.

  2. Veri fabrikasının konumu için bir değişken tanımlayın.

    $location = "East US"

  3. Azure kaynak grubunu oluşturmak için aşağıdaki komutu çalıştırın:

    New-AzResourceGroup $resourceGroupName $location

    Kaynak grubu zaten varsa, üzerine yazılmasını istemeyebilirsiniz. $resourceGroupName değişkenine farklı bir değer atayın ve komutu yeniden çalıştırın.

  4. 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. Örnek olarak ADFTutorialFactorySP1127 olabilir.

    $dataFactoryName = "ADFIncCopyTutorialFactory";

  5. Veri fabrikasını oluşturmak için aşağıdaki Set-AzDataFactoryV2 cmdlet'ini çalıştırın:

    Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location "East US" -Name $dataFactoryName

Aaşağıdaki noktaları unutmayın:

  • 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:

    The specified Data Factory name 'ADFv2QuickStartDataFactory' is already in use. Data Factory 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ı (Depolama, SQL Veritabanı, Azure SQL Yönetilen Örneği vb.) ve işlem (Azure HDInsight vb.) başka bölgelerde olabilir.

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, depolama hesabınıza ve SQL Veritabanınıza bağlı hizmetler oluşturacaksınız.

Bağlı bir depolama hizmeti oluşturun

  1. C:\ADF klasöründe aşağıdaki içerikle AzureStorageLinkedService.json adlı bir JSON dosyası oluşturun. (Henüz yoksa ADF klasörünü oluşturun.) Dosyayı kaydetmeden önce, <accountName> ve <accountKey> değerlerini depolama hesabınızın adı ve anahtarıyla değiştirin.

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

  2. PowerShell’de ADF klasörüne geçin.

  3. 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 "AzureStorageLinkedService" -File ".\AzureStorageLinkedService.json"

    Örnek çıktı aşağıdaki gibidir:

    LinkedServiceName : AzureStorageLinkedService
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureStorageLinkedService

SQL Veritabanı ilişkili hizmet oluşturma

  1. C:\ADF klasöründe aşağıdaki içerikle AzureSQLDatabaseLinkedService.json adlı bir JSON dosyası oluşturun. (Henüz yoksa ADF klasörünü oluşturun.) Dosyayı kaydetmeden önce sunucunuzun-adını< ve >veritabanınızın-adını< sunucunuzun ve veritabanınızın adıyla değiştirin>. Azure SQL Server'ınızı, veri fabrikanızın yönetilen kimliğine erişim vermek için yapılandırmanız da gerekir.

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

  2. PowerShell’de ADF klasörüne geçin.

  3. AzureSQLDatabaseLinkedService bağlı hizmetini oluşturmak için Set-AzDataFactoryV2LinkedService cmdlet'ini çalıştırın.

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

    Örnek çıktı aşağıdaki gibidir:

    LinkedServiceName : AzureSQLDatabaseLinkedService
ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDatabaseLinkedService
ProvisioningState :

Veri kümeleri oluşturma

Bu adımda, kaynak ve havuz verilerini temsil eden veri kümeleri oluşturacaksınız.

Kaynak veri kümesi oluşturma

  1. Aşağıdaki içeriğe sahip klasörde SourceDataset.json adlı bir JSON dosyası oluşturun:

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

    Bu öğreticide, tablo adı olarak data_source_table'ı kullanırsınız. Adı farklı olan bir tablo kullanıyorsanız değiştirin.

  2. 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 : ADF
DataFactoryName   : incrementalloadingADF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

Havuz veri kümesi oluşturma

  1. Aşağıdaki içeriğe sahip klasörde SinkDataset.json adlı bir JSON dosyası oluşturun:

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

    Önemli

    Bu kod parçacığı, blob depolama alanınızda adlı adftutorial bir blob kapsayıcınız olduğunu varsayar. Henüz yoksa kapsayıcıyı oluşturun veya var olan bir kapsayıcının adına ayarlayın. incrementalcopy çıktı klasörü kapsayıcıda mevcut değilse otomatik olarak oluşturulur. Bu öğreticide dosya adı @CONCAT('Incremental-', pipeline().RunId, '.txt') ifadesi kullanılarak dinamik olarak oluşturulur.

  2. 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 : ADF
DataFactoryName   : incrementalloadingADF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureBlobDataset

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.

  1. Aşağıdaki içeriğe sahip klasörde WatermarkDataset.json adlı bir JSON dosyası oluşturun:

    {
    "name": " WatermarkDataset ",
    "properties": {
        "type": "AzureSqlTable",
        "typeProperties": {
            "tableName": "watermarktable"
        },
        "linkedServiceName": {
            "referenceName": "AzureSQLDatabaseLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

  2. 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 : ADF
DataFactoryName   : incrementalloadingADF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

Bir işlem hattı oluştur

Bu öğreticide, tek bir işlem hattında birbirine bağlanmış iki Arama etkinliği, bir Kopyalama etkinliği ve bir Saklı Yordam etkinliği ile bir işlem hattı oluşturursunuz.

  1. Aynı klasörde aşağıdaki içerikle IncrementalCopyPipeline.json adlı bir JSON dosyası oluşturun:

    {
    "name": "IncrementalCopyPipeline",
    "properties": {
        "activities": [
            {
                "name": "LookupOldWaterMarkActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                    "type": "SqlSource",
                    "sqlReaderQuery": "select * from watermarktable"
                    },

                    "dataset": {
                    "referenceName": "WatermarkDataset",
                    "type": "DatasetReference"
                    }
                }
            },
            {
                "name": "LookupNewWaterMarkActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select MAX(LastModifytime) as NewWatermarkvalue from data_source_table"
                    },

                    "dataset": {
                    "referenceName": "SourceDataset",
                    "type": "DatasetReference"
                    }
                }
            },

            {
                "name": "IncrementalCopyActivity",
                "type": "Copy",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select * from data_source_table where LastModifytime > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and LastModifytime <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'"
                    },
                    "sink": {
                        "type": "BlobSink"
                    }
                },
                "dependsOn": [
                    {
                        "activity": "LookupNewWaterMarkActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    },
                    {
                        "activity": "LookupOldWaterMarkActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],

                "inputs": [
                    {
                        "referenceName": "SourceDataset",
                        "type": "DatasetReference"
                    }
                ],
                "outputs": [
                    {
                        "referenceName": "SinkDataset",
                        "type": "DatasetReference"
                    }
                ]
            },

            {
                "name": "StoredProceduretoWriteWatermarkActivity",
                "type": "SqlServerStoredProcedure",
                "typeProperties": {

                    "storedProcedureName": "usp_write_watermark",
                    "storedProcedureParameters": {
                        "LastModifiedtime": {"value": "@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}", "type": "datetime" },
                        "TableName":  { "value":"@{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}", "type":"String"}
                    }
                },

                "linkedServiceName": {
                    "referenceName": "AzureSQLDatabaseLinkedService",
                    "type": "LinkedServiceReference"
                },

                "dependsOn": [
                    {
                        "activity": "IncrementalCopyActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ]
            }
        ]

    }
}

  2. 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 : ADF
 DataFactoryName   : incrementalloadingADF
 Activities        : {LookupOldWaterMarkActivity, LookupNewWaterMarkActivity, IncrementalCopyActivity, StoredProceduretoWriteWatermarkActivity}
 Parameters        :

İşlem hattını çalıştırma

  1. 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" -ResourceGroupName $resourceGroupName -dataFactoryName $dataFactoryName

  2. Tüm etkinliklerin başarıyla çalıştığını görene kadar Get-AzDataFactoryV2ActivityRun cmdlet'ini çalıştırarak işlem hattının durumunu denetleyin. Yer tutucuları RunStartedAfter ve RunStartedBefore parametresi için uygun bulduğunuz süreyle değiştirin. Bu öğreticide -RunStartedAfter "2017/09/14" ve -RunStartedBefore "2017/09/15" kullanılır.

    Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $RunId -RunStartedAfter "<start time>" -RunStartedBefore "<end time>"

    Örnek çıktı aşağıdaki gibidir:

    ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupNewWaterMarkActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {NewWatermarkvalue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:42:42 AM
ActivityRunEnd    : 9/14/2017 7:42:50 AM
DurationInMs      : 7777
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupOldWaterMarkActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {TableName, WatermarkValue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:42:42 AM
ActivityRunEnd    : 9/14/2017 7:43:07 AM
DurationInMs      : 25437
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : IncrementalCopyActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {source, sink}
Output            : {dataRead, dataWritten, rowsCopied, copyDuration...}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:43:10 AM
ActivityRunEnd    : 9/14/2017 7:43:29 AM
DurationInMs      : 19769
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : StoredProceduretoWriteWatermarkActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {storedProcedureName, storedProcedureParameters}
Output            : {}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:43:32 AM
ActivityRunEnd    : 9/14/2017 7:43:47 AM
DurationInMs      : 14467
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

Sonuçları gözden geçirin

  1. Blob depolama (havuz deposu) alanında verilerin SinkDataset içinde tanımlanan dosyaya kopyalandığını görürsünüz. Geçerli öğreticide dosya adı Incremental- d4bf3ce2-5d60-43f3-9318-923155f61037.txt şeklindedir. Dosyayı açtığınızda, dosyada SQL veritabanındaki verilerle aynı kayıtları görebilirsiniz.

    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

  2. watermarktable içindeki en son değeri denetleyin. Eşik değerinin güncelleştirildiğini görürsünüz.

    Select * from watermarktable

    Örnek çıktı aşağıdaki gibidir:

    TabloAdı WatermarkValue
    veri_kaynağı_tablosu 2017-09-05 8:06:00.000

Delta veri yüklemeyi doğrulamak için veri kaynağı deposuna veri ekleme

  1. Yeni verileri SQL veritabanına (veri kaynağı deposu) ekleyin.

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

    SQL veritabanında güncelleştirilmiş veriler:

    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

  2. Invoke-AzDataFactoryV2Pipeline cmdlet'ini kullanarak IncrementalCopyPipeline işlem hattını yeniden çalıştırın. Yer tutucuları kendi kaynak grubu ve veri fabrikası adınızla değiştirin.

    $RunId = Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroupName $resourceGroupName -dataFactoryName $dataFactoryName

  3. Tüm etkinliklerin başarıyla çalıştığını görene kadar Get-AzDataFactoryV2ActivityRun cmdlet'ini çalıştırarak işlem hattının durumunu denetleyin. Yer tutucuları RunStartedAfter ve RunStartedBefore parametresi için uygun bulduğunuz süreyle değiştirin. Bu öğreticide -RunStartedAfter "2017/09/14" ve -RunStartedBefore "2017/09/15" kullanılır.

    Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $RunId -RunStartedAfter "<start time>" -RunStartedBefore "<end time>"

    Örnek çıktı aşağıdaki gibidir:

    ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupNewWaterMarkActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {NewWatermarkvalue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:52:26 AM
ActivityRunEnd    : 9/14/2017 8:52:58 AM
DurationInMs      : 31758
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupOldWaterMarkActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {TableName, WatermarkValue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:52:26 AM
ActivityRunEnd    : 9/14/2017 8:52:52 AM
DurationInMs      : 25497
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : IncrementalCopyActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {source, sink}
Output            : {dataRead, dataWritten, rowsCopied, copyDuration...}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:53:00 AM
ActivityRunEnd    : 9/14/2017 8:53:20 AM
DurationInMs      : 20194
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : StoredProceduretoWriteWatermarkActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {storedProcedureName, storedProcedureParameters}
Output            : {}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:53:23 AM
ActivityRunEnd    : 9/14/2017 8:53:41 AM
DurationInMs      : 18502
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

  4. Blob depolama alanında başka bir dosyanın oluşturulduğunu görürsünüz. Bu öğreticide yeni dosya adı Incremental-2fc90ab8-d42c-4583-aa64-755dba9925d7.txt şeklindedir. Bu dosyayı açın, içinde 2 satır kaydı göreceksiniz.

  5. watermarktable içindeki en son değeri denetleyin. Eşik değerinin tekrar güncelleştirildiğini görürsünüz.

    Select * from watermarktable

    örnek çıktı:

    TabloAdı WatermarkValue
    veri_kaynağı_tablosu 2017-09-07 09:01:00.000

İlgili içerik

Bu öğreticide aşağıdaki adımları gerçekleştirdiniz:

  • Filigran değerini depolamak için veri deposunu hazırlayın.
  • Veri fabrikası oluşturma.
  • Bağlı hizmet oluşturma.
  • Kaynak, havuz ve eşik veri kümeleri oluşturun.
  • Bir işlem hattı oluşturun.
  • Pipeline'ı çalıştır.
  • İşlem hattı çalışmasını izleyin.

Bu öğreticide işlem hattı verileri Azure SQL Veritabanı'daki tek bir tablodan Blob depolamaya kopyaladı. SQL Server veritabanındaki birden çok tablodan SQL Veritabanı'na veri kopyalamayı öğrenmek için aşağıdaki öğreticiye ilerleyin.

SQL Server'daki birden fazla tablodan Azure SQL Veritabanına artırımlı veri yükleme

  • Last updated on