Delta Lake를 사용하여 선택적으로 데이터 덮어쓰기

Azure Databricks는 Delta Lake 기능을 활용하여 선택적 덮어쓰기를 위한 두 가지 고유한 옵션을 지원합니다.

• replaceWhere 옵션은 지정된 조건자와 일치하는 모든 레코드를 원자성으로 바꿉니다.
• 동적 파티션 덮어쓰기를 사용하여 테이블을 분할하는 방법에 따라 데이터 디렉터리를 바꿀 수 있습니다.

대부분의 작업에서 Databricks는 덮어쓸 데이터를 지정하는 데 사용하는 replaceWhere 것이 좋습니다.

replaceWhere를 사용하여 임의 선택적 덮어쓰기

임의의 식과 일치하는 데이터만 선택적으로 덮어쓸 수 있습니다.

다음 명령은 start_date로 분할된 대상 테이블의 1월 이벤트를 replace_data의 원자 단위 데이터로 바꿉니다.

Python

(replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .table("events")
)

Scala

replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .table("events")

SQL

INSERT INTO TABLE events REPLACE WHERE start_date >= '2017-01-01' AND end_date <= '2017-01-31' SELECT * FROM replace_data

이 샘플 코드는 replace_data데이터를 작성하고, 모든 행이 조건자와 일치하는지 확인하고, 의미 체계를 사용하여 overwrite 원자성 대체를 수행합니다. 작업의 값이 제약 조건을 벗어나면 기본적으로 오류가 발생하여 이 작업이 실패합니다.

이 동작을 overwrite 조건자 범위 내의 값과 insert 지정된 범위를 벗어난 레코드로 변경할 수 있습니다. 이렇게 하려면 다음 설정 중 하나를 사용하여 false로 설정 spark.databricks.delta.replaceWhere.constraintCheck.enabled 하여 제약 조건 검사를 사용하지 않도록 설정합니다.

Python

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", False)

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", false)

SQL

SET spark.databricks.delta.replaceWhere.constraintCheck.enabled=false

레거시 동작

레거시 기본 동작은 replaceWhere 파티션 열에만 조건자 일치 데이터를 덮어씁니다. 이 레거시 모델을 사용하면 다음 명령이 대상 테이블의 1월 월을 원자성으로 대체합니다. 이 월은 다음의 데이터date로 분할됩니다df.

Python

(df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .table("people10m")
)

Scala

df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .table("people10m")

이전 동작으로 대체하려는 경우 플래그를 spark.databricks.delta.replaceWhere.dataColumns.enabled 사용하지 않도록 설정할 수 있습니다.

Python

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", False)

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", false)

SQL

SET spark.databricks.delta.replaceWhere.dataColumns.enabled=false

동적 파티션 덮어쓰기

동적 파티션 덮어쓰기가 쓰기가 새 데이터를 커밋하는 파티션만 업데이트합니다. 이러한 파티션의 모든 기존 데이터를 덮어쓰고 다른 데이터는 변경되지 않은 상태로 유지합니다.

Azure Databricks는 다음 두 가지 방법을 지원합니다.

• REPLACE USING (권장) - Databricks SQL 웨어하우스, 서버리스 컴퓨팅 및 클래식 컴퓨팅을 비롯한 모든 컴퓨팅 유형에서 작동합니다. Spark 세션 구성을 설정할 필요가 없습니다.
• partitionOverwriteMode (레거시) - 클래식 컴퓨팅 및 Spark 세션 구성 설정이 필요합니다. Databricks SQL 또는 서버리스 컴퓨팅에서는 지원되지 않습니다.

아래 섹션에서는 각 방법을 사용하는 방법을 보여 줍니다.

동적 파티션 덮어쓰기 REPLACE USING

Databricks Runtime 16.3 이상에서는 을 사용하여 REPLACE USING분할된 테이블에 대한 동적 파티션 덮어쓰기를 지원합니다. 이 메서드를 사용하면 Spark 세션 구성을 설정할 필요 없이 모든 컴퓨팅 형식에서 데이터를 선택적으로 덮어쓸 수 있습니다. REPLACE USING 는 Databricks SQL 웨어하우스, 서버리스 컴퓨팅 및 클래식 컴퓨팅에서 작동하는 컴퓨팅 독립적 원자성 덮어쓰기 동작을 사용하도록 설정합니다.

REPLACE USING 들어오는 데이터가 대상으로 하는 파티션만 덮어씁니다. 다른 모든 파티션은 변경되지 않은 상태로 유지됩니다.

다음 예제에서는 동적 파티션 덮어쓰기를 사용 하는 방법을 보여 줍니다 REPLACE USING. 현재 Python 또는 Scala가 아닌 SQL만 사용할 수 있습니다. 자세한 내용은 SQL 언어 참조를 참조 INSERT 하세요.

INSERT INTO TABLE events
  REPLACE USING (event_id, start_date)
  SELECT * FROM source_data

동적 파티션 덮어쓰기를 위해 다음 제약 조건 및 동작을 염두에 두세요.

• 테이블의 파티션 열 전체 집합을 USING 절에 지정해야 합니다.
• 기록된 데이터가 예상 파티션에만 영향을 주는지 항상 확인합니다. 잘못된 파티션의 단일 행은 의도치 않게 전체 파티션을 덮어쓸 수 있습니다.

값을 같음으로 처리하는 것과 같이 지원하는 것보다 REPLACE USING 사용자 지정 가능한 일치 논리가 NULL 필요한 경우 대신 보완 REPLACE ON 을 사용합니다. 자세한 내용은 INSERT를 참조하세요.

동적 파티션 덮어쓰기 partitionOverwriteMode (레거시 시스템)

Databricks Runtime 11.3 LTS 이상에서는 SQL의 INSERT OVERWRITE 또는 DataFrame의 df.write.mode("overwrite")을 사용하여 덮어쓰기 모드로 분할된 테이블에 대한 동적 파티션 덮어쓰기를 지원합니다. 이 유형의 덮어쓰기는 Databricks SQL 웨어하우스 또는 서버리스 컴퓨팅이 아닌 클래식 컴퓨팅에만 사용할 수 있습니다.

Spark 세션 구성 spark.sql.sources.partitionOverwriteMode를 dynamic으로 설정하여 동적 파티션 덮어쓰기 모드를 구성합니다. 또는 DataFrameWriter 옵션 partitionOverwriteMode을 dynamic로 설정할 수 있습니다. 있는 경우 쿼리별 옵션은 세션 구성에 정의된 모드를 재정의합니다. spark.sql.sources.partitionOverwriteMode에 대한 기본값은 static입니다.

다음 예제에서는 partitionOverwriteMode 사용을 보여 줍니다.

SQL

SET spark.sql.sources.partitionOverwriteMode=dynamic;
INSERT OVERWRITE TABLE default.people10m SELECT * FROM morePeople;

Python

(df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")
)

Scala

df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")

다음 제약 조건 및 동작을 partitionOverwriteMode에 유의하세요.

• overwriteSchema를 true로 설정할 수 없습니다.
• 동일한 partitionOverwriteMode 작업에서 replaceWhere와 DataFrameWriter를 둘 다 지정할 수 없습니다.
• 옵션을 사용하여 replaceWhere 조건을 지정 DataFrameWriter 하는 경우 Delta Lake는 해당 조건을 적용하여 덮어쓸 데이터를 제어합니다. 이 옵션은 세션 수준 구성보다 partitionOverwriteMode 우선합니다.
• 기록된 데이터가 예상 파티션에만 영향을 주는지 항상 확인합니다. 잘못된 파티션의 단일 행은 의도치 않게 전체 파티션을 덮어쓸 수 있습니다.