대규모 데이터 레이크를 위한 Delta 시간 이동 소개

Delta Lake 사용을 시작하는 데 필요한 단계별 지침에 대한
O'Reilly의 새로운 ebook 미리 보기를 확인하세요.

Experiment 데이터 재현, 롤백 및 감사를 위한 데이터 버전 관리

모든 사용자를 위해 Apache Spark를 기반으로 구축된 차세대 엔진인
Databricks Delta Lake 의 시간 이동 기능을 소개하게 되어 기쁩니다.
unified analytics 이 새로운 기능을 통해 Delta는 데이터 레이크에 저장한
빅데이터의 버전을 자동으로 지정하며, 사용자는 해당 데이터의 모든 기록
버전에 액세스할 수 있습니다. 이 임시 데이터 관리는 감사하고, 실수로
잘못된 쓰기 또는 삭제가 발생한 경우 데이터를 롤백하고,
Experiment 및 보고서를 재현할 수 있도록 하여 데이터 파이프라인을 간소화합니다.
조직은 마침내 분석을 위해 자체 클라우드 스토리지에서 깨끗하고 중앙 집중화되고
버전이 지정된 빅데이터 리포지토리로 표준화할 수 있습니다.

데이터 변경과 관련된 일반적인 문제

• 데이터 변경 감사: 데이터 변경 내용을 감사하는 것은 데이터 규정 준수 측면뿐만 아니라
  시간이 지남에 따라 데이터가 어떻게 변경되었는지 이해하기 위한 간단한 디버깅 측면에서
  매우 중요합니다. 기존 데이터 시스템에서 빅데이터 기술 및 클라우드로 전환하는 조직은
  이러한 시나리오에서 어려움을 겪고 있습니다.
• Experiment 및 보고서 재현: 모델 학습 중에 데이터 사이언티스트는 Experiment
  parameter 주어진 데이터 세트에 대해 서로 다른 사용하여 다양한 실행합니다.
  과학자들이 모델을 재현하기 위해 일정 기간 후에 Experiment 다시 방문할 때 일반적으로
  소스 데이터는 업스트림 파이프라인에 의해 수정됩니다. 많은 경우 이러한 업스트림 데이터
  변경을 인식하지 못하여 Experiment재현하는 데 어려움을 겪습니다.
  일부 과학자와 조직에서는 여러 개의 데이터 복사본을 만들어 모범 사례를 설계하므로
  스토리지 비용이 증가합니다. 보고서를 생성하는 애널리스트의 경우도 마찬가지입니다.
  
• 롤백: 데이터 파이프라인은 때때로 다운스트림 소비자에 대해 잘못된 데이터를 쓸 수 있습니다. 이는 인프라 불안정에서 지저분한 데이터, 파이프라인의 버그에 이르기까지 다양한 문제로 인해 발생할 수 있습니다. 디렉터리 또는 테이블에 간단한 추가를 수행하는 파이프라인의 경우 날짜 기반 분할을 통해 롤백을 쉽게 처리할 수 있습니다. 업데이트 및 삭제를 사용하면 매우 복잡해질 수 있으며 데이터 엔지니어는 일반적으로 이러한 시나리오를 처리하기 위해 복잡한 파이프라인을 엔지니어링해야 합니다.

시간 이동 소개

Delta의 시간 이동 기능은 위의 사용 사례에 대한 데이터 파이프라인 구축을 간소화합니다.
Delta 테이블 또는 디렉터리에 쓰면 모든 운영의 버전이 자동으로 지정됩니다.
다음과 같은 두 가지 방법으로 서로 다른 버전의 데이터에 액세스할 수 있습니다.

1. 타임스탬프 사용

Scala 구문:
타임스탬프 또는 날짜 문자열을 DataFrame 판독기에 옵션으로 제공할 수 있습니다.

val df = spark.read
  .format ( "델타")
  .option("타임스탬프AsOf", "2019-01-01")
  .load ( "/ 경로 / 대상 / 내 / 테이블")

Python:

df = 스파크.read \
  .format ( "델타") \
  .option("타임스탬프AsOf", "2019-01-01") \
  .load ( "/ 경로 / 대상 / 내 / 테이블")

SQL 구문:

"2019-01-01" 기준 my_table 타임스탬프에서 SELECT COUNT(*)
date_sub(current_date(), 1)my_table 타임스탬프에서 SELECT COUNT(*)
"2019-01-01 01:30:00.000" 기준 my_table 타임스탬프에서 SELECT COUNT(*)

판독기 코드가 액세스할 수 없는 라이브러리에 있고 데이터를 읽기 위해 입력 parameter
라이브러리에 전달하는 경우에도 yyyyMMddHHmmssSSS 형식의 타임스탬프를 경로에
전달하여 테이블의 시간을 거슬러 올라갈 수 있습니다.

val 입력 경로 = "/path/to/my/table@20190101000000000"
val df = loadData(입력 경로)

액세스할 수 없는 라이브러리의 함수Function in a library that you don't access to
def loadData(입력 경로 : 문자열) : DataFrame = {
  스파크.읽기
    .format ( "델타")
    .load(입력 경로)
}

입력 경로 = "/path/to/my/table@20190101000000000"
df = loadData(입력 경로)

# 액세스할 수 없는 라이브러리의 함수
def loadData(입력 경로):
  spark.read \를 반환합니다.
    .format ( "델타") \
    .load(입력 경로)
}

2. 버전 번호 사용
Delta모든 쓰기에는 버전 번호가 있으며 버전 번호를 사용하여
시간을 거슬러 올라갈 수도 있습니다.

Scala 구문:

val df = spark.read
  .format ( "델타")
  .option("버전AsOf", "5238")
  .load ( "/ 경로 / 대상 / 내 / 테이블")

val df = spark.read
  .format ( "델타")
  .load("/path/to/my/table@v5238")

Python 구문:

df = 스파크.read \
  .format ( "델타") \
  .option("버전AsOf", "5238") \
  .load ( "/ 경로 / 대상 / 내 / 테이블")

df = 스파크.read \
  .format ( "델타") \
  .load("/path/to/my/table@v5238")

SQL 구문:

5238 현재 my_table 버전에서 SELECT COUNT(*)
SELECT 개수(*) FROM my_table@v5238
SELECT count(*) FROM delta.'/path/to/my/table@v5238'

감사 데이터 변경

DESCRIBE HISTORY 명령 또는 UI를 통해 테이블 변경 내역을 볼 수 있습니다.

Experiment 및 보고서 재현

시간 이동은 또한 머신 러닝 및 Data Science에서 중요한 역할을 합니다.
모델 및 Experiment 의 재현성은 데이터 사이언티스트의 주요 고려 사항인데,
데이터 사이언티스트는 모델을 프로덕션에 배치하기 전에 100개의 모델을
생성하는 경우가 많으며, 시간이 많이 걸리는 프로세스에서 이전 모델로
돌아가고 싶어하기 때문입니다. 그러나 데이터 관리는 Data Science 도구와
분리되어 있는 경우가 많기 때문에 이를 수행하기가 정말 어렵습니다.

Databricks는 Delta의 시간 이동 기능을 머신 러닝 수명 주기를 위한 오픈 소스
플랫폼인 MLflow와 통합하여 이러한 재현성 문제를 해결합니다.
재현 가능한 머신 러닝 학습의 경우 타임스탬프가 지정된 URL을
MLflow parameter 로 경로에 기록하여 각 학습 작업에 사용된
데이터 버전을 추적할 수 있습니다. 이렇게 하면 이전 설정 및
데이터 세트로 돌아가 이전 모델을 재현할 수 있습니다.
데이터에 대해 업스트림 팀과 조정할 필요가 없으며 다른 Experiment대한
데이터 복제에 대해 걱정할 필요가 없습니다. 이것은 unified analytics의 힘,
Data Science data engineering와 밀접하게 결합 합니다.

롤백

또한 시간 이동을 사용하면 잘못된 쓰기의 경우 롤백을 쉽게 수행할 수 있습니다.
예를 들어 GDPR 파이프라인 작업에 실수로 사용자 정보를 삭제하는 버그가 있는
경우 파이프라인을 쉽게 수정할 수 있습니다.

my_table에 삽입
date_sub(current_date(), 1)my_table 타임스탬프에서 * 선택
여기서 userId = 111

다음과 같이 잘못된 업데이트를 수정할 수도 있습니다.

MERGE INTO my_table 대상
date_sub(current_date(), 1) 소스my_table 기준으로 타임스탬프 사용
ON 소스.userId = target.userId
일치하면 세트 * 업데이트

여러 다운스트림 작업에서 지속적으로 업데이트되는
Delta 테이블 보기 pin

AS OF query사용하면 이제 여러 다운스트림 작업에 대해 지속적으로 업데이트되는
Delta 테이블의 스냅샷을 pin 수 있습니다. Delta 테이블이 지속적으로 업데이트 되고
(예: 15초마다) 이 Delta 테이블에서 주기적으로 읽고 다른 대상을 업데이트하는
다운스트림 작업이 있는 상황을 고려합니다. 이러한 시나리오에서는 일반적으로
모든 대상 테이블이 동일한 상태를 반영하도록 원본 Delta 테이블의 일관된 뷰를 원합니다.
이제 다음과 같은 시나리오를 쉽게 처리할 수 있습니다.

버전 = spark.sql("SELECT max(버전) FROM (기록 설명 my_table)").collect()

# 아래의 모든 운영에 대해 최신 버전의 테이블을 사용합니다.

데이터 = spark.table("my_table@v%s" % 버전[0][0]

data.where("event_type = e1").write.jdbc("테이블1")
data.where("event_type = e2").write.jdbc("테이블2")
...
data.where("event_type = e10").write.jdbc("테이블10")

시계열 분석 간소화를 위한 query

또한 시간 이동은 시계열 분석을 간소화합니다. 예를 들어 지난주에 추가한 신규 고객 수를
확인하려는 경우 query 는 다음과 같이 매우 간단할 수 있습니다.

SELECT 개수(고유 사용자 ID) - (
SELECT 개수(고유 사용자 ID)
date_sub(current_date(), 7))의 my_table 타임스탬프에서
my_table 부터

결론

Delta 의 시간 이동은 개발자 생산성을 크게 향상시킵니다. 도움이 됩니다.

• 데이터 사이언티스트는 Experiment 더 잘 관리합니다.
• 데이터 엔지니어는 파이프라인을 단순화하고 잘못된 쓰기를 롤백합니다.
• 데이터 애널리스트는 쉽게 보고할 수 있습니다.

조직은 마침내 분석을 위해 자체 클라우드 스토리지의 깨끗하고
중앙 집중화되고 버전이 지정된 빅데이터 리포지토리를 표준화할 수 있습니다.
이 새로운 기능으로 무엇을 성취할 수 있을지 기대됩니다.

이 기능은 모든 사용자를 위한 공개 미리 보기로 사용할 수 있습니다.
기능에 대해 자세히 알아보세요.
작동 방식을 확인하려면 Databricks 무료 체험에 등록하세요 .