Azure Data Factory ve Synapse Analytics'te İşlem Hattı etkinliğini yürütme

  • Makale

UYGULANANLAR: Azure Data Factory Azure Synapse Analytics

İpucu

Kuruluşlar için hepsi bir arada analiz çözümü olan Microsoft Fabric'te Data Factory'yi deneyin. Microsoft Fabric , veri taşımadan veri bilimine, gerçek zamanlı analize, iş zekasına ve raporlamaya kadar her şeyi kapsar. Yeni bir deneme sürümünü ücretsiz olarak başlatmayı öğrenin!

İşlem Hattını Yürüt etkinliği, Data Factory veya Synapse işlem hattının başka bir işlem hattını çağırmasına olanak tanır.

Kullanıcı arabirimiyle İşlem Hattı Yürütme etkinliği oluşturma

İşlem hattında İşlem Hattı Yürütme etkinliğini kullanmak için aşağıdaki adımları tamamlayın:

  1. İşlem hattı Etkinlikleri bölmesinde işlem hattını arayın ve İşlem Hattı Yürütme etkinliğini işlem hattı tuvaline sürükleyin.

  2. Tuvalde henüz seçili değilse yeni İşlem Hattını Yürüt etkinliğini ve ayrıntılarını düzenlemek için Ayarlar sekmesini seçin.

    İşlem hattı yürütme etkinliğinin kullanıcı arabirimini gösterir.

  3. Mevcut bir işlem hattını seçin veya Yeni düğmesini kullanarak yeni bir işlem hattı oluşturun. Diğer seçenekleri belirleyin ve yapılandırmanızı tamamlamak için işlem hattı için gerekli parametreleri yapılandırın.

Sözdizimi

{
    "name": "MyPipeline",
    "properties": {
        "activities": [
            {
                "name": "ExecutePipelineActivity",
                "type": "ExecutePipeline",
                "typeProperties": {
                    "parameters": {                        
                        "mySourceDatasetFolderPath": {
                            "value": "@pipeline().parameters.mySourceDatasetFolderPath",
                            "type": "Expression"
                        }
                    },
                    "pipeline": {
                        "referenceName": "<InvokedPipelineName>",
                        "type": "PipelineReference"
                    },
                    "waitOnCompletion": true
                 }
            }
        ],
        "parameters": [
            {
                "mySourceDatasetFolderPath": {
                    "type": "String"
                }
            }
        ]
    }
}

Tür özellikleri

Özellik Açıklama İzin verilen değerler Zorunlu
Adı İşlem hattı yürütme etkinliğinin adı. String Yes
Tür Şu şekilde ayarlanmalıdır: ExecutePipeline. String Yes
ardışık düzen Bu işlem hattının çağırdığını bağımlı işlem hattına işlem hattı başvurusu. İşlem hattı başvuru nesnesinin iki özelliği vardır: referenceName ve type. referenceName özelliği, başvuru işlem hattının adını belirtir. type özelliği PipelineReference olarak ayarlanmalıdır. PipelineReference Yes
parametreler Çağrılan işlem hattına geçirilecek parametreler Parametre adlarını bağımsız değişken değerleriyle eşleyen bir JSON nesnesi Hayır
waitOnCompletion Etkinlik yürütmenin bağımlı işlem hattı yürütmesinin tamamlanmasını bekleyip beklemediğini tanımlar. Varsayılanı doğrudur Boolean Hayır

Örnek

Bu senaryoda iki işlem hattı vardır:

  • Ana işlem hattı - Bu işlem hattının çağrılan işlem hattını çağıran bir İşlem Hattı Yürütme etkinliği vardır. Ana işlem hattı iki parametre alır: masterSourceBlobContainer, masterSinkBlobContainer.
  • Çağrılan işlem hattı - Bu işlem hattının bir Azure Blob kaynağından Azure Blob havuzuna veri kopyalayan bir Kopyalama etkinliği vardır. Çağrılan işlem hattı iki parametre alır: sourceBlobContainer, sinkBlobContainer.

Ana işlem hattı tanımı

{
  "name": "masterPipeline",
  "properties": {
    "activities": [
      {
        "type": "ExecutePipeline",
        "typeProperties": {
          "pipeline": {
            "referenceName": "invokedPipeline",
            "type": "PipelineReference"
          },
          "parameters": {
            "sourceBlobContainer": {
              "value": "@pipeline().parameters.masterSourceBlobContainer",
              "type": "Expression"
            },
            "sinkBlobContainer": {
              "value": "@pipeline().parameters.masterSinkBlobContainer",
              "type": "Expression"
            }
          },
          "waitOnCompletion": true
        },
        "name": "MyExecutePipelineActivity"
      }
    ],
    "parameters": {
      "masterSourceBlobContainer": {
        "type": "String"
      },
      "masterSinkBlobContainer": {
        "type": "String"
      }
    }
  }
}

Çağrılan işlem hattı tanımı

{
  "name": "invokedPipeline",
  "properties": {
    "activities": [
      {
        "type": "Copy",
        "typeProperties": {
          "source": {
            "type": "BlobSource"
          },
          "sink": {
            "type": "BlobSink"
          }
        },
        "name": "CopyBlobtoBlob",
        "inputs": [
          {
            "referenceName": "SourceBlobDataset",
            "type": "DatasetReference"
          }
        ],
        "outputs": [
          {
            "referenceName": "sinkBlobDataset",
            "type": "DatasetReference"
          }
        ]
      }
    ],
    "parameters": {
      "sourceBlobContainer": {
        "type": "String"
      },
      "sinkBlobContainer": {
        "type": "String"
      }
    }
  }
}

Bağlı hizmet

{
    "name": "BlobStorageLinkedService",
    "properties": {
    "type": "AzureStorage",
    "typeProperties": {
      "connectionString": "DefaultEndpointsProtocol=https;AccountName=*****;AccountKey=*****"
    }
  }
}

Kaynak veri kümesi

{
    "name": "SourceBlobDataset",
    "properties": {
    "type": "AzureBlob",
    "typeProperties": {
      "folderPath": {
        "value": "@pipeline().parameters.sourceBlobContainer",
        "type": "Expression"
      },
      "fileName": "salesforce.txt"
    },
    "linkedServiceName": {
      "referenceName": "BlobStorageLinkedService",
      "type": "LinkedServiceReference"
    }
  }
}

Havuz veri kümesi

{
    "name": "sinkBlobDataset",
    "properties": {
    "type": "AzureBlob",
    "typeProperties": {
      "folderPath": {
        "value": "@pipeline().parameters.sinkBlobContainer",
        "type": "Expression"
      }
    },
    "linkedServiceName": {
      "referenceName": "BlobStorageLinkedService",
      "type": "LinkedServiceReference"
    }
  }
}

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

Bu örnekte ana işlem hattını çalıştırmak için masterSourceBlobContainer ve masterSinkBlobContainer parametreleri için aşağıdaki değerler geçirilir:

{
  "masterSourceBlobContainer": "executetest",
  "masterSinkBlobContainer": "executesink"
}

Ana işlem hattı, aşağıdaki örnekte gösterildiği gibi bu değerleri çağrılan işlem hattına iletir:

{
    "type": "ExecutePipeline",
    "typeProperties": {
      "pipeline": {
        "referenceName": "invokedPipeline",
        "type": "PipelineReference"
      },
      "parameters": {
        "sourceBlobContainer": {
          "value": "@pipeline().parameters.masterSourceBlobContainer",
          "type": "Expression"
        },
        "sinkBlobContainer": {
          "value": "@pipeline().parameters.masterSinkBlobContainer",
          "type": "Expression"
        }
      },

      ....
}

Uyarı

İşlem Hattı Yürütme etkinliği, dizi parametresini alt işlem hattına dize olarak geçirir. Bunun nedeni yükün üst işlem hattından alt öğeye >dize olarak geçirilmesidir. Alt işlem hattına geçirilen girişi denetlediğimizde bunu görebiliriz. Daha fazla ayrıntı için bu bölüme bakın.

İlgili içerik

Desteklenen diğer denetim akışı etkinliklerine bakın: