Creación de canalizaciones de Unity Catalog mediante la clonación de canalizaciones de metastore de Hive

La solicitud clone a pipeline de la API REST de Databricks copia un pipeline existente que publica en el metastore de Hive a un nuevo pipeline que publica en Unity Catalog. Cuando llamas a clone a pipeline, sucede lo siguiente:

  • Copia el código fuente y la configuración de la canalización existente a una nueva, aplicando cualquier sustitución de configuración que haya especificado.
  • Actualiza las definiciones de las vistas materializadas y de las tablas de streaming, así como las referencias, con los cambios necesarios para que esos objetos sean administrados por Unity Catalog.
  • Inicia una actualización del pipeline para migrar los datos y metadatos existentes, como los puntos de control, de cualquier tabla de streaming del pipeline. Esto permite que dichas tablas de streaming reanuden el procesamiento en el mismo punto que el pipeline original.

Una vez completada la operación de clonación, tanto las canalizaciones originales como las nuevas se pueden ejecutar de forma independiente.

En esta página se incluyen ejemplos de llamada a la solicitud de API directamente y a través de un script de Python desde un cuaderno de Databricks.

Antes de empezar

Se requieren lo siguiente antes de clonar una canalización:

  • Para clonar una canalización de metastore de Hive, las tablas y vistas definidas en la canalización deben publicar tablas en un esquema de destino. Para obtener información sobre cómo agregar un esquema de destino a una canalización, consulte Configuración de una canalización para publicar en metastore de Hive.

  • Las referencias a tablas o vistas administradas por el metastore de Hive en la canalización que se va a clonar deben estar completamente calificadas con el catálogo (hive_metastore), el esquema y el nombre de la tabla. Por ejemplo, en el código siguiente que crea un customers conjunto de datos, el argumento de nombre de tabla debe actualizarse a hive_metastore.sales.customers:

    @dp.table
def customers():
  return spark.read.table("sales.customers").where(...)

  • No edite el código fuente de la canalización de metastore de Hive de origen mientras una operación de clonación está en curso, incluidos los cuadernos configurados como parte de la canalización y los módulos almacenados en carpetas de Git o archivos del área de trabajo.

  • La canalización del metastore de Hive de origen no debe ejecutarse al iniciar la operación de clonación. Si se está ejecutando una actualización, deténgala o espere a que se complete.

A continuación se muestran otras consideraciones importantes antes de clonar una canalización:

  • Si las tablas de la canalización de metastore de Hive especifican una ubicación de almacenamiento mediante el argumento /> en SQL, pase la configuración de a la solicitud de clonación. La configuración de esta configuración se incluye en las instrucciones siguientes.
  • Si el pipeline del metastore de Hive incluye una fuente Auto Loader que especifica un valor para la opción cloudFiles.schemaLocation, y el pipeline del metastore de Hive sigue operativo tras crear el clon en Unity Catalog, debe establecer la opción mergeSchema en true tanto en el pipeline del metastore de Hive como en el pipeline clonado de Unity Catalog. Al añadir esta opción al pipeline del metastore de Hive antes de la clonación, la opción se copia al nuevo pipeline.

Clonación de una canalización con la API REST de Databricks

En el ejemplo siguiente se usa el curl comando para llamar a la clone a pipeline solicitud en la API REST de Databricks:

curl -X POST \
     --header "Authorization: Bearer <personal-access-token>"  \
     <databricks-instance>/api/2.0/pipelines/<pipeline-id>/clone \
     --data @clone-pipeline.json

Reemplazar:

  • <personal-access-token> con un token de acceso personal de Databricks.
  • <databricks-instance> con el nombre de instancia del área de trabajo de Azure Databricks , por ejemplo adb-1234567890123456.7.azuredatabricks.net
  • <pipeline-id> con el identificador único de la canalización del metastore de Hive que se va a clonar. Puede encontrar el identificador de canalización en la interfaz de usuario de canalizaciones.

clone-pipeline.json:

{
  "catalog": "<target-catalog-name>",
  "target": "<target-schema-name>",
  "name": "<new-pipeline-name>",
  "clone_mode": "MIGRATE_TO_UC",
  "configuration": {
    "pipelines.migration.ignoreExplicitPath": "true"
  }
}

Reemplazar:

  • <target-catalog-name> con el nombre de un catálogo en Unity Catalog al que debe publicar la nueva canalización. Debe ser un catálogo existente.
  • <target-schema-name> con el nombre de un esquema en Unity Catalog al que debe publicar la nueva canalización si es diferente del nombre del esquema actual. Este parámetro es opcional y, si no se especifica, se usa el nombre de esquema existente.
  • <new-pipeline-name> con un nombre opcional para la nueva canalización. Si no se especifica, la nueva canalización se denomina mediante el nombre de canalización de origen con [UC] anexado.

clone_mode especifica el modo que se va a usar para la operación de clonación. MIGRATE_TO_UC es la única opción admitida.

Use el configuration campo para especificar configuraciones en la nueva canalización. Los valores establecidos aquí invalidan las configuraciones de la canalización original.

La respuesta de la solicitud REST API clone es el ID de canalización de la nueva canalización de Unity Catalog.

Clonación de una canalización desde un cuaderno de Databricks

En el ejemplo siguiente se llama a la solicitud create a pipeline desde un script de Python. Puede usar un cuaderno de Databricks para ejecutar este script:

  1. Crea un nuevo cuaderno para el script. Consulte Creación de un cuaderno.
  2. Copie el siguiente script Python en la primera celda del cuaderno.
  3. Actualice los valores de marcador de posición en el script reemplazándolos por:
    • <databricks-instance> con el nombre de instancia del área de trabajo de Azure Databricks , por ejemplo adb-1234567890123456.7.azuredatabricks.net
    • <pipeline-id> con el identificador único de la canalización del metastore de Hive que se va a clonar. Puede encontrar el identificador de canalización en la interfaz de usuario de canalizaciones.
    • <target-catalog-name> con el nombre de un catálogo en Unity Catalog al que debe publicar la nueva canalización. Debe ser un catálogo existente.
    • <target-schema-name> con el nombre de un esquema en Unity Catalog al que debe publicar la nueva canalización si es diferente del nombre del esquema actual. Este parámetro es opcional y, si no se especifica, se usa el nombre de esquema existente.
    • <new-pipeline-name> con un nombre opcional para la nueva canalización. Si no se especifica, la nueva canalización se denomina mediante el nombre de canalización de origen con [UC] anexado.
  4. Ejecute el script. Consulte Ejecución de cuadernos de Databricks.
import requests

# Your Databricks workspace URL, with no trailing spaces
WORKSPACE = "<databricks-instance>"

# The pipeline ID of the Hive metastore pipeline to clone
SOURCE_PIPELINE_ID = "<pipeline-id>"
# The target catalog name in Unity Catalog
TARGET_CATALOG = "<target-catalog-name>"
# (Optional) The name of a target schema in Unity Catalog. If empty, the same schema name as the Hive metastore pipeline is used
TARGET_SCHEMA = "<target-schema-name>"
# (Optional) The name of the new pipeline. If empty, the following is used for the new pipeline name: f"{originalPipelineName} [UC]"
CLONED_PIPELINE_NAME = "<new-pipeline-name>"

# This is the only supported clone mode
CLONE_MODE = "MIGRATE_TO_UC"

# Specify override configurations
OVERRIDE_CONFIGS = {"pipelines.migration.ignoreExplicitPath": "true"}

def get_token():
    ctx = dbutils.notebook.entry_point.getDbutils().notebook().getContext()
    return getattr(ctx, "apiToken")().get()

def check_source_pipeline_exists():
    data = requests.get(
        f"{WORKSPACE}/api/2.0/pipelines/{SOURCE_PIPELINE_ID}",
        headers={"Authorization": f"Bearer {get_token()}"},
    )

    assert data.json()["pipeline_id"] == SOURCE_PIPELINE_ID, "The provided source pipeline does not exist!"

def request_pipeline_clone():
    payload = {
      "catalog": TARGET_CATALOG,
      "clone_mode": CLONE_MODE,
    }
    if TARGET_SCHEMA != "":
      payload["target"] = TARGET_SCHEMA
    if CLONED_PIPELINE_NAME != "":
      payload["name"] = CLONED_PIPELINE_NAME
    if OVERRIDE_CONFIGS:
      payload["configuration"] = OVERRIDE_CONFIGS

    data = requests.post(
        f"{WORKSPACE}/api/2.0/pipelines/{SOURCE_PIPELINE_ID}/clone",
        headers={"Authorization": f"Bearer {get_token()}"},
        json=payload,
    )
    response = data.json()
    return response

check_source_pipeline_exists()
request_pipeline_clone()

Limitaciones

A continuación se muestran las limitaciones de la clone a pipeline solicitud de API:

  • La clonación de un pipeline desde el metastore de Hive a Unity Catalog no es compatible con los paquetes de automatización declarativos.
  • Solo se admite la clonación desde una canalización configurada para usar el metastore de Hive a una canalización de Unity Catalog.
  • Solo puede crear un clon en el mismo área de trabajo de Azure Databricks que la canalización desde la que está clonando.
  • La canalización que va a clonar solo puede incluir los siguientes orígenes de streaming:
    • Fuentes Delta
    • Auto Loader, incluidos los orígenes de datos admitidos por Auto Loader. Consulte Carga de archivos desde el almacenamiento de objetos en la nube.
    • Apache Kafka con Structured Streaming. Sin embargo, el origen de Kafka no se puede configurar para usar la kafka.group.id opción . Consulte Conexión a Apache Kafka.
    • Amazon Kinesis con transmisión estructurada. Sin embargo, no se puede configurar el origen de Kinesis para establecer consumerMode a efo.
  • Si la canalización de metastore de Hive que va a clonar utiliza el modo de notificación de archivos de Auto Loader, Databricks recomienda no ejecutar la canalización de metastore de Hive después de la clonación. Esto se debe a que al ejecutar la canalización del metastore de Hive se eliminan algunos eventos de notificación de archivos del clon de Unity Catalog. Si la canalización del metastore de Hive de origen se ejecuta después completarse la operación de clonación, puede reponer los archivos que falten mediante Auto Loader con la opción cloudFiles.backfillInterval. Para obtener información sobre el modo de notificación de archivos del cargador automático, consulte Configuración de flujos de cargador automático en modo de notificación de archivos. Para obtener información sobre el relleno retrospectivo de archivos con Auto Loader, consulte Activar rellenos retrospectivos periódicos mediante cloudFiles.backfillInterval y Common.
  • Las tareas de mantenimiento de canalización se pausan automáticamente para ambas canalizaciones mientras la clonación está en curso.
  • Lo siguiente se aplica a las consultas de viaje en el tiempo en las tablas de la canalización clonada de Unity Catalog:
    • Si una versión de tabla se escribió originalmente en un objeto administrado por el metastore de Hive, las consultas de viaje en el tiempo que usan una cláusula timestamp_expression no están definidas al consultar el objeto clonado de Unity Catalog.
    • Sin embargo, si la versión de la tabla se escribió en el objeto clonado de Unity Catalog, las consultas de viaje en el tiempo que utilizan una cláusula timestamp_expression funcionan correctamente.
    • Las consultas de viaje en el tiempo que utilizan una cláusula version funcionan correctamente al consultar un objeto clonado de Unity Catalog, incluso cuando la versión se escribió originalmente en el objeto administrado del metastore de Hive.
  • Para ver otras limitaciones al usar canalizaciones de Lakeflow con el catálogo de Unity, consulte Limitaciones de canalización del catálogo de Unity.
  • Para conocer las limitaciones del catálogo de Unity, consulte Introducción al catálogo de Unity.
  • Last updated on