Ejecutar la actividad de Pipeline en Azure Data Factory y en Synapse Analytics

SE APLICA A: Azure Data Factory Azure Synapse Analytics

Sugerencia

Data Factory en Microsoft Fabric es la próxima generación de Azure Data Factory, con una arquitectura más sencilla, inteligencia artificial integrada y nuevas características. Si no está familiarizado con la integración de datos, comience con Fabric Data Factory. Las cargas de trabajo de ADF existentes pueden actualizarse a Fabric para acceder a nuevas funcionalidades en ciencia de datos, análisis en tiempo real e informes.

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:

  1. 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.

  2. Seleccione la nueva actividad "Ejecutar canalización" en el lienzo, si aún no está seleccionada, y su pestaña de Configuración para editar sus detalles.

    Muestra la interfaz de usuario para una actividad de ejecución de canalización.

  3. 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
nombre Nombre de la actividad de ejecución de canalización. Cadena Sí
tipo Debe establecerse en ExecutePipeline. Cadena 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 verdadero. Boolean No

Muestra

Este escenario consta de dos canalizaciones:

  • Canalización principal - Esta canalización contiene una actividad "Ejecutar canalización" que invoca a la canalización específica. 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 lista 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.

Contenido relacionado

Vea otras actividades de flujo de control admitidas:

  • Last updated on