Eseguire l'attività pipeline in Azure Data Factory e Synapse Analytics

Articolo
12/09/2023

SI APPLICA A: Azure Data Factory Azure Synapse Analytics

Suggerimento

Provare Data Factory in Microsoft Fabric, una soluzione di analisi completa per le aziende. Microsoft Fabric copre tutti gli elementi, dallo spostamento dei dati all'analisi scientifica dei dati, all'analisi in tempo reale, alla business intelligence e alla creazione di report. Scopri come avviare gratuitamente una nuova versione di valutazione .

L'attività Execute Pipeline consente a una pipeline di Data Factory o Synapse di richiamare un'altra pipeline.

Creare un'attività Execute Pipeline con l'interfaccia utente

Per usare un'attività Execute Pipeline in una pipeline, completare la procedura seguente:

Cercare la pipeline nel riquadro Attività pipeline e trascinare un'attività Esegui pipeline nell'area di disegno della pipeline.
Selezionare la nuova attività Execute Pipeline (Esegui pipeline) nell'area di disegno se non è già selezionata e la relativa scheda Impostazioni per modificarne i dettagli.
Selezionare una pipeline esistente o crearne una nuova usando il pulsante Nuovo. Selezionare altre opzioni e configurare i parametri per la pipeline in base alle esigenze per completare la configurazione.

Sintassi

{
    "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"
                }
            }
        ]
    }
}

Proprietà del tipo

Proprietà	Descrizione	Valori consentiti	Obbligatorio
name	Nome dell'attività di esecuzione della pipeline.	Stringa	Sì
Tipo	Deve essere impostata su: ExecutePipeline.	Stringa	Sì
pipeline	Pipeline di riferimento per la pipeline dipendente chiamata da questa pipeline. Un oggetto di riferimento della pipeline ha due proprietà: referenceName e type. La proprietà referenceName specifica il nome della pipeline di riferimento. La proprietà type deve essere impostata su PipelineReference.	PipelineReference	Sì
parameters	Parametri da passare alla pipeline chiamata	Oggetto JSON che esegue il mapping dei nomi di parametro ai valori degli argomenti	No
waitOnCompletion	Definisce se l'esecuzione dell'attività attende che l'esecuzione della pipeline dipendente termini. Il valore predefinito è true.	Boolean	No

Esempio

Questo scenario include due pipeline:

Pipeline master: questa pipeline ha una sola attività di esecuzione della pipeline che chiama la pipeline richiamata. La pipeline master accetta due parametri: masterSourceBlobContainer e masterSinkBlobContainer.
Pipeline richiamata: questa pipeline ha un'attività Copy che copia i dati da un'origine BLOB di Azure al sink di BLOB di Azure. La pipeline richiamata accetta due parametri: sourceBlobContainer, sinkBlobContainer.

Definizione della pipeline master

{
  "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"
      }
    }
  }
}

Definizione della pipeline richiamata

{
  "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"
      }
    }
  }
}

Servizio collegato

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

Set di dati di origine

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

Set di dati sink

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

Esecuzione della pipeline

Per eseguire la pipeline master di questo esempio, vengono passati i valori seguenti per i parametri masterSourceBlobContainer e masterSinkBlobContainer:

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

La pipeline master inoltra questi valori alla pipeline richiamata, come illustrato nell'esempio seguente:

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

      ....
}

Avviso

L'attività Esegui pipeline passa il parametro della matrice come stringa alla pipeline figlio. Ciò è dovuto al fatto che il payload viene passato dalla pipeline padre al >figlio come stringa. È possibile vederlo quando si controlla l'input passato alla pipeline figlio. Per altri dettagli, vedere questa sezione .

Vedere altre attività del flusso di controllo supportate: