Lakebase 변경 데이터 피드

Lakebase 변경 데이터 피드란?

Lakebase는 기본 CDF(변경 데이터 피드)를 도입하여 다운스트림 파이프라인, 모델 및 애플리케이션에 대한 운영 데이터를 잠금 해제합니다. Lakebase Postgres 테이블의 모든 삽입, 업데이트 및 삭제는 미리 쓰기 로그에서 캡처되고 Unity 카탈로그 관리 델타 테이블에 새 행으로 저장되며 최대 15초마다 일괄 처리되고 플러시됩니다. 변경 기록은 모든 컴퓨팅 엔진에서 읽을 수 있는 열린 형식으로 저장됩니다.

대상 테이블은 델타 변경 데이터 피드와 동일한 모양을 따릅니다. 각 행에는 LSN, 트랜잭션 ID 및 타임스탬프가 포함됩니다 _pg_change_type. 운영 변경 사항은 별도의 외부 CDC 스택을 구축하지 않고도 ETL, 감사 및 다운스트림 시스템과 소비자를 위한 주요 데이터 소스가 됩니다.

사용 사례

Lakebase CDF는 다운스트림 파이프라인 및 애플리케이션이 변경에 대응할 수 있도록 운영 데이터를 레이크하우스에 제공합니다.

미리 보기 활성화

작업 영역 관리자는 작업 영역 미리 보기 페이지에서Lakebase 변경 데이터 피드 미리 보기를 사용하도록 설정해야 합니다.

Requirements

• Lakebase 자동 크기 조정: Postgres 17 을 실행하는 Lakebase 자동 크기 조정 프로젝트 입니다.
• 원본 데이터베이스: 테이블은 Lakebase의 databricks_postgres 데이터베이스에 있어야 합니다. 각 프로젝트는 이 기본 데이터베이스를 사용하여 만들어집니다. 이것은 알려진 제한 사항입니다.
• Unity 카탈로그: CDF를 구성하는 ID에는 대상 카탈로그 및 스키마에 대한 USE CATALOG, USE SCHEMA, CREATE TABLE 권한이 필요합니다. 개체에 대한 권한 부여를 참조하세요.
• 기본 스토리지: 기본 스토리지로 구성된 대상 카탈로그는 지원되지 않습니다.
• Lakebase 프로젝트: Postgres 역할에는 Lakebase 프로젝트에 대한 CAN MANAGE 권한이 필요합니다. 프로젝트 소유자에게는 기본적으로 CAN MANAGE 권한이 있습니다. 프로젝트 권한 관리를 참조하세요.
• 데이터 형식:데이터 형식 매핑을 참조하세요. 직접 델타에 해당하는 형식이 없는 형식은 STRING으로 저장됩니다.

Lakebase CDF 설정

시작하려면 피드에서 원하는 테이블에서 전체 복제본 ID를 설정한 다음(1단계) Lakebase 앱에서 CDF를 시작합니다(2단계). 데이터는 사용자가 선택한 Unity Catalog의 카탈로그와 스키마에 lb_<table_name>_history Delta 테이블로 표시됩니다.

1단계: 복제본 ID 전체 설정

Lakebase 테이블이 CDF에 참여하려면 REPLICA IDENTITY FULL이 설정되어 있어야 합니다. 기본적으로 Postgres는 행이 업데이트되거나 삭제될 때 기본 키만 기록합니다. 전체 ID를 설정하면 Postgres가 미리 쓰기 로그에 이전 및 이후 행 상태를 모두 기록하도록 지시하며, CDF는 전체 변경 기록을 작성해야 합니다.

Lakebase SQL 편집기 또는 Postgres 클라이언트에서 이러한 명령을 실행할 수 있습니다.

단일 테이블

ALTER TABLE <table_name> REPLICA IDENTITY FULL;

스키마의 모든 기존 테이블

스키마의 모든 기존 테이블(public 이 예제)에서 복제본 ID를 설정하려면 다음을 실행합니다.

DO $$
DECLARE r record;
BEGIN
  FOR r IN
    SELECT table_schema, table_name
    FROM information_schema.tables
    WHERE table_schema = 'public'
      AND table_type = 'BASE TABLE'
  LOOP
    EXECUTE format(
      'ALTER TABLE %I.%I REPLICA IDENTITY FULL;',
      r.table_schema, r.table_name
    );
  END LOOP;
END $$;

이후 테이블에 자동 적용

새로 만든 모든 테이블이 자동으로 수신 REPLICA IDENTITY FULL되도록 하려면 Postgres 이벤트 트리거를 설치합니다. CREATE TABLE가 실행될 때마다 그 후에 실행되며 새 테이블에 IDENTITY를 설정합니다.

CREATE OR REPLACE FUNCTION public.set_full_replica_identity()
RETURNS event_trigger
LANGUAGE plpgsql
AS $$
DECLARE
  obj record;
BEGIN
  FOR obj IN
    SELECT * FROM pg_event_trigger_ddl_commands()
    WHERE command_tag = 'CREATE TABLE'
  LOOP
    EXECUTE format(
      'ALTER TABLE %s REPLICA IDENTITY FULL;',
      obj.object_identity
    );
  END LOOP;
END $$;

CREATE EVENT TRIGGER set_full_replica_identity_on_create
ON ddl_command_end
WHEN TAG IN ('CREATE TABLE')
EXECUTE FUNCTION public.set_full_replica_identity();

이벤트 트리거를 이전 탭의 루프와 결합하여 한 설정에서 기존 테이블과 이후 테이블을 모두 처리합니다.

SELECT n.nspname AS table_schema,
       c.relname AS table_name,
       CASE c.relreplident
         WHEN 'd' THEN 'default'
         WHEN 'n' THEN 'nothing'
         WHEN 'f' THEN 'full'
         WHEN 'i' THEN 'index'
       END AS replica_identity
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind = 'r'
  AND n.nspname = 'public'
ORDER BY n.nspname, c.relname;

2단계: 변경 데이터 피드 시작

Lakebase CDF는 스키마 수준에서 구성됩니다. 시작되면 원본 스키마의 모든 현재 및 이후 테이블이 피드에 포함됩니다.

1. Azure Databricks 작업 영역의 앱 전환기(오른쪽 위)에서 Lakebase Postgres 엽니다.
2. Lakebase 프로젝트와 사용하려는 분기(예: 프로덕션 또는 주)를 선택합니다.
3. 분기 개요를 열고 데이터 피드 변경 탭을 클릭합니다.
4. 시작을 클릭합니다.
5. 구성 대화 상자에서 다음을 수행합니다.
   • 데이터베이스: 기본값은 databricks_postgres입니다.
   • 스키마: 원본 Postgres 스키마를 선택합니다.
   • 대상 카탈로그: 대상 Unity Catalog 카탈로그를 선택합니다.
   • 스키마: 대상 Unity 카탈로그 스키마를 선택합니다.
6. 시작을 클릭하여 피드를 시작합니다.

테이블은 대상에 다음과 같이 lb_<table_name>_history표시됩니다. 이를 찾으려면 사이드바에서 카탈로그 를 열고 대상 카탈로그 및 스키마로 이동한 다음 테이블 탭을 엽니다.

Lakebase의 데이터 피드 변경 탭에는 다음 두 개의 하위 탭이 있습니다.

• 스키마: 각 원본 스키마, Unity 카탈로그의 대상 카탈로그 및 스키마 및 상태를 나열합니다.
• 테이블: 각 원본 테이블, 해당 대상 lb_<table_name>_history 테이블, 상태(Streaming 또는 Snapshotting), 커밋된 LSN(피드가 Delta에 어디까지 기록했는지 나타내며, 초기 스냅샷이 진행 중인 동안에는 -로 표시됨), 그리고 마지막 업데이트(테이블에 마지막으로 변경 사항이 수신된 시간)를 나열합니다.

Lakebase SQL 편집기에서 이를 실행하여 Postgres에서 피드 상태를 검사할 수도 있습니다.

SELECT * FROM wal2delta.tables;

결과에는 테이블당 table_oid, status(STREAMING 또는 SNAPSHOTTING), committed_lsn, 그리고 last_write_time가 포함됩니다.

대상 테이블 스키마

CDF는 대상 카탈로그 및 스키마에 명명된 lb_<table_name>_history 원본 테이블당 하나의 델타 테이블을 씁니다. 원본 열 외에도 각 행에는 다음 시스템 열이 있습니다.

일반적인 변경 패턴

• 초기 스냅샷: CDF가 기존 Lakebase 테이블에서 처음 실행되면 각 기존 행이 으로 _pg_change_type = 'insert'작성됩니다.
• 업데이트: 업데이트는 두 개의 행을 생성합니다. 하나는 이전 행이고 _pg_change_type = 'update_preimage' 다른 하나는 _pg_change_type = 'update_postimage' (새 행)입니다.
• 삭제: 삭제 작업은 _pg_change_type = 'delete'이 포함된 한 행을 생성합니다.

이러한 이벤트는 델타 변경 데이터 피드와 동일한 변경 이벤트이므로 동일한 다운스트림 패턴이 적용됩니다.

작동 방식

• 이름 충돌: 두 원본 테이블이 동일한 대상 이름에 매핑되는 경우(예를 들어 sales.users 및 marketing.users가 모두 lb_users_history에 매핑되는 경우), CDF는 첫 번째 테이블을 lb_users_history에 기록하고 두 번째 테이블에는 자동으로 접미사를 붙여 lb_users_history_1에 기록합니다. Unity 카탈로그에서 대상 테이블의 이름을 바꿀 수 있으며 피드가 계속 작동합니다.
• 스키마 수준 범위: Lakebase 스키마에서 CDF를 시작하면 해당 스키마의 모든 현재 및 이후 테이블이 포함됩니다. 빈 테이블은 건너뜁니다. 테이블에는 대상에 표시할 행이 하나 이상 있어야 합니다.
• 삭제된 원본 테이블: Lakebase에서 테이블을 삭제하면 Unity 카탈로그의 대상 델타 테이블이 유지됩니다.

다운스트림 파이프라인 빌드

Lakebase CDF는 운영 변경에 대응하는 다운스트림 파이프라인을 위해 설계되었습니다. 아래 패턴은 가장 간단한 것에서 가장 유연한 순서로 피드를 사용하는 세 가지 방법을 보여 줍니다.

예제 시나리오입니다. 전자 상거래 앱은 Postgres orders 테이블에 주문을 기록하고, 각 행에는 item_id 및 quantity가 포함됩니다. 물류 팀에는 라이브 재고 수준이 필요합니다. CDF를 사용하면 orders에 대한 모든 변경 사항이 Unity Catalog의 lb_orders_history Delta 테이블에 저장됩니다. 다운스트림 파이프라인은 주문이 배치, 편집 또는 취소될 때마다 해당 변경 피드를 읽고 테이블을 업데이트 inventory_levels 합니다.

구체화된 뷰를 사용하여 현재 인벤토리 계산

가장 간단한 패턴은 기록 테이블에 대한 SQL 구체화 뷰 입니다. MV는 새 변경 이벤트가 도착할 때마다 점진적으로 갱신되며, 후속 소비자는 이를 다른 테이블과 마찬가지로 조회합니다.

CREATE MATERIALIZED VIEW inventory_levels AS
SELECT
  item_id,
  SUM(
    CASE
      -- New orders (and the "new half" of updates) decrement inventory
      WHEN _pg_change_type IN ('insert', 'update_postimage') THEN -quantity
      -- Cancellations (and the "old half" of updates) restore inventory
      WHEN _pg_change_type IN ('delete', 'update_preimage') THEN quantity
      ELSE 0
    END
  ) AS current_inventory,
  MAX(_timestamp) AS last_transaction_ts,
  MAX(_pg_lsn) AS last_lsn
FROM lb_orders_history
GROUP BY item_id;

모든 업데이트에 대해 생성된 두 행은 순 변경을 제외하고 서로를 취소하므로 주문이 편집될 때 실행 합계가 올바르게 유지됩니다.

Spark 선언적 파이프라인을 사용하여 변경 내용 스트리밍

구조화된 메달리온 아키텍처의 경우 브론즈, 실버, 골드 테이블을 선언하려면 Spark Declarative Pipelines (SDP)를 사용합니다. SDP는 검사점과 종속성 관리를 대신 처리해 주며, 이를 서로 연결된 파이프라인으로 실행합니다.

import dlt
from pyspark.sql import functions as F

@dlt.table
def inventory_adjustments():
    return (
        spark.readStream.table("<catalog>.<schema>.lb_orders_history")
        .withColumn(
            "delta",
            F.when(F.col("_pg_change_type").isin("insert", "update_postimage"), -F.col("quantity"))
             .when(F.col("_pg_change_type").isin("delete", "update_preimage"), F.col("quantity"))
             .otherwise(0),
        )
        .select("item_id", "delta", "_timestamp")
    )

@dlt.expect_or_drop("non_negative_stock", "on_hand >= 0")
@dlt.table
def inventory_levels():
    return (
        spark.read.table("LIVE.inventory_adjustments")
        .groupBy("item_id")
        .agg(F.sum("delta").alias("on_hand"))
    )

inventory_adjustments는 lb_orders_history를 사용해 readStream를 증분 방식으로 읽고 이벤트별 델타를 생성합니다. inventory_levels 현재 재고를 계산하기 위해 item_id 기준으로 집계합니다. 기대치는 재고를 부정적으로 밀어붙이는 행을 떨어뜨려 버그 업스트림을 알립니다.

전체 엔드투엔드 연습은 자습서: 변경 데이터 캡처를 사용하여 ETL 파이프라인 빌드를 참조하세요.

Spark 구조적 스트리밍을 사용한 사용자 지정 처리

완전한 제어가 필요한 경우(예: 사용자 지정 병합, 부작용 또는 여러 싱크)에는 Spark Structured Streaming을 사용해 히스토리 테이블을 직접 읽고 foreachBatch를 사용하여 대상에 씁니다.

from pyspark.sql import functions as F
from delta.tables import DeltaTable

def update_inventory(batch_df, batch_id):
    deltas = (
        batch_df
        .withColumn(
            "delta",
            F.when(F.col("_pg_change_type").isin("insert", "update_postimage"), -F.col("quantity"))
             .when(F.col("_pg_change_type").isin("delete", "update_preimage"), F.col("quantity"))
             .otherwise(0),
        )
        .groupBy("item_id")
        .agg(F.sum("delta").alias("delta"))
    )

    target = DeltaTable.forName(spark, "<catalog>.<schema>.inventory_levels")
    (target.alias("t")
        .merge(deltas.alias("s"), "t.item_id = s.item_id")
        .whenMatchedUpdate(set={"on_hand": F.expr("t.on_hand + s.delta")})
        .whenNotMatchedInsert(values={"item_id": "s.item_id", "on_hand": "s.delta"})
        .execute())

(spark.readStream.table("<catalog>.<schema>.lb_orders_history")
    .writeStream
    .foreachBatch(update_inventory)
    .option("checkpointLocation", "/Volumes/<catalog>/<schema>/checkpoints/inventory_levels")
    .start())

각 마이크로배치는 변경 이벤트를 item_id를 기준으로 집계하고 순 델타를 inventory_levels에 병합합니다.

의도적으로 점진적입니다. 각 lb_<table_name>_history 테이블은 추가 전용 델타 테이블입니다. 원본의 각 변경 사항은 새 행으로 기록되며, _pg_change_type가 해당 작업을 표시합니다. Databricks SQL 구체화된 뷰, Lakeflow Spark 선언적 파이프라인 흐름 및 Spark 구조적 스트리밍 작업은 모두 델타 트랜잭션 로그에서 새 행을 증분 방식으로 처리하므로 다운스트림 파이프라인은 변경된 내용에 비례하여 작동합니다. 변경 의미 체계가 행 데이터에 이미 인코딩되어 있으므로 기록 테이블에서 델타 변경 데이터 피드 를 사용하도록 설정할 필요가 없습니다.

데이터 형식 매핑

CDF는 대부분의 표준 PostgreSQL 기본 형식을 지원합니다. 직접 델타에 해당하는 형식이 없는 형식은 STRING으로 저장됩니다.

STRING으로 저장된 형식:

• 지리/기하 도형(PostGIS): PostGIS 확장의 형식(예: geometry, geography).
• 벡터(pgvector):vector pgvector 확장의 형식입니다.
• 복합/구조체 형식: 로 정의된 CREATE TYPE ... AS (field_name type, ...)사용자 지정 형식 이들은 이름이 있는 필드를 가진 행과 유사한 형식입니다.
• 맵: 확장의 hstore와 같은 맵과 유사한 키-값 유형입니다. Postgres에는 기본 제공 지도 형식이 없습니다. hstore 는 열에 키-값 쌍을 저장하는 일반적인 방법입니다.

스키마 변경 관리

• Postgres에서 테이블 이름을 바꾸면(예:ALTER TABLE users RENAME TO customers) 피드를 계속할 수 있습니다. 대상 델타 테이블 이름은 변경되지 않습니다.lb_users_history
• 스키마 변경 (열 추가, 열 삭제 또는 열의 데이터 형식 변경)은 영향을 받는 테이블의 다시 스냅샷을 트리거합니다. CDF는 Postgres에서 전체 테이블을 다시 읽고 대상 델타 테이블로 다시 작성합니다.

Lakebase CDF 사용 안 함

CDF를 사용하지 않도록 설정하면 프로젝트의 모든 Lakebase 스키마에 대한 피드가 중지됩니다.

1. Azure Databricks 작업 영역의 앱 전환기(오른쪽 위)에서 Lakebase Postgres 엽니다.
2. Lakebase 프로젝트 및 CDF를 구성한 분기를 선택합니다.
3. 분기 개요를 열고 데이터 피드 변경 탭을 클릭합니다.
4. 비활성화를 클릭합니다. 확인 대화 상자에서 변경 내용이 델타 테이블로 흐르는 것을 중지한다는 경고를 검토한 다음 다시 사용 안 함을 클릭하여 확인합니다.

CDF를 사용하지 않도록 설정해도 컴퓨팅이 다시 시작되지 않습니다.

제한 사항 및 문제 해결

데이터 피드 변경 탭에서 또는 Lakebase에서 이를 실행하여 테이블별 상태(스냅샷, 건너뛰기 또는 스트리밍)를 볼 수 있습니다.

SELECT * FROM wal2delta.tables;

피드에 테이블이 표시되지 않는 일반적인 이유:

• REPLICA IDENTITY FULL 설정되지 않음: 테이블에 대해 ALTER TABLE <table_name> REPLICA IDENTITY FULL;를 실행합니다. 1단계: 복제본 ID 전체 설정 참조
• 분할된 테이블: 레이크베이스 분할 테이블은 지원되지 않습니다. 분할된 테이블을 포함하는 스키마로 인해 해당 테이블이 실패합니다.
• 빈 테이블: 행이 0인 테이블은 하나 이상의 행이 있을 때까지 건너뜁니다.

• 대상 스토리지의 프라이빗 엔드포인트: 대상 Unity 카탈로그 카탈로그에 대한 관리형 스토리지가 프라이빗 엔드포인트를 통해서만 액세스할 수 있는 경우(예: 스토리지 계정에 대한 공용 네트워크 액세스가 비활성화된 경우) Lakebase CDF는 지원되지 않습니다. 해결 방법으로 관리되는 스토리지에 공개적으로 연결할 수 있는 카탈로그를 구성하고 해당 카탈로그를 CDF 대상으로 사용합니다.

다음 단계

• Spark 선언적 파이프라인을 사용하여 증분 ETL을 빌드합니다. 자습서: 전체 연습을 위해 변경 데이터 캡처를 사용하여 ETL 파이프라인 빌드를 참조하세요.
• Databricks SQL을 사용하여 브론즈 계층을 쿼리합니다. Databricks SQL을 사용하여 데이터 웨어하우징 시작 참조
• 대상 델타 테이블에 대한 시간 이동 쿼리가 있는 감사 기록입니다.