파이프라인 간에 테이블을 이동하세요

이 문서에서는 파이프라인 간에 스트리밍 테이블과 구체화된 뷰를 이동하는 방법을 설명합니다. 이동 후 흐름을 이동하는 파이프라인은 원래 파이프라인이 아닌 테이블을 업데이트합니다. 이 기능은 다음을 비롯한 많은 시나리오에서 유용합니다.

• 큰 파이프라인을 더 작은 파이프라인으로 분할합니다.
• 여러 파이프라인을 더 큰 단일 파이프라인에 병합합니다.
• 파이프라인의 일부 테이블에 대한 새로 고침 빈도를 변경합니다.
• 레거시 게시 모드를 사용하는 파이프라인에서 기본 게시 모드로 테이블을 이동합니다. 레거시 게시 모드에 대한 자세한 내용은 파이프라인에 대한 레거시 게시 모드를 참조하세요. 전체 파이프라인의 게시 모드를 한 번에 마이그레이션하는 방법을 보려면 파이프라인 에서 기본 게시 모드 사용을 참조하세요.
• 여러 작업 영역의 파이프라인 간에 테이블을 이동합니다.

요구 사항

파이프라인 간에 테이블을 이동하기 위한 요구 사항은 다음과 같습니다.

• 명령을 실행할 ALTER ... 때는 Databricks Runtime 16.3 이상을 사용하고, 작업 영역 간 테이블 이동에는 Databricks Runtime 17.2를 사용해야 합니다.

• 원본 및 대상 파이프라인은 모두 다음이어야 합니다.

  • 작업을 실행하는 Azure Databricks 사용자 계정 또는 서비스 주체가 소유
  • 메타스토어를 공유하는 작업 영역에서 메타스토어를 확인하려면 함수를 참조 current_metastore 하세요.

• 대상 파이프라인은 기본 게시 모드를 사용해야 합니다. 이렇게 하면 여러 카탈로그 및 스키마에 테이블을 게시할 수 있습니다.

  또는 두 파이프라인 모두 레거시 게시 모드를 사용해야 하며 둘 다 설정에서 카탈로그와 대상 값이 같아야 합니다. 레거시 게시 모드에 대한 자세한 내용은 LIVE 스키마(레거시)를 참조하세요.

파이프라인 간에 테이블을 이동하기

다음 지침에서는 스트리밍 테이블 또는 구체화된 뷰를 한 파이프라인에서 다른 파이프라인으로 이동하는 방법을 설명합니다.

1. 실행 중인 경우 원본 파이프라인을 중지합니다. 완전히 중지 될 때까지 기다립니다.

2. 원본 파이프라인의 코드에서 테이블 정의를 제거하고 나중에 참조할 수 있도록 어딘가에 저장합니다.

   파이프라인이 올바르게 실행되는 데 필요한 지원 쿼리 또는 코드를 포함합니다.

3. Notebook 또는 SQL 편집기에서 다음 SQL 명령을 실행하여 원본 파이프라인에서 대상 파이프라인으로 테이블을 다시 할당합니다.

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

   SQL 명령은 원본 파이프라인의 작업 영역에서 실행해야 합니다.

   이 명령은 `ALTER MATERIALIZED VIEW`를 Unity 카탈로그에서 관리하는 구체화 뷰에, `ALTER STREAMING TABLE`를 스트리밍 테이블에 각각 사용합니다. Hive 메타스토어 테이블에서 동일한 작업을 수행하려면 .를 사용합니다 ALTER TABLE.

   예를 들어 IDsales가 있는 파이프라인으로 명명된 abcd1234-ef56-ab78-cd90-1234efab5678 스트리밍 테이블을 이동하려면 다음 명령을 실행합니다.

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

   비고

   pipelineId 유효한 파이프라인 식별자여야 합니다. 값은 null 허용되지 않습니다.

4. 대상 파이프라인의 코드에 테이블의 정의를 추가합니다.

   비고

   카탈로그 또는 대상 스키마가 원본과 대상 간에 다른 경우 쿼리를 정확히 복사하지 못할 수 있습니다. 정의에서 부분적으로 정규화된 테이블은 다르게 처리될 수 있습니다. 테이블 이름을 정규화하기 위해 이동하는 동안 정의를 업데이트해야 할 수 있습니다.

   비고

   대상 파이프라인의 코드에서 추가 흐름(Python의 경우 append_flow(once=True)이 있는 쿼리, SQL의 경우 INTO ONCE를 사용한INSERT 쿼리)를 제거하거나 주석 처리합니다. 자세한 내용은 제한 사항을 참조하세요.

이동이 완료되었습니다. 이제 원본 및 대상 파이프라인을 모두 실행할 수 있습니다. 대상 파이프라인이 테이블을 업데이트합니다.

Troubleshooting

다음 표에서는 파이프라인 간에 테이블을 이동할 때 발생할 수 있는 오류를 설명합니다.

파이프라인에 여러 테이블이 있는 예제

파이프라인에는 둘 이상의 테이블이 포함될 수 있습니다. 파이프라인 간에 한 번에 하나의 테이블을 이동할 수 있습니다. 이 시나리오에는 원본 파이프라인에서 서로 순차적으로 읽는 세 개의 테이블(table_a, table_b, table_c)이 있습니다. 한 테이블을 table_b다른 파이프라인으로 이동하려고 합니다.

초기 소스 파이프라인 코드:

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"))
    )

원본에서 테이블 정의를 복사하고 제거한 후, table_b의 pipelineId를 업데이트하여 table_b을 다른 파이프라인으로 이동합니다.

먼저 일정을 일시 중지하고 원본 및 대상 파이프라인 모두에서 업데이트가 완료되기를 기다립니다. 그런 다음 원본 파이프라인을 수정하여 이동 중인 테이블에 대한 코드를 제거합니다. 업데이트된 소스 파이프라인 예제 코드는 다음과 같습니다.

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"))
    )

SQL 편집기로 이동하여 명령을 실행합니다 ALTER pipelineId .

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

다음으로 대상 파이프라인으로 이동하여 .의 정의를 추가합니다 table_b. 파이프라인 설정에서 기본 카탈로그와 스키마가 동일한 경우 코드 변경이 필요하지 않습니다.

대상 파이프라인 코드:

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"))
    )

기본 카탈로그와 스키마가 파이프라인 설정에서 다른 경우 파이프라인의 카탈로그 및 스키마를 사용하여 정규화된 이름을 추가해야 합니다.

예를 들어 대상 파이프라인 코드는 다음과 같습니다.

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"))
    )

원본 파이프라인 및 대상 파이프라인 모두에 대해 실행하거나 일정을 다시 사용할 수 있도록 설정합니다.

이제 파이프라인이 분리되었습니다. table_c에 대한 쿼리는 table_b에서 읽고(현재 대상 파이프라인) table_b는 table_a에서 읽습니다(원본 파이프라인). 트리거된 실행을 수행하는 경우 원본 파이프라인 table_b 에서 더 이상 관리되지 않으므로 원본 파이프라인에서 트리거된 실행이 업데이트되지 않습니다. 소스 파이프라인은 table_b를 파이프라인 외부의 테이블로 취급합니다. 이는 파이프라인에서 관리되지 않는 Unity 카탈로그의 델타 테이블에서 읽는 구체화된 뷰를 정의하는 것과 비슷합니다.

제한 사항

다음은 파이프라인 간에 테이블을 이동하기 위한 제한 사항입니다.

• Databricks SQL을 사용하여 만든 구체화된 뷰 및 스트리밍 테이블은 지원되지 않습니다.
• Python append_flow(once=True) 흐름 및 SQL INSERT INTO ONCE 흐름과 같은 한 번 추가 흐름은 지원되지 않습니다. 해당 실행 상태는 유지되지 않으며 대상 파이프라인에서 다시 실행될 수 있습니다. 대상 파이프라인에서 "한 번 추가 흐름"을 제거하거나 주석 처리하여 이러한 흐름이 다시 실행되지 않도록 합니다.
• 프라이빗 테이블 또는 뷰는 지원되지 않습니다.
• 원본 및 대상 파이프라인은 파이프라인이어야 합니다. Null 파이프라인은 지원되지 않습니다.
• 원본 및 대상 파이프라인은 동일한 작업 영역 또는 동일한 메타스토어를 공유하는 다른 작업 영역에 있어야 합니다.
• 원본 및 대상 파이프라인은 모두 이동 작업을 실행하는 사용자가 소유해야 합니다.
• 원본 파이프라인에서 기본 게시 모드를 사용하는 경우 대상 파이프라인도 기본 게시 모드를 사용해야 합니다. 기본 게시 모드를 사용하여 파이프라인에서 LIVE 스키마(레거시)를 사용하는 파이프라인으로 테이블을 이동할 수 없습니다. LIVE 스키마(레거시)를 참조하세요.
• 원본 및 대상 파이프라인이 모두 LIVE 스키마(레거시)를 사용하는 경우 설정에 동일한 catalog 값과 target 값이 있어야 합니다.