Delen via

Tabellen verplaatsen tussen pijplijnen

In dit artikel wordt beschreven hoe u streamingtabellen en gerealiseerde weergaven tussen pijplijnen verplaatst. Na de verplaatsing is het de pijplijn naar waar u de stroom verplaatst die de tabel bijwerkt, in plaats van de oorspronkelijke pijplijn. Dit is handig in veel scenario's, waaronder:

  • Splits een grote pijplijn in kleinere pijplijnen.
  • Voeg meerdere pijplijnen samen in één grotere pijplijn.
  • Wijzig de vernieuwingsfrequentie voor sommige tabellen in een pijplijn.
  • Verplaats tabellen van een pijplijn die gebruikmaakt van de verouderde publicatiemodus naar de standaardpublicatiemodus. Zie voor details over de verouderde publicatiemodus Verouderde publicatiemodus voor pijplijnen. Zie De standaardpublicatiemodus inschakelen in een pijplijn om te zien hoe u de publicatiemodus voor een hele pijplijn tegelijk kunt migreren.
  • Tabellen verplaatsen tussen pijplijnen in verschillende werkruimten.

Requirements

Hier volgen de vereisten voor het verplaatsen van een tabel tussen pijplijnen.

  • U moet Databricks Runtime 16.3 of hoger gebruiken bij het uitvoeren van de ALTER ... opdracht en Databricks Runtime 17.2 voor verplaatsing tussen werkruimtetabellen.

  • Zowel bron- als doelpijplijnen moeten zijn:

    • Eigendom van het Azure Databricks-gebruikersaccount of de service-principal die de bewerking uitvoert
    • In werkruimten die een metastore delen. Als u de metastore wilt controleren, zie current_metastore functie.

  • De doelpijplijn moet de standaardpublicatiemodus gebruiken. Hiermee kunt u tabellen publiceren naar meerdere catalogi en schema's.

    Afwisselend moeten beide pijplijnen de verouderde publicatiemodus gebruiken en moeten beide dezelfde catalogus en doelwaarde hebben in de instellingen. Zie LIVE-schema (verouderd) voor informatie over de verouderde publicatiemodus.

Opmerking

Deze functie biedt geen ondersteuning voor het verplaatsen van een pijplijn met behulp van de standaardpublicatiemodus naar een pijplijn met behulp van de verouderde publicatiemodus.

Een tabel verplaatsen tussen pijplijnen

In de volgende instructies wordt beschreven hoe u een streamingtabel of gerealiseerde weergave van de ene pijplijn naar de andere verplaatst.

  1. Stop de bronpijplijn als deze actief is. Wacht totdat het volledig stopt.

  2. Verwijder de definitie van de tabel uit de code van de bronpijplijn en sla deze ergens op voor toekomstig gebruik.

    Voeg ondersteunende query's of code toe die nodig zijn voor een correcte uitvoering van de pijplijn.

  3. Voer vanuit een notebook of een SQL-editor de volgende SQL-opdracht uit om de tabel van de bronpijplijn opnieuw toe te laten aan de doelpijplijn:

    ALTER [MATERIALIZED VIEW | STREAMING TABLE | TABLE] <table-name>
  SET TBLPROPERTIES("pipelines.pipelineId"="<destination-pipeline-id>");

    Houd er rekening mee dat de SQL-opdracht moet worden uitgevoerd vanuit de werkruimte van de bronpijplijn.

    De opdracht gebruikt ALTER MATERIALIZED VIEW voor Unity Catalog beheerde gerealiseerde weergaven en ALTER STREAMING TABLE voor streamingtabellen. Als u dezelfde actie wilt uitvoeren op een Hive-metastore-tabel, gebruikt u ALTER TABLE.

    Als u bijvoorbeeld een streamingtabel met de naam sales wilt verplaatsen naar een pijplijn met de id abcd1234-ef56-ab78-cd90-1234efab5678, voert u de volgende opdracht uit:

    ALTER STREAMING TABLE sales
  SET TBLPROPERTIES("pipelines.pipelineId"="abcd1234-ef56-ab78-cd90-1234efab5678");

    Opmerking

    De pipelineId moet een geldige pijpleiding-identificatie zijn. De null waarde is niet toegestaan.

  4. Voeg de definitie van de tabel toe aan de code van de doelpijplijn.

    Opmerking

    Als de catalogus of het doelschema verschilt tussen de bron en het doel, werkt het kopiëren van de query mogelijk niet. Gedeeltelijk gekwalificeerde tabellen in de definitie kunnen dit op een andere manier oplossen. Mogelijk moet u de definitie bijwerken terwijl u overgaat naar de volledige kwalificatie van de tabelnamen.

    Opmerking

    Verwijder of maak een opmerking bij toevoeg zodra stromen (in Python, query's met append_flow(once=True), in SQL, query's met INSERT INTO ONCE) uit de code van de doelpijplijn. Zie Beperkingen voor meer informatie.

De verplaatsing is voltooid. U kunt nu zowel de bron- als de doelpijplijnen uitvoeren. De bestemmingspijplijn werkt de tabel bij.

Probleemoplossingsproces

In de volgende tabel worden fouten beschreven die kunnen optreden bij het verplaatsen van een tabel tussen pijplijnen.

Fout Description
DESTINATION_PIPELINE_NOT_IN_DIRECT_PUBLISHING_MODE De bronpijplijn bevindt zich in de standaardpublicatiemodus en de bestemming maakt gebruik van de modus LIVE-schema (verouderd). Dit wordt niet ondersteund. Als de bron gebruikmaakt van de standaardpublicatiemodus, moet de bestemming ook.
PIPELINE_TYPE_NOT_WORKSPACE_PIPELINE_TYPE Alleen het verplaatsen van tabellen tussen pijplijnen wordt ondersteund. Het verplaatsen van streamingtabellen en gerealiseerde weergaven die zijn gemaakt met Databricks SQL, worden niet ondersteund.
DESTINATION_PIPELINE_NOT_FOUND Dit pipelines.pipelineId moet een geldige pijplijn zijn. De pipelineId waarde kan niet null zijn.
De tabel wordt niet bijgewerkt in de doelbestemming na het verplaatsen. Als u dit in dit geval snel wilt verhelpen, verplaatst u de tabel terug naar de bronpijplijn volgens dezelfde instructies.
PIPELINE_PERMISSION_DENIED_NOT_OWNER Zowel de bron- als doelpijplijnen moeten eigendom zijn van de gebruiker die de verplaatsingsbewerking uitvoert.
TABLE_ALREADY_EXISTS De tabel in het foutbericht bestaat al. Dit kan gebeuren als er al een back-uptabel voor de pijplijn bestaat. In dit geval wordt DROP de tabel vermeld in de fout.

Voorbeeld met meerdere tabellen in een pijplijn

Pijplijnen kunnen meer dan één tabel bevatten. U kunt nog steeds één tabel tegelijk tussen pijplijnen verplaatsen. In dit scenario zijn er drie tabellen (table_a, table_b, table_c) die sequentieel van elkaar in de bronpijplijn worden gelezen. We willen één tabel table_bnaar een andere pijplijn verplaatsen.

Initiële broncode van de pijplijn

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.table
def table_a():
    return spark.read.table("source_table")

# Table to be moved to new pipeline:
@dp.table
def table_b():
    return (
        spark.read.table("table_a")
        .select(col("column1"), col("column2"))
    )

@dp.table
def table_c():
    return (
        spark.read.table("table_b")
        .groupBy(col("column1"))
        .agg(sum("column2").alias("sum_column2"))
    )

We verplaatsen table_b naar een andere pijplijn door eerst de tabeldefinitie uit de bron te kopiëren en vervolgens te verwijderen, en daarna de pipelineId van table_b bij te werken.

Pauzeer eerst alle planningen en wacht tot updates zijn voltooid op zowel de bron- als doelpijplijnen. Wijzig vervolgens de bronpijplijn om code te verwijderen voor de tabel die wordt verplaatst. De bijgewerkte voorbeeldcode van de bron-pijplijn wordt:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.table
def table_a():
    return spark.read.table("source_table")

# Removed, to be in new pipeline:
# @dp.table
# def table_b():
#     return (
#         spark.read.table("table_a")
#         .select(col("column1"), col("column2"))
#     )

@dp.table
def table_c():
    return (
        spark.read.table("table_b")
        .groupBy(col("column1"))
        .agg(sum("column2").alias("sum_column2"))
    )

Ga naar de SQL-editor om de ALTER pipelineId opdracht uit te voeren.

ALTER MATERIALIZED VIEW table_b
  SET TBLPROPERTIES("pipelines.pipelineId"="<new-pipeline-id>");

Vervolgens gaat u naar de doelpijplijn en voegt u de definitie van table_b toe. Als de standaardcatalogus en het standaardschema hetzelfde zijn in de pijplijninstellingen, zijn er geen codewijzigingen vereist.

De doelpijplijncode:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.table(name="table_b")
def table_b():
    return (
        spark.read.table("table_a")
        .select(col("column1"), col("column2"))
    )

Als de standaardcatalogus en het standaardschema verschillen in de pijplijninstellingen, moet u de volledig gekwalificeerde naam toevoegen met behulp van de catalogus en het schema van de pijplijn.

De doelpijplijncode kan bijvoorbeeld het volgende zijn:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.table(name="source_catalog.source_schema.table_b")
def table_b():
    return (
        spark.read.table("source_catalog.source_schema.table_a")
        .select(col("column1"), col("column2"))
    )

Voer planningen uit (of schakel ze opnieuw in) voor zowel de bron- als doelpijplijnen.

De pijplijnen zijn nu niet aaneengesloten. De query voor table_c leest van table_b (nu in de doelpijplijn) en table_b leest van table_a (in de bronpijplijn). Wanneer u een geactiveerde uitvoering op de bronpijplijn table_b uitvoert, wordt deze niet bijgewerkt omdat deze niet meer wordt beheerd door de bronpijplijn. De bronpijplijn beschouwt table_b als een tabel buiten de pijplijn. Dit is vergelijkbaar met het definiëren van een gerealiseerde weergave die wordt gelezen vanuit een Delta-tabel in Unity Catalog die niet wordt beheerd door de pijplijn.

Beperkingen

Hier volgen beperkingen voor het verplaatsen van tabellen tussen pijplijnen.

  • Gerealiseerde weergaven en streamingtabellen die zijn gemaakt met Databricks SQL, worden niet ondersteund.
  • De eenmaal toevoegen flows - Python append_flow(once=True) flows en SQL INSERT INTO ONCE flows - worden niet ondersteund. Hun uitvoeringsstatussen worden niet behouden; ze kunnen mogelijk opnieuw worden uitgevoerd in de doelpijplijn. Verwijder of markeer de 'append once' stromen in de doelpijplijn om te voorkomen dat deze stromen opnieuw worden uitgevoerd.
  • Privétabellen of weergaven worden niet ondersteund.
  • De bron- en doelpijplijnen moeten pijplijnen zijn. Null-pijplijnen worden niet ondersteund.
  • Bron- en doelpijplijnen moeten zich in dezelfde werkruimte bevinden of in verschillende werkruimten die dezelfde metastore delen.
  • Zowel bron- als doelpijplijnen moeten eigendom zijn van de gebruiker die de verplaatsingsbewerking uitvoert.
  • Als de bronpijplijn gebruikmaakt van de standaardpublicatiemodus, moet de doelpijplijn ook de standaardpublicatiemodus gebruiken. U kunt een tabel niet verplaatsen van een pijplijn met behulp van de standaardpublicatiemodus naar een pijplijn die gebruikmaakt van het LIVE-schema (verouderd). Zie LIVE-schema (verouderd).
  • Als de bron- en doelpijplijnen beide gebruikmaken van het LIVE-schema (verouderd), moeten ze dezelfde catalog- en target-waarden hebben in hun instellingen.
  • Last updated on