Azure Databricks의 스키마 진화

스키마 진화 는 시간이 지남에 따라 데이터 구조의 변화에 적응하는 시스템의 기능을 나타냅니다. 이러한 변경 내용은 반구조적 데이터, 이벤트 스트림 또는 새 필드가 추가되거나 데이터 형식이 이동하거나 중첩된 구조가 진화하는 타사 원본으로 작업할 때 일반적입니다.

일반적인 변경 내용은 다음과 같습니다.

새 열: 이전에 정의되지 않은 추가 필드, 때로는 사용자 지정 백필 값이 있는 필드입니다.
열 이름 바꾸기: 열 이름 변경(예: 에서 namefull_name.
삭제된 열: 테이블 스키마에서 열을 제거합니다.
형식 확장: 열의 형식을 더 넓은 형식으로 변경합니다. 예를 들어, INT 필드가 DOUBLE로 변하는 경우입니다.
다른 형식 변경: 열의 형식 변경 예를 들어, INT 필드가 STRING로 변하는 경우입니다.

스키마 진화를 지원하는 것은 빈번한 수동 업데이트 없이 변경 데이터를 수용할 수 있는 복원력 있는 장기 실행 파이프라인을 빌드하는 데 중요합니다.

Components

Azure Databricks 스키마 진화에는 4가지 주요 구성 요소 범주가 포함되며, 각 처리 스키마는 독립적으로 변경됩니다.

커넥터: 외부 원본에서 데이터를 수집하는 구성 요소입니다. 여기에는 자동 로더, Kafka, Kinesis 및 Lakeflow 커넥터가 포함됩니다.
형식 파서: 원시 형식(예from_json: , from_avrofrom_xml및 from_protobuf.)을 디코딩하는 함수
엔진: 쿼리를 실행하는 처리 엔진으로, 구조적 스트리밍이 포함됩니다.
데이터 세트: 데이터를 유지 및 제공하는 스트리밍 테이블, 구체화된 뷰, 델타 테이블 및 뷰입니다.

스키마 진화

데이터 엔지니어링 아키텍처의 스키마 진화의 각 구성 요소는 독립적입니다. 데이터 처리 흐름에서 원하는 동작을 달성하기 위해 개별 구성 요소에서 스키마 진화를 구성해야 합니다.

예를 들어 자동 로더를 사용하여 델타 테이블에 데이터를 수집하는 경우 두 개의 지속형 스키마가 있습니다. 하나는 해당 스키마 위치에서 자동 로더에 의해 관리되고 다른 하나는 대상 델타 테이블의 스키마입니다. 안정적인 상태에서 이 두 가지는 동일합니다. 자동 로더가 들어오는 데이터에 따라 스키마를 발전하는 경우 델타 테이블도 해당 스키마를 발전시켜야 합니다. 그렇지 않으면 쿼리가 실패합니다. 이 경우 (a) 스키마 진화를 사용하도록 설정하거나 직접 DDL 명령을 사용하여 대상 델타 테이블 스키마를 업데이트하거나 (b) 대상 델타 테이블을 완전히 다시 작성할 수 있습니다.

커넥터별 스키마 진화 지원

다음 섹션에서는 각 Azure Databricks 구성 요소가 다양한 유형의 스키마 변경을 처리하는 방법을 자세히 설명합니다.

자동 로더

자동 로더는 열 변경 및 형식 확대를 지원합니다. cloudFiles.schemaEvolutionMode 및 rescuedDataColumn을(를) 사용하여 자동 스키마 진화를 구성. 수동으로 schemaHints을(를) 설정하거나 변경할 수 없는 schema을(를) 사용할 수 있습니다. 스키마를 자동으로 발전시키는 경우 스트림은 처음에 실패합니다. 다시 시작할 때는 진화된 스키마가 사용됩니다. 자동 로더 스키마 진화는 어떻게 작동하나요?를 참조하세요.

새 열: 선택된 schemaEvolutionMode에 따라 지원됩니다. 스키마에 새 열을 추가하는 데 필요한 수동 다시 시작으로 실패합니다.
열 이름 바꾸기: 선택한 항목에 schemaEvolutionMode 따라 지원됩니다. 이름이 바뀐 열은 새로 추가된 열로 간주되며, 새로운 행에는 이전 열이 NULL로 채워집니다. 스키마를 업데이트하는 데 필요한 수동 다시 시작으로 실패합니다.
삭제된 열: 지원됩니다. 삭제된 열에 대해 새 행이 NULL로 설정되는 소프트 삭제로 처리됩니다.
형식 확장: Databricks Runtime 16.4 이상에서 schemaEvolutionMode로 설정하여 지원됩니다. 지원되는 데이터 형식 변경 내용은 자동으로 확장됩니다. 지원되지 않는 형식 변경 내용은 rescuedDataColumn에 캡처됩니다. 자동 로더를 사용하여 자동 형식 확대를 참조하세요.
기타 유형 변경: 지원되지 않습니다. 설정된 rescuedDataColumn 이고 rescueDataColumn 가 schemaEvolutionMode로 설정된 경우 형식 변경 내용이 rescue 에 캡처됩니다. 그렇지 않으면 수동 스키마 변경이 필요합니다.

델타 커넥터

델타 커넥터는 스키마 진화를 지원할 수 있습니다. 열 매핑 및 schemaTrackingLocation이 활성화된 델타 테이블에서 읽는 경우 열 이름 바꾸기 및 삭제된 열에 대한 스키마 진화를 지원합니다. 스트림을 중지하지 않고 스키마를 발전하려면 각 변경 내용에 대해 올바른 Spark 구성을 설정해야 합니다. 그렇지 않으면, 스트림은 변경이 감지될 때마다 업데이트된 스키마로 변경한 후 중지됩니다. 그런 다음 스트리밍 쿼리를 수동으로 다시 시작하여 처리를 다시 시작해야 합니다.

새 열: 지원됩니다. mergeSchema 사용하도록 설정하면 새 열이 자동으로 추가됩니다. 그렇지 않으면 쿼리가 실패하고 스트림을 다시 시작하여 스키마에 새 열을 추가해야 하지만 델타 테이블에 다시 쓸 필요는 없습니다.
열 이름 바꾸기: 지원됨 Spark 구성 spark.databricks.delta.streaming.allowSourceColumnRename을 사용하여 스트리밍 쿼리 내에서 스키마를 발전할 수 있습니다.
삭제된 열: 지원됩니다. Spark 구성 spark.databricks.delta.streaming.allowSourceColumnDrop을 사용하여 스트리밍 쿼리 내에서 스키마를 발전할 수 있습니다.
형식 확장: Databricks Runtime 16.4 LTS 이상에서 지원됩니다. mergeSchema 대상 테이블에서 활성화 및 형식 확대를 사용하도록 설정하면 형식 변경 내용이 자동으로 처리됩니다. table 속성을 사용하여 타입 확장을 delta.enableTypeWidening 활성화할 수 있습니다. 타입 확장을 참조하세요.
기타 유형 변경: 지원되지 않습니다.

SaaS 및 CDC 커넥터

SaaS 및 CDC 커넥터는 열이 변경되면 스키마를 자동으로 진화합니다. 변경 내용이 감지되면 자동 다시 시작을 통해 처리됩니다. 형식을 변경하려면 전체 새로 고침이 필요합니다.

새 열: 지원됩니다. 스키마 불일치를 해결하기 위해 쿼리가 자동으로 다시 시작됩니다.
열 이름 바꾸기: 지원됨 스키마 불일치를 해결하기 위해 쿼리가 자동으로 다시 시작됩니다. 이름이 바뀐 열은 추가된 새 열로 처리됩니다.
삭제된 열: 지원됩니다. 삭제된 열은 일시적으로 삭제된 것으로 처리되며, 이 경우 삭제된 열의 새 행은 NULL로 설정됩니다.
형식 확장: 지원되지 않습니다. 스키마를 업데이트하려면 전체 새로 고침이 필요합니다.
기타 유형 변경: 지원되지 않습니다. 스키마를 업데이트하려면 전체 새로 고침이 필요합니다.

Kinesis, Kafka, Pub/Sub 및 Pulsar 커넥터

네이티브 스키마 진화가 지원되지 않습니다. 각 커넥터 함수는 이진 Blob을 반환합니다. 스키마 진화는 형식 파서에서 처리됩니다.

새 열: 형식 파서에서 처리됩니다.
열 이름 바꾸기: 형식 파서에서 처리됩니다.
삭제된 열: 형식 파서에서 처리됩니다.
형식 확장: 형식 파서에 의해 처리됩니다.
다른 형식 변경: 형식 파서에서 처리됩니다.

형식 파서별 스키마 진화 지원

`from_json` 파서

from_json 파서는 스키마 진화를 지원하지 않습니다. 스키마를 수동으로 업데이트해야 합니다. Lakeflow 파이프라인 내에서 from_json를 사용할 때 schemaLocationKey 및 schemaEvolutionMode를 사용하여 자동 스키마 진화를 활성화할 수 있습니다.

새 열: 자동 스키마 진화를 사용하도록 설정하면 자동 로더처럼 동작합니다.
열 이름 바꾸기: 자동 스키마 진화를 사용하도록 설정하면 자동 로더처럼 동작합니다.
삭제된 열: 자동 스키마 진화를 사용하도록 설정하면 자동 로더처럼 동작합니다.
형식 확장: 자동 스키마 진화를 사용하도록 설정하면 자동 로더처럼 동작합니다.
다른 형식 변경: 자동 스키마 진화를 사용하도록 설정하면 자동 로더처럼 동작합니다.

`from_avro` 및 `from_protobuf` 파서

from_avro 파서와 from_protobuf 파서는 같은 방식으로 동작합니다. Confluent 스키마 레지스트리에서 스키마를 가져오거나 사용자가 스키마를 제공할 수 있으며 스키마를 수동으로 업데이트해야 합니다. 또는 from_avro 함수 내에서 from_protobuf 스키마 진화라는 개념은 없으며 실행 엔진 및 스키마 레지스트리에서 처리해야 합니다.

새 열: Confluent 스키마 레지스트리에서 지원됩니다. 그렇지 않으면 사용자가 스키마를 수동으로 업데이트해야 합니다.
열 이름 바꾸기: Confluent 스키마 레지스트리에서 지원됩니다. 그렇지 않으면 사용자가 스키마를 수동으로 업데이트해야 합니다.
삭제된 열: Confluent 스키마 레지스트리에서 지원됩니다. 그렇지 않으면 사용자가 스키마를 수동으로 업데이트해야 합니다.
형식 확장: Confluent 스키마 레지스트리에서 지원됩니다. 그렇지 않으면 사용자가 스키마를 수동으로 업데이트해야 합니다.
기타 형식 변경: Confluent 스키마 레지스트리에서 지원됩니다. 그렇지 않으면 사용자가 스키마를 수동으로 업데이트해야 합니다.

`from_csv` 및 `from_xml` 파서

from_csv 및 from_xml 파서는 스키마 진화를 지원하지 않습니다.

새 열: 지원되지 않음
열 이름 바꾸기: 지원되지 않음
삭제된 열: 지원되지 않음
형식 확장: 지원되지 않음
기타 형식 변경 내용: 지원되지 않음

엔진별 스키마 진화 지원

구조적 스트리밍

스트리밍 쿼리의 스키마는 계획 단계 중에 잠기며 모든 마이크로 일괄 처리는 다시 계획하지 않고 해당 계획을 다시 사용합니다. 원본 스키마가 실행 중간에 변경되면 쿼리가 실패하고 Spark가 새 스키마에 대해 다시 계획할 수 있도록 사용자가 스트리밍 쿼리를 다시 시작해야 합니다.

스트림이 기록할 데이터 세트도 스키마 진화를 지원해야 합니다.

새 열: 지원됩니다. 쿼리가 실패하고 스키마 불일치를 해결하려면 스트림을 다시 시작해야 합니다.
열 이름 바꾸기: 지원됨 쿼리가 실패하고 스키마 불일치를 해결하려면 스트림을 다시 시작해야 합니다.
삭제된 열: 지원됩니다. 쿼리가 실패하고 스키마 불일치를 해결하려면 스트림을 다시 시작해야 합니다.
형식 확대: 지원됨. 쿼리가 실패하고 스키마 불일치를 해결하려면 스트림을 다시 시작해야 합니다.
기타 형식 변경: 지원됨. 쿼리가 실패하고 스키마 불일치를 해결하려면 스트림을 다시 시작해야 합니다.

데이터 세트별 스키마 진화

스트리밍 테이블

스트리밍 테이블은 기본적으로 병합 스키마 진화 동작을 지원합니다. 스키마를 업데이트해도 수동으로 다시 시작할 필요는 없지만 임의 스키마를 변경하려면 전체 새로 고침이 필요합니다.

새 열: 지원됩니다. 스키마 불일치를 해결하기 위해 쿼리가 자동으로 다시 시작됩니다.
열 이름 바꾸기: 지원됨 스키마 불일치를 해결하기 위해 쿼리가 다시 시작됩니다. 이름이 바뀐 열은 추가된 새 열로 처리됩니다.
삭제된 열: 지원됩니다. 삭제된 열은 일시 삭제로 처리됩니다. 여기서 삭제된 열의 새 행은 NULL로 설정됩니다.
형식 확대: 지원됨. 파이프라인 수준이나 테이블에서 직접적으로 형식 범위 확장을 사용하도록 설정해야 합니다. Lakeflow 파이프라인의 형식 확대를 참조하세요.
기타 유형 변경: 지원되지 않습니다. 스키마를 업데이트하려면 전체 새로 고침이 필요합니다.

머터리얼라이즈드 뷰

스키마 또는 정의 쿼리에 대한 업데이트는 구체화된 뷰의 전체 다시 계산을 트리거합니다.

새 열: 전체 다시 계산이 트리거되었습니다.
열 이름 바꾸기: 전체 다시 계산이 트리거됩니다.
삭제된 열: 전체 다시 계산이 트리거됩니다.
형식 확장: 전체 다시 계산이 트리거됩니다.
기타 형식 변경: 전체 다시 계산이 트리거됩니다.

델타 테이블

델타 테이블은 테이블 데이터를 다시 작성하지 않고 열 형식을 변경, 삭제 및 확대하는 등 테이블 스키마를 업데이트하는 다양한 구성을 지원합니다. 지원되는 구성에는 병합 스키마 진화, 열 매핑, 형식 확장 및 overwriteSchema가 포함됩니다.

새 열: 지원됩니다. 델타 테이블을 다시 작성할 필요 없이 병합 스키마 진화를 사용하도록 설정하면 자동으로 진화합니다. 병합 스키마 진화를 사용하도록 설정하지 않으면 업데이트가 실패합니다.
열 이름 바꾸기: 지원됨 열 매핑을 사용하도록 설정된 수동 ALTER TABLE DDL 명령을 통해 이름을 바꿀 수 있습니다. 델타 테이블을 다시 작성할 필요가 없습니다.
삭제된 열: 지원됩니다. 열 매핑을 사용하도록 설정된 수동 ALTER TABLE DDL 명령을 통해 열을 삭제할 수 있습니다. 델타 테이블을 다시 작성할 필요가 없습니다.
형식 확대: 지원됨. 형식 확장 및 병합 스키마 진화를 사용하도록 설정하면 형식 변경이 자동으로 적용됩니다. 형식 확대를 사용하는 경우 수동 ALTER TABLE DDL 명령을 통해 열을 확장할 수 있습니다. 둘 중 하나를 구성하지 않으면 작업이 실패합니다. 자동 스키마 진화로 확장된 형식 참조.
다른 형식 변경: 지원되지만 델타 테이블을 완전히 다시 작성해야 합니다. overwriteSchema을(를) 활성화하여 전체 델타 테이블을 재작성해야 합니다. 그렇지 않으면 작업이 실패합니다.

Views

뷰에 새 스키마와 column_list 일치하지 않거나 구문 분석할 수 없는 쿼리가 있는 경우 보기가 잘못됩니다. 그렇지 않으면 형식 변경을 위해 SCHEMA TYPE EVOLUTION 에서 스키마 진화를 활성화할 수 있으며, 형식 진화의 상위 개념인 SCHEMA EVOLUTION 을 사용하여 새 열 추가, 열 이름 변경, 열 삭제에도 대응할 수 있습니다.

새 열: 지원됩니다. SCHEMA EVOLUTION 모드에서는 명시적 작업이 없는 경우 수동 개입 없이 보기가 자동으로 진화합니다column_list. 그렇지 않으면 보기가 유효하지 않을 수 있으며 사용자가 쿼리할 수 없습니다.
열 이름 바꾸기: 지원됩니다. SCHEMA EVOLUTION 모드에서는 명시적 작업이 없는 경우 수동 개입 없이 보기가 자동으로 진화합니다column_list. 그렇지 않으면 보기가 유효하지 않게 될 수 있습니다.
삭제된 열: 지원됩니다. SCHEMA EVOLUTION 모드에서는 명시적 작업이 없는 경우 수동 개입 없이 보기가 자동으로 진화합니다column_list. 그렇지 않으면 보기가 유효하지 않게 될 수 있습니다.
형식 확대: 지원됨. SCHEMA TYPE EVOLUTION 모드를 사용하면 형식이 변경될 때 보기가 자동으로 진화합니다. SCHEMA EVOLUTION 모드에서는 명시적 작업이 없는 경우 수동 개입 없이 보기가 자동으로 진화합니다column_list. 그렇지 않으면 보기가 유효하지 않게 될 수 있습니다.
기타 형식 변경: 지원됨. SCHEMA TYPE EVOLUTION 모드를 사용하면 형식이 변경될 때 보기가 자동으로 진화합니다. SCHEMA EVOLUTION 모드에서는 명시적 작업이 없는 경우 수동 개입 없이 보기가 자동으로 진화합니다column_list. 그렇지 않으면 보기가 유효하지 않게 될 수 있습니다.

예시

다음 예제에서는 Confluent 스키마 레지스트리에 등록된 Avro로 인코딩된 페이로드를 사용하여 Kafka 토픽을 수집하고 스키마 진화를 사용하도록 설정된 관리되는 델타 테이블에 쓰는 방법을 보여 줍니다.

주요 사항이 설명되었습니다.

Kafka 커넥터와 통합합니다.
Kafka 스키마 레지스트리에서 from_avro 사용하여 Avro 레코드를 디코딩합니다.
스키마 진화를 avroSchemaEvolutionMode를 설정하여 처리합니다.
추가 변경이 허용되도록 설정된 델타 테이블에 mergeSchema 씁니다.

코드는 Confluent 스키마 레지스트리를 사용하여 Avro로 인코딩된 데이터를 출력하는 Kafka 토픽이 있다고 가정합니다.

# ----- CONFIG: fill these in -----
# Catalog and schema:
CATALOG = "<catalog_name>"
SCHEMA = "<schema_name>"
# Schema Registry:
# (This is where the producer evolves the schema)
SCHEMA_REG = "<schema registry endpoint>"
SR_USER    = "<api key>"
SR_PASS    = "<api secret>"
# Confluent Cloud: SASL_SSL broker:
BOOTSTRAP = "<server:ip>"
# Kafka topic:
TOPIC = "<topic>"
# ----- end: config -----

BRONZE_TABLE = f"{CATALOG}.{SCHEMA}.bronze_users"
CHECKPOINT = f"/Volumes/{CATALOG}/{SCHEMA}/checkpoints/bronze_users"

# Kafka auth (example for Confluent Cloud SASL/PLAIN over SSL)
KAFKA_OPTS = {
  "kafka.security.protocol": "SASL_SSL",
  "kafka.sasl.mechanism": "PLAIN",
  "kafka.sasl.jaas.config": f"kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username='{SR_USER}' password='{SR_PASS}';"
}

# ----- Evolution knobs -----
# spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", value = True)

from pyspark.sql.functions import col
from pyspark.sql.avro.functions import from_avro

# Build reader
reader = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", BOOTSTRAP)
  .option("subscribe", TOPIC)
  .option("startingOffsets", "earliest")
)

# Attach Kafka auth options
for k, v in KAFKA_OPTS.items():
    reader = reader.option(k, v)

# --- No native schema evolution supported. Returns a binary blob. ---
raw_df = reader.load()

# Decode Avro with Schema Registry
# --- The format parser handles updating the schema using the schema registry ---
decoded = from_avro(
    data=col("value"),
    jsonFormatSchema=None, # using SR
    subject=f"{TOPIC}-value",
    schemaRegistryAddress=SCHEMA_REG,
    options={
      "confluent.schema.registry.basic.auth.credentials.source": "USER_INFO",
      "confluent.schema.registry.basic.auth.user.info": f"{SR_USER}:{SR_PASS}",
      # Behavior on schema changes:
      "avroSchemaEvolutionMode": "restart", # fail-fast so you can restart and adopt new fields
      "mode": "FAILFAST"
    }
).alias("payload")

bronze_df = raw_df.select(decoded, "timestamp").select("payload.*", "timestamp")

# Write to a managed Delta table as a STREAM
# --- Need to enable schema evolution separately for streaming to a Delta separately with mergeSchema --
(bronze_df.writeStream
  .format("delta")
  .option("checkpointLocation", CHECKPOINT)
  .option("ignoreChanges", "true")
  .outputMode("append")
  .option("mergeSchema", "true") # only supports adding new columns. Renaming, dropping, and type changes need to be handled separately.
  .trigger(availableNow=True) # Use availableNow trigger for Databricks SQL/Unity Catalog
  .toTable(BRONZE_TABLE)
)

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-07-10