Créer des pipelines multi-destinations

S'applique à : connecteurs SaaS connecteurs de base de données connecteurs basés sur la query

À l'aide de connecteurs d'ingestion gérés dans Lakeflow Connect, vous pouvez écrire dans plusieurs catalogues et schémas de destination à partir d'un seul pipeline. Vous pouvez également ingérer plusieurs objets dans le même schéma. Cependant, les connecteurs gérés ne prennent pas en charge les noms de table en double dans le même schéma de destination, vous devez donc spécifier un nouveau nom pour l'une des tables afin de les différencier. Consultez Nommer une table de destination.

Exemple : ingérer deux objets dans des schémas différents

Les définitions d'exemple de pipeline dans cette section montrent comment ingérer deux objets dans différents schémas, selon l'interface de création de pipeline et le système source.

Google Analytics

Declarative Automation Bundles

Voici un exemple de fichier YAML que vous pouvez utiliser dans vos bundles :

YAML

resources:
  pipelines:
    pipeline_ga4:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1> # Location of the pipeline event log
      ingestion_definition:
        connection_name: <connection-name>
        objects:
          - table:
              source_url: <project-1-id>
              source_schema: <property-name>
              destination_catalog: <target-catalog-1>
              destination_schema: <target-schema-1>
          - table:
              source_url: <project-2-id>
              source_schema: <property-name>
              destination_catalog: <target-catalog-2>
              destination_schema: <target-schema-2>

Databricks notebook

Voici un exemple de spécification de pipeline Python que vous pouvez utiliser dans votre Notebook :

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<connection-name>",
    "objects": [
      {
        "table": {
          "source_catalog": "<project-1-id>",
          "source_schema": "<property-1-name>",
          "destination_catalog": "<target-catalog-1>",
          "destination_schema": "<target-schema-1>",
        },
        "table": {
          "source_catalog": "<project-2-id>",
          "source_schema": "<property-2-name>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
        }
      }
    ]
  }
}
"""

Databricks CLI

Voici un exemple de définition de pipeline JSON que vous pouvez utiliser avec des commandes CLI :

JSON

{
  "resources": {
    "pipelines": {
      "pipeline_ga4": {
        "name": "<pipeline-name>",
        "catalog": "<target-catalog-1>",
        "schema": "<target-schema-1>",
        "ingestion_definition": {
          "connection_name": "<connection-name>",
          "objects": [
            {
              "table": {
                "source_url": "<project-1-id>",
                "source_schema": "<property-1-name>",
                "destination_catalog": "<target-catalog-1>",
                "destination_schema": "<target-schema-1>"
              },
              "table": {
                "source_url": "<project-2-id>",
                "source_schema": "<property-2-name>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>"
              }
            }
          ]
        }
      }
    }
  }
}

MySQL

Declarative Automation Bundles

Voici un exemple de fichier de ressources YAML que vous pouvez utiliser dans votre bundle :

YAML

resources:
  pipelines:
    gateway:
      name: <gateway-name>
      gateway_definition:
        connection_id: <connection-id>
        gateway_storage_catalog: <destination-catalog>
        gateway_storage_schema: <destination-schema>
        gateway_storage_name: <destination-schema>
      target: <destination-schema>
      catalog: <destination-catalog>

    pipeline_mysql:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1> # Location of the pipeline event log
      ingestion_definition:
        ingestion_gateway_id: ${resources.pipelines.gateway.id}
        objects:
          - table:
              source_schema: <source-schema-1>
              source_table: <source-table-1>
              destination_catalog: <target-catalog-1> # Location of this table
              destination_schema: <target-schema-1> # Location of this table
          - table:
              source_schema: <source-schema-2>
              source_table: <source-table-2>
              destination_catalog: <target-catalog-2> # Location of this table
              destination_schema: <target-schema-2> # Location of this table

Databricks notebook

Voici des exemples de spécifications de passerelle d'ingestion et de pipeline d'ingestion que vous pouvez utiliser dans un Notebook Python :

Python

gateway_pipeline_spec = {
  "pipeline_type": "INGESTION_GATEWAY",
  "name": <gateway-name>,
  "catalog": <destination-catalog>,
  "target": <destination-schema>,
  "gateway_definition": {
    "connection_id": <connection-id>,
    "gateway_storage_catalog": <destination-catalog>,
    "gateway_storage_schema": <destination-schema>,
    "gateway_storage_name": <destination-schema>
    }
}

ingestion_pipeline_spec = {
  "pipeline_type": "MANAGED_INGESTION",
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "ingestion_gateway_id": <gateway-pipeline-id>,
    "source_type": "MYSQL",
    "objects": [
      {
        "table": {
          "source_schema": "<source-schema-1>",
          "source_table": "<source-table-1>",
          "destination_catalog": "<destination-catalog-1>",
          "destination_schema": "<destination-schema-1>",
        },
        "table": {
          "source_catalog": "<source-catalog-2>",
          "source_schema": "<source-schema-2>",
          "source_table": "<source-table-2>",
          "destination_catalog": "<destination-catalog-2>",
          "destination_schema": "<destination-schema-2>",
        }
      }
    ]
  }
}

Databricks CLI

Pour créer la passerelle d'ingestion à l'aide de l'interface de ligne de commande Databricks :

databricks pipelines create --json '{
"name": "'"<gateway-name>"'",
"gateway_definition": {
  "connection_id": "'"<connection-id>"'",
  "gateway_storage_catalog": "'"<staging-catalog>"'",
  "gateway_storage_schema": "'"<staging-schema>"'",
  "gateway_storage_name": "'"<gateway-name>"'"
  }
}'

Pour créer le pipeline d'ingestion à l'aide de la CLI Databricks :

databricks pipelines create --json '{
"name": "'"<pipeline-name>"'",
"ingestion_definition": {
  "ingestion_gateway_id": "'"<gateway-id>"'",
  "objects": [
    {"table": {
        "source_schema": "<source-schema-1>",
        "source_table": "<source-table-1>",
        "destination_catalog": "'"<destination-catalog-1>"'",
        "destination_schema": "'"<destination-schema-1>"'"
        }},
    {"table": {
        "source_schema": "<source-schema-2>",
        "source_table": "<source-table-2>",
        "destination_catalog": "'"<destination-catalog-2>"'",
        "destination_schema": "'"<destination-schema-2>"'"
        }}
    ]
  }
}'

Salesforce

Declarative Automation Bundles

Voici un exemple de fichier YAML que vous pouvez utiliser dans vos bundles :

YAML

resources:
  pipelines:
    pipeline_sfdc:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1> # Location of the pipeline event log
      ingestion_definition:
        connection_name: <connection-name>
        objects:
          - table:
              source_schema: <source-schema-1>
              source_table: <source-table-1>
              destination_catalog: <target-catalog-1> # Location of this table
              destination_schema: <target-schema-1> # Location of this table
          - table:
              source_schema: <source-schema-2>
              source_table: <source-table-2>
              destination_catalog: <target-catalog-2> # Location of this table
              destination_schema: <target-schema-2> # Location of this table

Databricks notebook

Voici un exemple de spécification de pipeline Python que vous pouvez utiliser dans votre Notebook :

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<connection-name>",
    "objects": [
      {
        "table": {
          "source_schema": "<source-schema-1>",
          "source_table": "<source-table-1>",
          "destination_catalog": "<target-catalog-1>",
          "destination_schema": "<target-schema-1>",
        },
        "table": {
          "source_schema": "<source-schema-2>",
          "source_table": "<source-table-2>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
        }
      }
    ]
  }
}
"""

Databricks CLI

Voici un exemple de définition de pipeline JSON que vous pouvez utiliser avec des commandes CLI :

JSON

{
  "resources": {
    "pipelines": {
      "pipeline_sfdc": {
        "name": "<pipeline-name>",
        "catalog": "<target-catalog-1>",
        "schema": "<target-schema-1>",
        "ingestion_definition": {
          "connection_name": "<connection-name>",
          "objects": [
            {
              "table": {
                "source_schema": "<source-schema-1>",
                "source_table": "<source-table-1>",
                "destination_catalog": "<target-catalog-1>",
                "destination_schema": "<target-schema-1>"
              },
              "table": {
                "source_schema": "<source-schema-2>",
                "source_table": "<source-table-2>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>"
              }
            }
          ]
        }
      }
    }
  }
}

SQL Server

Declarative Automation Bundles

Voici un exemple de fichier de ressources YAML que vous pouvez utiliser dans votre bundle :

YAML

resources:
  pipelines:
    gateway:
      name: <gateway-name>
      gateway_definition:
        connection_id: <connection-id>
        gateway_storage_catalog: <destination-catalog>
        gateway_storage_schema: <destination-schema>
        gateway_storage_name: <destination-schema>
      target: <destination-schema>
      catalog: <destination-catalog>

    pipeline_sqlserver:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1> # Location of the pipeline event log
      ingestion_definition:
        connection_name: <connection-name>
        objects:
          - table:
              source_schema: <source-schema-1>
              source_table: <source-table-1>
              destination_catalog: <target-catalog-1> # Location of this table
              destination_schema: <target-schema-1> # Location of this table
          - table:
              source_schema: <source-schema-2>
              source_table: <source-table-2>
              destination_catalog: <target-catalog-2> # Location of this table
              destination_schema: <target-schema-2> # Location of this table

Databricks notebook

Voici des exemples de spécifications de passerelle d'ingestion et de pipeline d'ingestion que vous pouvez utiliser dans un Notebook Python :

Python

gateway_pipeline_spec = {
  "pipeline_type": "INGESTION_GATEWAY",
  "name": <gateway-name>,
  "catalog": <destination-catalog>,
  "target": <destination-schema>,
  "gateway_definition": {
    "connection_id": <connection-id>,
    "gateway_storage_catalog": <destination-catalog>,
    "gateway_storage_schema": <destination-schema>,
    "gateway_storage_name": <destination-schema>
    }
}

ingestion_pipeline_spec = {
  "pipeline_type": "MANAGED_INGESTION",
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "ingestion_gateway_id": <gateway-pipeline-id>,
    "source_type": "SQLSERVER",
    "objects": [
      {
        "table": {
          "source_schema": "<source-schema-1>",
          "source_table": "<source-table-1>",
          "destination_catalog": "<destination-catalog-1>",
          "destination_schema": "<destination-schema-1>",
        },
        "table": {
          "source_schema": "<source-schema-2>",
          "source_table": "<source-table-2>",
          "destination_catalog": "<destination-catalog-2>",
          "destination_schema": "<destination-schema-2>",
        }
      }
    ]
  }
}

Databricks CLI

Pour créer la passerelle d'ingestion à l'aide de l'interface de ligne de commande Databricks :

databricks pipelines create --json '{
"name": "'"<gateway-name>"'",
"gateway_definition": {
  "connection_id": "'"<connection-id>"'",
  "gateway_storage_catalog": "'"<staging-catalog>"'",
  "gateway_storage_schema": "'"<staging-schema>"'",
  "gateway_storage_name": "'"<gateway-name>"'"
  }
}'

Pour créer le pipeline d'ingestion à l'aide de la CLI Databricks :

databricks pipelines create --json '{
"name": "'"<pipeline-name>"'",
"ingestion_definition": {
  "ingestion_gateway_id": "'"<gateway-id>"'",
  "objects": [
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-1>"'",
        "destination_schema": "'"<destination-schema-1>"'"
        }},
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-2>"'",
        "destination_schema": "'"<destination-schema-2>"'"
        }}
    ]
  }
}'

Workday

Declarative Automation Bundles

Voici un exemple de fichier YAML que vous pouvez utiliser dans vos bundles :

YAML

resources:
  pipelines:
    pipeline_workday:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1> # Location of the pipeline event log
      ingestion_definition:
        connection_name: <connection-name>
        objects:
          - report:
              source_url: <report-url-1>
              destination_catalog: <target-catalog-1>
              destination_schema: <target-schema-1>
          - report:
              source_url: <report-url-2>
              destination_catalog: <target-catalog-2>
              destination_schema: <target-schema-2>

Databricks notebook

Voici un exemple de spécification de pipeline Python que vous pouvez utiliser dans votre Notebook :

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<connection-name>",
    "objects": [
      {
        "report": {
          "source_url": "<report-url-1>",
          "destination_catalog": "<target-catalog-1>",
          "destination_schema": "<target-schema-1>",
        },
        "report": {
          "source_url": "<report-url-2>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
        }
      }
    ]
  }
}
"""

Databricks CLI

Voici un exemple de définition de pipeline JSON que vous pouvez utiliser avec des commandes CLI :

JSON

{
  "resources": {
    "pipelines": {
      "pipeline_workday": {
        "name": "<pipeline-name>",
        "catalog": "<target-catalog-1>",
        "schema": "<target-schema-1>",
        "ingestion_definition": {
          "connection_name": "<connection-name>",
          "objects": [
            {
              "report": {
                "source_url": "<report-url-1>",
                "destination_catalog": "<target-catalog-1>",
                "destination_schema": "<target-schema-1>"
              },
              "report": {
                "source_url": "<report-url-2>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>"
              }
            }
          ]
        }
      }
    }
  }
}

Exemple : Ingérer un objet trois fois

La définition de pipeline exemple suivante montre comment ingérer un objet dans trois tables de destination différentes. Dans l'exemple, la troisième table cible reçoit un nom unique pour différencier l'ingestion d'un objet dans le même schéma de destination deux fois (les doublons ne sont pas pris en charge).

Google Analytics

Declarative Automation Bundles

Voici un exemple de fichier YAML que vous pouvez utiliser dans vos bundles :

YAML

resources:
  pipelines:
    pipeline_sfdc:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1>	# Location of the pipeline event log
      ingestion_definition:
        connection_name: <connection-name>
        objects:
          - table:
              source_url: <project-id>
              source_schema: <property-name>
              destination_catalog: <target-catalog-1> # Location of first copy
              destination_schema: <target-schema-1>	# Location of first copy
          - table:
              source_url: <project-id>
              source_schema: <property-name>
              destination_catalog: <target-catalog-2> # Location of second copy
              destination_schema: <target-schema-2> # Location of second copy
	      - table:
              source_url: <project-id>
              source_schema: <property-name>
              destination_catalog: <target-catalog-2> # Location of third copy
              destination_schema: <target-schema-2> # Location of third copy
              destination_table: <custom-target-table-name> # Specify destination table name

Databricks notebook

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<connection-name>",
    "objects": [
      {
        "table": {
          "source_catalog": "<project-id>",
          "source_schema": "<property-name>",
          "destination_catalog": "<target-catalog-1>",
          "destination_schema": "<target-schema-1>",
        },
        "table": {
          "source_catalog": "<project-id>",
          "source_schema": "<property-name>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
        },
        "table": {
          "source_catalog": "<project-id>",
          "source_schema": "<property-name>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
          "destination_table": "<custom-target-table-name>",
        },
      }
    ]
  }
}
"""

Databricks CLI

Voici un exemple de définition de pipeline JSON que vous pouvez utiliser avec des commandes CLI :

JSON

{
  "resources": {
    "pipelines": {
      "pipeline_ga4": {
        "name": "<pipeline-name>",
        "catalog": "<target-catalog-1>",
        "schema": "<target-schema-1>",
        "ingestion_definition": {
          "connection_name": "<connection-name>",
          "objects": [
            {
              "table": {
                "source_url": "<project-id>",
                "source_schema": "<property-name>",
                "destination_catalog": "<target-catalog-1>",
                "destination_schema": "<target-schema-1>"
              },
              "table": {
                "source_url": "<project-id>",
                "source_schema": "<property-name>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>"
              },
              "table": {
                "source_url": "<project-id>",
                "source_schema": "<property-name>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>",
                "destination_table": "<custom-target-table-name>"
              }
            }
          ]
        }
      }
    }
  }
}

MySQL

Declarative Automation Bundles

Voici un exemple de fichier de ressources YAML que vous pouvez utiliser dans votre bundle :

YAML

resources:
  pipelines:
    gateway:
      name: <gateway-name>
      gateway_definition:
        connection_id: <connection-id>
        gateway_storage_catalog: <destination-catalog>
        gateway_storage_schema: <destination-schema>
        gateway_storage_name: <destination-schema-name>
      target: <destination-schema>
      catalog: <destination-catalog>

    pipeline_mysql:
      name: <pipeline-name>
      catalog: <destination-catalog-1> # Location of the pipeline event log
      schema: <destination-schema-1> # Location of the pipeline event log
      ingestion_definition:
        ingestion_gateway_id: ${resources.pipelines.gateway.id}
        objects:
          - table:
              source_catalog: <source-catalog>
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <destination-catalog-1> # Location of first copy
              destination_schema: <destination-schema-1> # Location of first copy
          - table:
              source_catalog: <source-catalog>
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <destination-catalog-2> # Location of second copy
              destination_schema: <destination-schema-2> # Location of second copy
          - table:
              source_catalog: <source-catalog>
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <destination-catalog-2> # Location of third copy
              destination_schema: <destination-schema-2> # Location of third copy
              destination_table: <custom-destination-table-name> # Specify destination table name

Databricks notebook

Voici des exemples de spécifications de passerelle d'ingestion et de pipeline d'ingestion que vous pouvez utiliser dans un Notebook Python :

Python

gateway_pipeline_spec = {
  "pipeline_type": "INGESTION_GATEWAY",
  "name": <gateway-name>,
  "catalog": <destination-catalog>,
  "target": <destination-schema>,
  "gateway_definition": {
    "connection_id": <connection-id>,
    "gateway_storage_catalog": <destination-catalog>,
    "gateway_storage_schema": <destination-schema>,
    "gateway_storage_name": <destination-schema>
    }
}

ingestion_pipeline_spec = {
  "pipeline_type": "MANAGED_INGESTION",
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "ingestion_gateway_id": <gateway-pipeline-id>,
    "source_type": "MYSQL",
    "objects": [
      {
        "table": {
          "source_catalog": <source-catalog>,
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<destination-catalog-1>",
          "destination_schema": "<destination-schema-1>",
        },
        "table": {
          "source_catalog": <source-catalog>,
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<destination-catalog-2>",
          "destination_schema": "<destination-schema-2>",
        },
        "table": {
          "source_catalog": <source-catalog>,
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<destination-catalog-2>",
          "destination_schema": "<destination-schema-2>",
          "destination_table": "<custom-destination-table-name>",
        }
      }
    ]
  }
}

Databricks CLI

Pour créer la passerelle d'ingestion à l'aide de l'interface de ligne de commande Databricks :

databricks pipelines create --json '{
"name": "'"<gateway-name>"'",
"gateway_definition": {
  "connection_id": "'"<connection-id>"'",
  "gateway_storage_catalog": "'"<staging-catalog>"'",
  "gateway_storage_schema": "'"<staging-schema>"'",
  "gateway_storage_name": "'"<gateway-name>"'"
  }
}'

Pour créer le pipeline d'ingestion à l'aide de la CLI Databricks :

databricks pipelines create --json '{
"name": "'"<pipeline-name>"'",
"ingestion_definition": {
  "ingestion_gateway_id": "'"<gateway-id>"'",
  "objects": [
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-1>"'",
        "destination_schema": "'"<target-schema-1>"'"
        }},
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-2>"'",
        "destination_schema": "'"<target-schema-2>"'"
        }},
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-2>"'",
        "destination_schema": "'"<target-schema-2>"'",
        "destination_table": "<custom-destination-table-name>"
        }}
    ]
  }
}'

Salesforce

Declarative Automation Bundles

Voici un exemple de fichier YAML que vous pouvez utiliser dans vos bundles :

YAML

resources:
  pipelines:
    pipeline_sfdc:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1>	# Location of the pipeline event log
      ingestion_definition:
        connection_name: <connection-name>
        objects:
          - table:
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <target-catalog-1> # Location of first copy
              destination_schema: <target-schema-1>	# Location of first copy
          - table:
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <target-catalog-2> # Location of second copy
              destination_schema: <target-schema-2> # Location of second copy
	      - table:
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <target-catalog-2> # Location of third copy
              destination_schema: <target-schema-2> # Location of third copy
              destination_table: <custom-target-table-name> # Specify destination table name

Databricks notebook

Voici un exemple de spécification de pipeline Python que vous pouvez utiliser dans votre Notebook :

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<connection-name>",
    "objects": [
      {
        "table": {
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<target-catalog-1>",
          "destination_schema": "<target-schema-1>",
        },
        "table": {
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
        },
        "table": {
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
          "destination_table": "<custom-target-table-name>",
        }
      }
    ]
  }
}
"""

Databricks CLI

Voici un exemple de définition de pipeline JSON que vous pouvez utiliser avec des commandes CLI :

JSON

{
  "resources": {
    "pipelines": {
      "pipeline_sfdc": {
        "name": "<pipeline-name>",
        "catalog": "<target-catalog-1>",
        "schema": "<target-schema-1>",
        "ingestion_definition": {
          "connection_name": "<connection-name>",
          "objects": [
            {
              "table": {
                "source_schema": "<source-schema>",
                "source_table": "<source-table>",
                "destination_catalog": "<target-catalog-1>",
                "destination_schema": "<target-schema-1>"
              },
              "table": {
                "source_schema": "<source-schema>",
                "source_table": "<source-table>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>"
              },
              "table": {
                "source_schema": "<source-schema>",
                "source_table": "<source-table>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>",
                "destination_table": "<custom-target-table-name>"
              }
            }
          ]
        }
      }
    }
  }
}

SQL Server

Declarative Automation Bundles

Voici un exemple de fichier de ressources YAML que vous pouvez utiliser dans votre bundle :

YAML

resources:
  pipelines:

    gateway:
      name: <gateway-name>
      gateway_definition:
        connection_id: <connection-id>
        gateway_storage_catalog: <destination-catalog>
        gateway_storage_schema: <destination-schema>
        gateway_storage_name: <destination-schema-name>
      target: <destination-schema>
      catalog: <destination-catalog>

    pipeline_sqlserver:
      name: <pipeline-name>
      catalog: <destination-catalog-1> # Location of the pipeline event log
      schema: <destination-schema-1>	# Location of the pipeline event log
      ingestion_definition:
        ingestion_gateway_id: <gateway-id>
        objects:
          - table:
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <destination-catalog-1> # Location of first copy
              destination_schema: <destination-schema-1>	# Location of first copy
          - table:
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <destination-catalog-2> # Location of second copy
              destination_schema: <destination-schema-2> # Location of second copy
        - table:
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <destination-catalog-2> # Location of third copy
              destination_schema: <destination-schema-2> # Location of third copy
              destination_table: <custom-destination-table-name> # Specify destination table name

Databricks notebook

Voici des exemples de spécifications de passerelle d'ingestion et de pipeline d'ingestion que vous pouvez utiliser dans un Notebook Python :

Python

gateway_pipeline_spec = {
  "pipeline_type": "INGESTION_GATEWAY",
  "name": <gateway-name>,
  "catalog": <destination-catalog>,
  "target": <destination-schema>,
  "gateway_definition": {
    "connection_id": <connection-id>,
    "gateway_storage_catalog": <destination-catalog>,
    "gateway_storage_schema": <destination-schema>,
    "gateway_storage_name": <destination-schema>
    }
}

ingestion_pipeline_spec = {
  "pipeline_type": "MANAGED_INGESTION",
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "ingestion_gateway_id": <gateway-pipeline-id>,
    "source_type": "SQLSERVER",
    "objects": [
      {
        "table": {
          "source_catalog": <source-catalog>,
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<destination-catalog-1>",
          "destination_schema": "<destination-schema-1>",
        },
        "table": {
          "source_catalog": <source-catalog>,
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<destination-catalog-2>",
          "destination_schema": "<destination-schema-2>",
        },
        "table": {
          "source_catalog": <source-catalog>,
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<destination-catalog-2>",
          "destination_schema": "<destination-schema-2>",
          "destination_table": "<custom-destination-table-name>",
        }
      }
    ]
  }
}

Databricks CLI

Pour créer la passerelle d'ingestion à l'aide de l'interface de ligne de commande Databricks :

databricks pipelines create --json '{
"name": "'"<gateway-name>"'",
"gateway_definition": {
  "connection_id": "'"<connection-id>"'",
  "gateway_storage_catalog": "'"<staging-catalog>"'",
  "gateway_storage_schema": "'"<staging-schema>"'",
  "gateway_storage_name": "'"<gateway-name>"'"
  }
}'

Pour créer le pipeline d'ingestion à l'aide de la CLI Databricks :

databricks pipelines create --json '{
"name": "'"<pipeline-name>"'",
"ingestion_definition": {
  "ingestion_gateway_id": "'"<gateway-id>"'",
  "objects": [
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-1>"'",
        "destination_schema": "'"<target-schema-1>"'"
        }},
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-2>"'",
        "destination_schema": "'"<target-schema-2>"'"
        }},
    {"table": {
        "source_catalog": "<source-catalog>",
        "source_schema": "<source-schema>",
        "source_table": "<source-table>",
        "destination_catalog": "'"<destination-catalog-2>"'",
        "destination_schema": "'"<target-schema-2>"'",
        "destination_table": "<custom-destination-table-name>"
        }}
    ]
  }
}'

Workday

Declarative Automation Bundles

Voici un exemple de fichier YAML que vous pouvez utiliser dans vos bundles :

YAML

resources:
  pipelines:
    pipeline_sfdc:
      name: <pipeline-name>
      catalog: <target-catalog-1> # Location of the pipeline event log
      schema: <target-schema-1>	# Location of the pipeline event log
      ingestion_definition:
        connection_name: <connection-name>
        objects:
          - report:
              source_url: <report-url>
              destination_catalog: <target-catalog-1> # Location of first copy
              destination_schema: <target-schema-1>	# Location of first copy
          - report:
              source_url: <report-url>
              destination_catalog: <target-catalog-2> # Location of second copy
              destination_schema: <target-schema-2> # Location of second copy
	      - report:
              source_url: <report-url>
              destination_catalog: <target-catalog-2> # Location of third copy
              destination_schema: <target-schema-2> # Location of third copy
              destination_table: <custom-target-table-name> # Specify destination table name

Databricks notebook

Voici un exemple de spécification de pipeline Python que vous pouvez utiliser dans votre Notebook :

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<connection-name>",
    "objects": [
      {
        "report": {
          "source_url": "<report-url>",
          "destination_catalog": "<target-catalog-1>",
          "destination_schema": "<target-schema-1>",
        },
        "report": {
          "source_url": "<report-url>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
        },
        "report": {
          "source_url": "<report-url>",
          "destination_catalog": "<target-catalog-2>",
          "destination_schema": "<target-schema-2>",
          "destination_table": "<custom-target-table-name>",
        }
      }
    ]
  }
}
"""

Databricks CLI

Voici un exemple de définition de pipeline JSON que vous pouvez utiliser avec des commandes CLI :

JSON

{
  "resources": {
    "pipelines": {
      "pipeline_workday": {
        "name": "<pipeline-name>",
        "catalog": "<target-catalog-1>",
        "schema": "<target-schema-1>",
        "ingestion_definition": {
          "connection_name": "<connection-name>",
          "objects": [
            {
              "report": {
                "source_url": "<report-url>",
                "destination_catalog": "<target-catalog-1>",
                "destination_schema": "<target-schema-1>"
              },
              "report": {
                "source_url": "<report-url>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>"
              },
              "report": {
                "source_url": "<report-url>",
                "destination_catalog": "<target-catalog-2>",
                "destination_schema": "<target-schema-2>",
                "destination_table": "<custom-target-table-name>"
              }
            }
          ]
        }
      }
    }
  }
}