Azure Data Factory ve Azure Synapse Analytics'te ForEach etkinliği

UYGULANANLAR: Azure Data Factory Azure Synapse Analytics

ForEach Etkinliği, Azure Data Factory veya Synapse işlem hattında yinelenen bir denetim akışı tanımlar. Bu etkinlik bir koleksiyon üzerinde yinelemek için kullanılır ve bir döngüde belirtilen etkinlikleri yürütür. Bu etkinliğin döngü uygulaması, programlama dillerindeki Foreach döngü yapısına benzer.

Kullanıcı arabirimiyle ForEach etkinliği oluşturma

İşlem hattında ForEach etkinliği kullanmak için aşağıdaki adımları tamamlayın:

1. ForEach etkinliğinizin girişi olarak herhangi bir dizi türü değişkenini veya diğer etkinliklerden gelen çıkışları kullanabilirsiniz. Dizi değişkeni oluşturmak için işlem hattı tuvalinin arka planını seçin ve ardından Değişkenler sekmesini seçerek aşağıda gösterildiği gibi bir dizi türü değişkeni ekleyin.

   

2. İşlem hattı Etkinlikleri bölmesinde ForEach'i arayın ve bir ForEach etkinliğini işlem hattı tuvaline sürükleyin.

3. Tuvalde henüz seçili değilse yeni ForEach etkinliğini ve ayrıntılarını düzenlemek için Ayarlar sekmesini seçin.

   

4. Öğeler alanını seçin ve dinamik içerik düzenleyicisi bölmesini açmak için Dinamik içerik ekle bağlantısını seçin.

   

5. Dinamik içerik düzenleyicisinde filtrelenecek giriş dizinizi seçin. Bu örnekte, ilk adımda oluşturulan değişkeni seçiyoruz.

   

6. Giriş Öğeleri dizisindeki her öğe için yürütülecek bir veya daha fazla etkinlik eklemek için ForEach etkinliğindeki Etkinlikler düzenleyicisini seçin.

   

7. ForEach etkinliği içinde oluşturduğunuz tüm etkinliklerde, Öğeler listesinden ForEach etkinliğinin yinelemede olduğu geçerli öğeye başvurabilirsiniz. Bir özellik değeri belirtmek için dinamik ifadeyi kullanabileceğiniz her yerde geçerli öğeye başvurabilirsiniz. Geçerli öğeyi döndürmek için dinamik içerik düzenleyicisinde ForEach yineleyicisini seçin.

   

Sözdizimi

Özellikler bu makalenin devamında açıklanmıştır. items özelliği koleksiyondur ve koleksiyondaki her öğeye aşağıdaki söz diziminde gösterildiği gibi kullanılarak @item() başvurulur:

{  
   "name":"MyForEachActivityName",
   "type":"ForEach",
   "typeProperties":{  
      "isSequential":"true",
        "items": {
            "value": "@pipeline().parameters.mySinkDatasetFolderPathCollection",
            "type": "Expression"
        },
      "activities":[  
         {  
            "name":"MyCopyActivity",
            "type":"Copy",
            "typeProperties":{  
               ...
            },
            "inputs":[  
               {  
                  "referenceName":"MyDataset",
                  "type":"DatasetReference",
                  "parameters":{  
                     "MyFolderPath":"@pipeline().parameters.mySourceDatasetFolderPath"
                  }
               }
            ],
            "outputs":[  
               {  
                  "referenceName":"MyDataset",
                  "type":"DatasetReference",
                  "parameters":{  
                     "MyFolderPath":"@item()"
                  }
               }
            ]
         }
      ]
   }
}

Tür özellikleri

Paralel yürütme

isSequential false olarak ayarlanırsa, etkinlik en fazla 50 eşzamanlı yinelemeyle paralel olarak yinelenir. Bu ayar dikkatli kullanılmalıdır. Eşzamanlı yinelemeler aynı klasöre ancak farklı dosyalara yazıyorsa, bu yaklaşım uygundur. Eşzamanlı yinelemeler aynı dosyaya eşzamanlı olarak yazıyorsa, bu yaklaşım büyük olasılıkla bir hataya neden olur.

Yineleme ifade dili

ForEach etkinliğinde, özellik öğeleri için yinelenecek bir dizi sağlayın." ForEach etkinliğindeki tek bir numaralandırmayı yinelemek için kullanın @item() . Örneğin, öğeler bir diziyse: [1, 2, 3], ilk yinelemede 1, @item() ikinci yinelemede 2 ve üçüncü yinelemede 3 döndürür. 0 ile başlayan ve 9 ile biten on kez yinelemek için like ifadesini de kullanabilirsiniz @range(0,10) .

Tek bir etkinlik üzerinde yineleme

Senaryo: Azure Blob'daki aynı kaynak dosyadan Azure Blob'daki birden çok hedef dosyaya kopyalayın.

İşlem hattı tanımı

{
    "name": "<MyForEachPipeline>",
    "properties": {
        "activities": [
            {
                "name": "<MyForEachActivity>",
                "type": "ForEach",
                "typeProperties": {
                    "isSequential": "true",
                    "items": {
                        "value": "@pipeline().parameters.mySinkDatasetFolderPath",
                        "type": "Expression"
                    },
                    "activities": [
                        {
                            "name": "MyCopyActivity",
                            "type": "Copy",
                            "typeProperties": {
                                "source": {
                                    "type": "BlobSource",
                                    "recursive": "false"
                                },
                                "sink": {
                                    "type": "BlobSink",
                                    "copyBehavior": "PreserveHierarchy"
                                }
                            },
                            "inputs": [
                                {
                                    "referenceName": "<MyDataset>",
                                    "type": "DatasetReference",
                                    "parameters": {
                                        "MyFolderPath": "@pipeline().parameters.mySourceDatasetFolderPath"
                                    }
                                }
                            ],
                            "outputs": [
                                {
                                    "referenceName": "MyDataset",
                                    "type": "DatasetReference",
                                    "parameters": {
                                        "MyFolderPath": "@item()"
                                    }
                                }
                            ]
                        }
                    ]
                }
            }
        ],
        "parameters": {
            "mySourceDatasetFolderPath": {
                "type": "String"
            },
            "mySinkDatasetFolderPath": {
                "type": "String"
            }
        }
    }
}

Blob veri kümesi tanımı

{  
   "name":"<MyDataset>",
   "properties":{  
      "type":"AzureBlob",
      "typeProperties":{  
         "folderPath":{  
            "value":"@dataset().MyFolderPath",
            "type":"Expression"
         }
      },
      "linkedServiceName":{  
         "referenceName":"StorageLinkedService",
         "type":"LinkedServiceReference"
      },
      "parameters":{  
         "MyFolderPath":{  
            "type":"String"
         }
      }
   }
}

Parametre değerlerini çalıştırma

{
    "mySourceDatasetFolderPath": "input/",
    "mySinkDatasetFolderPath": [ "outputs/file1", "outputs/file2" ]
}

Birden çok etkinlikte yineleme

Bir ForEach etkinliğindeki birden çok etkinlik (örneğin, kopyalama ve web etkinlikleri) üzerinde yineleme yapmak mümkündür. Bu senaryoda, birden çok etkinliği ayrı bir işlem hattında soyutlamanızı öneririz. Ardından, birden çok etkinliğe sahip ayrı işlem hattını çağırmak için ForEach etkinliğiyle işlem hattındaki ExecutePipeline etkinliğini kullanabilirsiniz.

Sözdizimi

{
  "name": "masterPipeline",
  "properties": {
    "activities": [
      {
        "type": "ForEach",
        "name": "<MyForEachMultipleActivities>"
        "typeProperties": {
          "isSequential": true,
          "items": {
            ...
          },
          "activities": [
            {
              "type": "ExecutePipeline",
              "name": "<MyInnerPipeline>"
              "typeProperties": {
                "pipeline": {
                  "referenceName": "<copyHttpPipeline>",
                  "type": "PipelineReference"
                },
                "parameters": {
                  ...
                },
                "waitOnCompletion": true
              }
            }
          ]
        }
      }
    ],
    "parameters": {
      ...
    }
  }
}

Örnek

Senaryo: İşlem Hattı Yürütme etkinliğiyle bir ForEach etkinliği içindeki bir InnerPipeline üzerinde yineleme yapın. İç işlem hattı, şema tanımları parametreli olarak kopyalanır.

Ana İşlem Hattı tanımı

{
  "name": "masterPipeline",
  "properties": {
    "activities": [
      {
        "type": "ForEach",
        "name": "MyForEachActivity",
        "typeProperties": {
          "isSequential": true,
          "items": {
            "value": "@pipeline().parameters.inputtables",
            "type": "Expression"
          },
          "activities": [
            {
              "type": "ExecutePipeline",
              "typeProperties": {
                "pipeline": {
                  "referenceName": "InnerCopyPipeline",
                  "type": "PipelineReference"
                },
                "parameters": {
                  "sourceTableName": {
                    "value": "@item().SourceTable",
                    "type": "Expression"
                  },
                  "sourceTableStructure": {
                    "value": "@item().SourceTableStructure",
                    "type": "Expression"
                  },
                  "sinkTableName": {
                    "value": "@item().DestTable",
                    "type": "Expression"
                  },
                  "sinkTableStructure": {
                    "value": "@item().DestTableStructure",
                    "type": "Expression"
                  }
                },
                "waitOnCompletion": true
              },
              "name": "ExecuteCopyPipeline"
            }
          ]
        }
      }
    ],
    "parameters": {
      "inputtables": {
        "type": "Array"
      }
    }
  }
}

İç işlem hattı tanımı

{
  "name": "InnerCopyPipeline",
  "properties": {
    "activities": [
      {
        "type": "Copy",
        "typeProperties": {
          "source": {
            "type": "SqlSource",
            }
          },
          "sink": {
            "type": "SqlSink"
          }
        },
        "name": "CopyActivity",
        "inputs": [
          {
            "referenceName": "sqlSourceDataset",
            "parameters": {
              "SqlTableName": {
                "value": "@pipeline().parameters.sourceTableName",
                "type": "Expression"
              },
              "SqlTableStructure": {
                "value": "@pipeline().parameters.sourceTableStructure",
                "type": "Expression"
              }
            },
            "type": "DatasetReference"
          }
        ],
        "outputs": [
          {
            "referenceName": "sqlSinkDataset",
            "parameters": {
              "SqlTableName": {
                "value": "@pipeline().parameters.sinkTableName",
                "type": "Expression"
              },
              "SqlTableStructure": {
                "value": "@pipeline().parameters.sinkTableStructure",
                "type": "Expression"
              }
            },
            "type": "DatasetReference"
          }
        ]
      }
    ],
    "parameters": {
      "sourceTableName": {
        "type": "String"
      },
      "sourceTableStructure": {
        "type": "String"
      },
      "sinkTableName": {
        "type": "String"
      },
      "sinkTableStructure": {
        "type": "String"
      }
    }
  }
}

Kaynak veri kümesi tanımı

{
  "name": "sqlSourceDataset",
  "properties": {
    "type": "SqlServerTable",
    "typeProperties": {
      "tableName": {
        "value": "@dataset().SqlTableName",
        "type": "Expression"
      }
    },
    "structure": {
      "value": "@dataset().SqlTableStructure",
      "type": "Expression"
    },
    "linkedServiceName": {
      "referenceName": "sqlserverLS",
      "type": "LinkedServiceReference"
    },
    "parameters": {
      "SqlTableName": {
        "type": "String"
      },
      "SqlTableStructure": {
        "type": "String"
      }
    }
  }
}

Havuz veri kümesi tanımı

{
  "name": "sqlSinkDataSet",
  "properties": {
    "type": "AzureSqlTable",
    "typeProperties": {
      "tableName": {
        "value": "@dataset().SqlTableName",
        "type": "Expression"
      }
    },
    "structure": {
      "value": "@dataset().SqlTableStructure",
      "type": "Expression"
    },
    "linkedServiceName": {
      "referenceName": "azureSqlLS",
      "type": "LinkedServiceReference"
    },
    "parameters": {
      "SqlTableName": {
        "type": "String"
      },
      "SqlTableStructure": {
        "type": "String"
      }
    }
  }
}

Ana işlem hattı parametreleri

{
    "inputtables": [
        {
            "SourceTable": "department",
            "SourceTableStructure": [
              {
                "name": "departmentid",
                "type": "int"
              },
              {
                "name": "departmentname",
                "type": "string"
              }
            ],
            "DestTable": "department2",
            "DestTableStructure": [
              {
                "name": "departmentid",
                "type": "int"
              },
              {
                "name": "departmentname",
                "type": "string"
              }
            ]
        }
    ]
    
}

Çıkışları toplama

Foreach etkinliğinin çıkışlarını toplamak için değişkenleri ve Değişken Ekle etkinliğini kullanabilirsiniz.

İlk olarak, işlem hattında bir array değişken bildirin. Ardından, her foreach döngüsünün içinde Değişken Ekle etkinliğini çağırabilirsiniz. Daha sonra, toplamayı dizinizden alabilirsiniz.

Sınırlamalar ve geçici çözümler

ForEach etkinliğinin bazı sınırlamaları ve önerilen geçici çözümler aşağıdadır.

İlgili içerik

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

• İşlem Hattı Yürütme Etkinliği
• Meta Veri Alma Etkinliği
• Arama Etkinliği
• Web Etkinliği