Untuk Setiap aktivitas di Azure Data Factory dan Azure Synapse Analytics
BERLAKU UNTUK:
Azure Data Factory Azure Synapse Analytics
Tip
Cobalah Data Factory di Microsoft Fabric, solusi analitik all-in-one untuk perusahaan. Microsoft Fabric mencakup semuanya mulai dari pergerakan data hingga ilmu data, analitik real time, kecerdasan bisnis, dan pelaporan. Pelajari cara memulai uji coba baru secara gratis!
ForEach Activity mendefinisikan aliran kontrol berulang dalam pipa Azure Data Factory atau Synapse. Aktivitas ini digunakan untuk mengulangi koleksi dan menjalankan aktivitas tertentu dalam perulangan. Implementasi perulangan dari aktivitas ini mirip dengan struktur perulangan Foreach dalam bahasa pemrograman.
Membuat aktivitas ForEach dengan UI
Untuk menggunakan aktivitas ForEach dalam alur, selesaikan langkah-langkah berikut:
Anda dapat menggunakan variabel jenis larik apa pun atau output dari aktivitas lain sebagai input untuk aktivitas ForEach. Untuk membuat variabel larik, pilih latar belakang kanvas alur, lalu pilih tab Variabel untuk menambahkan variabel jenis larik seperti yang ditunjukkan di bawah ini.
Cari ForEach di panel Aktivitas alur, dan seret aktivitas ForEach ke kanvas alur.
Pilih aktivitas Filter baru di kanvas jika belum dipilih, dan tab Pengaturan, untuk mengedit detailnya.
Pilih bidang Item lalu pilih tautan Tambahkan konten dinamis untuk membuka panel penyunting konten dinamis.
Pilih larik input Anda yang akan difilter di editor konten dinamis. Dalam contoh ini, kita memilih variabel yang dibuat pada langkah pertama.
Pilih editor Aktivitas pada aktivitas ForEach untuk menambahkan satu atau beberapa aktivitas yang akan dijalankan untuk setiap item dalam array Item input.
Pada aktivitas apa pun yang Anda buat dalam aktivitas ForEach, Anda dapat mereferensikan item saat ini yang sedang diulangi aktivitas ForEach dari daftar Item. Anda dapat mereferensikan item saat ini di mana saja Anda dapat menggunakan ekspresi dinamis untuk menentukan nilai properti. Di editor konten dinamis, pilih iterator ForEach untuk menampilkan item saat ini.
Sintaks
Properti dijelaskan nanti di artikel ini. Properti item adalah koleksi dan setiap item dalam koleksi disebut dengan menggunakan seperti yang @item() ditunjukkan dalam sintaks berikut:
{
"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()"
}
}
]
}
]
}
}
Properti jenis
| Properti | Deskripsi | Nilai yang diizinkan | Wajib |
|---|---|---|---|
| nama | Nama untuk setiap aktivitas. | String | Ya |
| jenis | Harus diatur ke ForEach | String | Ya |
| isSequential | Menentukan apakah perulangan harus dijalankan secara berurutan atau paralel. Maksimal 50 iterasi loop dapat dieksekusi sekaligus secara paralel). Misalnya, jika Anda memiliki aktivitas ForEach yang berulang-ulang atas aktivitas salinan dengan 10 sumber berbeda dan himpunan data sink dengan isSequential diatur ke False, semua salinan dieksekusi sekaligus. Defaultnya adalah False. Jika "isSequential" diatur ke False, pastikan bahwa ada konfigurasi yang benar untuk menjalankan beberapa executable. Jika tidak, properti ini harus digunakan dengan hati-hati untuk menghindari menimbulkan konflik tulis. Untuk informasi selengkapnya, lihat bagian Eksekusi paralel. |
Boolean | Tidak. Defaultnya adalah False. |
| batchCount | Jumlah batch yang akan digunakan untuk mengontrol jumlah eksekusi paralel (ketika isSequential diatur ke salah). Ini adalah batas konkurensi atas, tetapi untuk setiap aktivitas tidak akan selalu dieksekusi pada angka ini | Bilangan bulat (maksimum 50) | Tidak. defaultnya adalah 20. |
| Item | Ekspresi yang mengembalikan JSON Array untuk di-iterasi. | Ekspresi (yang mengembalikan Array JSON) | Ya |
| Aktivitas | Aktivitas yang akan dilaksanakan. | Array Aktivitas | Ya |
Eksekusi paralel
Jika isSequential disetel ke false, aktivitas berulang secara paralel dengan maksimum 50 iterasi bersamaan. Pengaturan ini harus digunakan dengan hati-hati. Jika iterasi bersamaan menulis ke folder yang sama tetapi untuk file yang berbeda, pendekatan ini baik-baik saja. Jika iterasi bersamaan menulis secara bersamaan ke file yang sama persis, pendekatan ini kemungkinan besar menyebabkan kesalahan.
Bahasa pemrograman ekspresi iterasi
Dalam aktivitas aliran kontrol seperti aktivitas ForEach, Anda dapat menyediakan larik untuk diulangi untuk itemproperti." Gunakan @item() untuk mengulangi satu enumerasi dalam aktivitas ForEach. Misalnya, jika item adalah array: [1, 2, 3], @item() mengembalikan 1 dalam iterasi pertama, 2 di iterasi kedua, dan 3 dalam iterasi ketiga. Anda juga dapat menggunakan @range(0,10) ekspresi suka untuk melakukan iterasi sepuluh kali dimulai dengan 0 berakhiran dengan 9.
Iterasi selama satu aktivitas
Skenario: Salin dari file sumber yang sama di Azure Blob ke beberapa file tujuan di Azure Blob.
Definisi alur
{
"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"
}
}
}
}
Definisi himpunan data blob
{
"name":"<MyDataset>",
"properties":{
"type":"AzureBlob",
"typeProperties":{
"folderPath":{
"value":"@dataset().MyFolderPath",
"type":"Expression"
}
},
"linkedServiceName":{
"referenceName":"StorageLinkedService",
"type":"LinkedServiceReference"
},
"parameters":{
"MyFolderPath":{
"type":"String"
}
}
}
}
Jalankan nilai parameter
{
"mySourceDatasetFolderPath": "input/",
"mySinkDatasetFolderPath": [ "outputs/file1", "outputs/file2" ]
}
Iterasi selama beberapa kegiatan
Dimungkinkan untuk melakukan iterasi melalui beberapa aktivitas (misalnya: menyalin dan aktivitas web) dalam aktivitas ForEach. Dalam skenario ini, kami sarankan Anda mengabstraksi beberapa aktivitas ke dalam alur terpisah. Kemudian, Anda dapat menggunakan aktivitas ExecutePipeline di alur dengan aktivitas ForEach untuk memanggil alur terpisah dengan beberapa aktivitas.
Sintaks
{
"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": {
...
}
}
}
Contoh
Skenario: Iterasi melalui InnerPipeline dalam aktivitas ForEach dengan aktivitas Execute Pipeline. Salinan alur bagian dalam dengan definisi skema di parameter.
Definisi Alur Master
{
"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"
}
}
}
}
Definisi Alur Inner
{
"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"
}
}
}
}
Definisi himpunan data sumber
{
"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"
}
}
}
}
Definisi himpunan data sink
{
"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"
}
}
}
}
Parameter alur master
{
"inputtables": [
{
"SourceTable": "department",
"SourceTableStructure": [
{
"name": "departmentid",
"type": "int"
},
{
"name": "departmentname",
"type": "string"
}
],
"DestTable": "department2",
"DestTableStructure": [
{
"name": "departmentid",
"type": "int"
},
{
"name": "departmentname",
"type": "string"
}
]
}
]
}
Menggabungkan output
Untuk menggabungkan output aktivitas foreach, silakan gunakan Variabel dan aktivitas Variabel Append.
Pertama, deklarasikan arrayvariabel dalam alur. Kemudian, panggil aktivitas Variabel Append di dalam setiap loop foreach. Selanjutnya, Anda dapat mengambil agregasi dari array Anda.
Batasan dan solusi
Berikut adalah beberapa keterbatasan aktivitas ForEach dan solusi yang disarankan.
| Pembatasan | Solusi Sementara |
|---|---|
| Anda tidak dapat menumpuk perulangan ForEach di dalam perulangan ForEach lain (atau perulangan Until). | Rancang alur dua tingkat di mana pipa luar dengan perulangan ForEach luar mengulangi melalui pipa bagian dalam dengan perulangan berlapis. |
Kegiatan ForEach memiliki maksimum batchCount 50 untuk pemrosesan paralel, dan maksimal 100.000 item. |
Rancang alur dua tingkat di mana pipa luar dengan aktivitas ForEach berulang di atas alur bagian dalam. |
| SetVariable tidak dapat digunakan di dalam aktivitas ForEach yang berjalan secara paralel karena variabel global untuk seluruh alur, mereka tidak tercakup ke ForEach atau aktivitas lainnya. | Pertimbangkan untuk menggunakan ForEach berurutan atau gunakan Jalankan Alur di dalam ForEach (Variabel/Parameter yang dihandel di Alur elemen anak). |
Konten terkait
Lihat aktivitas alur kontrol yang didukung lainnya: