Actividad de ejecución de canalización en Azure Data Factory y Synapse Analytics

Artículo
10/20/2023

SE APLICA A: Azure Data Factory Azure Synapse Analytics

Sugerencia

Pruebe Data Factory en Microsoft Fabric, una solución de análisis todo en uno para empresas. Microsoft Fabric abarca todo, desde el movimiento de datos hasta la ciencia de datos, el análisis en tiempo real, la inteligencia empresarial y los informes. Obtenga información sobre cómo iniciar una nueva evaluación gratuita.

La actividad de ejecución de canalización permite que una canalización de Data Factory o Synapse invoque a otra.

Creación de una actividad de ejecución de canalización con UI

Para usar una actividad de ejecución de canalización en una canalización, complete los pasos siguientes:

Busque canalización en el panel Actividades de canalización y arrastre una actividad de ejecución de canalización al lienzo de canalización.
Seleccione la nueva actividad de ejecución de canalización en el lienzo si aún no está seleccionada y su pestaña Configuración para editar sus detalles.
Seleccione una canalización existente o cree una nueva con el botón Nuevo. Seleccione otras opciones y configure los parámetros de la canalización según sea necesario para completar la configuración.

Sintaxis

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

Propiedades de tipo

Propiedad	Descripción	Valores permitidos	Obligatorio
name	Nombre de la actividad de ejecución de canalización.	String	Sí
type	Se debe establecer en ExecutePipeline.	String	Sí
pipeline	Referencia a la canalización dependiente que invoca esta canalización. Un objeto de referencia de canalización tiene dos propiedades: referenceName y type. La propiedad referenceName especifica el nombre de la canalización de referencia. La propiedad type se debe establecer en PipelineReference.	PipelineReference	Sí
parámetros	Parámetros que se deben pasar a la canalización invocada	Objeto JSON que asigna nombres de parámetro a los valores de argumento	No
waitOnCompletion	Define si la ejecución de la actividad espera que finalice la ejecución de la canalización dependiente. El valor predeterminado es true.	Boolean	No

Muestra

Este escenario consta de dos canalizaciones:

Canalización principal. Esta canalización tiene una actividad de ejecución de canalización que llama a la canalización invocada. La canalización principal toma dos parámetros: masterSourceBlobContainer y masterSinkBlobContainer.
Canalización invocada. Esta canalización tiene una actividad de copia que copia los datos de un origen de Azure Blob a un receptor de Azure Blob. La canalización invocada toma dos parámetros: sourceBlobContainer y sinkBlobContainer.

Definición de la canalización principal

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

Definición de la canalización invocada

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

Servicio vinculado

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

Conjunto de datos de origen

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

Conjunto de datos del receptor

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

Ejecutar la canalización

Para ejecutar la canalización principal de este ejemplo, se pasan los siguientes valores para los parámetros masterSourceBlobContainer y masterSinkBlobContainer:

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

La canalización principal reenvía estos valores a la canalización invocada, como se muestra en el ejemplo siguiente:

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

      ....
}

Advertencia

La actividad de ejecución de canalización pasa un parámetro de matriz como cadena a la canalización secundaria. Esto se debe a que la carga útil se pasa de la canalización primaria a la secundaria como una cadena. Esto se puede ver cuando se comprueba la entrada que se pasa a la canalización secundaria. Consulte esta sección para obtener más detalles.

Vea otras actividades de flujo de control admitidas: