신청하기
빅 데이터

Apache Iceberg 분기 및 AWS Glue Data Quality를 사용하여 Write-Audit-Publish 패턴 구축

처럼
좋아했다
플라톤에 의해 재발행
플라톤에 의해 재발행

시간

20 Min읽기
조회 수 : 222

오늘날 세계에서 데이터의 중요성을 감안할 때, 조직은 지속적으로 유입되는 대규모 데이터를 관리하고 품질과 신뢰성을 검증하는 두 가지 과제에 직면합니다. 고품질 데이터만 게시하는 것의 중요성은 과장할 수 없습니다. 이는 정확한 분석, 신뢰할 수 있는 머신 러닝(ML) 모델 및 건전한 의사 결정의 기반입니다. 마찬가지로 중요한 것은 데이터 무결성을 유지하는 것뿐만 아니라 규정 준수, 오류 분석 및 잠재적인 데이터 복구를 위해 문제가 있는 데이터를 분리하고 감사하는 기능입니다.

AWS 접착제 데이터 품질을 효과적으로 모니터링하고 관리하는 데 사용할 수 있는 서버리스 데이터 통합 ​​서비스입니다. AWS Glue 데이터 품질. 오늘날 많은 고객이 다음을 사용하여 데이터 품질 검증 파이프라인을 구축합니다. 데이터 품질 정의 언어 (DQDL) 정적 규칙이 있기 때문에 동적 규칙및 이상 감지 기능, 매우 간단합니다.

아파치 빙산 오픈 테이블 형식으로 제공됩니다. 원자성, 일관성, 격리성 및 내구성 (ACID) 거래를 데이터 레이크에 전달하여 데이터 관리를 간소화합니다. 주요 기능 중 하나는 다음을 사용하여 데이터를 관리하는 기능입니다. 가지각 지점은 자체 라이프사이클을 가지고 있어 유연하고 효율적인 데이터 관리 전략이 가능합니다.

이 게시물에서는 AWS Glue Data Quality 및 Iceberg 브랜치를 사용하여 Apache Iceberg 테이블에 데이터를 수집할 때 데이터 품질을 유지하기 위한 강력한 전략을 살펴봅니다. 게시된 데이터의 품질을 검증하는 두 가지 일반적인 전략을 논의합니다. WAP(Write-Audit-Publish) 패턴을 자세히 살펴보고 Apache Iceberg에서 작동하는 방식을 보여줍니다.

데이터 품질 관리를 위한 전략

스트리밍 환경에서 데이터 품질을 검토할 때 두 가지 주요 전략이 나타납니다. 배달 못한 편지 대기열 (DLQ) 접근 방식과 WAP 패턴. 각 전략은 고유한 장점과 고려 사항을 제공합니다.

  • DLQ 접근 방식 – 문제가 있는 항목을 고품질 데이터에서 분리하여 깨끗한 데이터만 기본 데이터 세트에 포함되도록 합니다.
  • WAP 패턴 – 브랜치를 사용하여 문제가 있는 항목을 고품질 데이터에서 분리하여 깨끗한 데이터만 메인 브랜치에 게시되도록 합니다.

DLQ 접근 방식

DLQ 전략은 고품질 데이터를 문제가 있는 항목에서 효율적으로 분리하여 깨끗한 데이터만 기본 데이터 세트에 포함되도록 하는 데 중점을 둡니다. 작동 방식은 다음과 같습니다.

  1. 데이터가 스트림으로 들어오면 검증 프로세스를 거칩니다.
  2. 유효한 데이터는 다운스트림 사용자가 참조하는 테이블에 직접 기록됩니다.
  3. 잘못되었거나 문제가 있는 데이터는 추후 분석 및 잠재적 복구를 위해 별도의 DLQ로 리디렉션됩니다.

다음 스크린샷은 이 흐름을 보여줍니다.

bdb4341_0_1_dlq

장점은 다음과 같습니다.

  • 간단 – DLQ 접근 방식은 특히 작성자가 한 명뿐인 경우 구현하기 쉽습니다.
  • 저 레이턴시 – 유효한 데이터는 다운스트림 소비자를 위해 메인 브랜치에서 즉시 사용 가능합니다.
  • 유효하지 않은 데이터에 대한 별도 처리 – 감사 및 복구 목적으로 DLQ를 처리하는 전담 작업을 가질 수 있습니다.

DLQ 전략은 복잡한 데이터 환경에서 상당한 과제를 제시할 수 있습니다. 동일한 Iceberg 테이블에 여러 동시 작성자가 있는 경우 일관된 DLQ 구현을 유지하는 것이 어려워집니다. 이 문제는 쓰기에 다른 엔진(예: Spark, Trino 또는 Python)을 사용하는 경우 더욱 복잡해집니다. DLQ 로직이 서로 다를 수 있기 때문에 시스템 유지 관리가 더 복잡해질 수 있습니다. 또한 잘못된 데이터를 별도로 저장하면 관리 오버헤드가 발생할 수 있습니다.

또한, 저지연 요구 사항의 경우 처리 검증 단계에서 추가 지연이 발생할 수 있습니다. 이는 데이터 품질과 전달 속도의 균형을 맞추는 데 어려움을 겪습니다.

이러한 과제를 합리적인 방법으로 해결하기 위해 다음 섹션에서는 WAP 패턴을 소개합니다.

WAP 패턴

WAP 패턴은 3단계 프로세스를 구현합니다.

  1. 쓰다 – 데이터는 처음에 스테이징 브랜치에 기록됩니다.
  2. 회계 감사 – 스테이징 브랜치에서 품질 검사를 수행합니다.
  3. 게시이다. – 검증된 데이터는 소비를 위해 메인 브랜치에 병합됩니다.

다음 스크린샷은 이 흐름을 보여줍니다.

bdb4341_0_2_wap

장점은 다음과 같습니다.

  • 유연한 데이터 지연 관리 – WAP 패턴에서 원시 데이터는 데이터 검증 없이 스테이징 브랜치에 수집되고, 그런 다음 고품질 데이터는 데이터 검증과 함께 메인 브랜치에 수집됩니다. 이 특성을 통해 스테이징 브랜치에서 긴급하고 대기 시간이 짧은 데이터 처리를 달성하고 메인 브랜치에서 고품질 데이터 처리를 달성할 수 있는 유연성이 있습니다.
  • 통합 데이터 품질 관리 – WAP 패턴은 감사 및 게시 로직을 작성자 애플리케이션에서 분리합니다. 여러 작성자 또는 다양한 데이터 소스가 있어도 품질 관리에 대한 통합된 접근 방식을 제공합니다. 감사 단계는 쓰기 또는 게시 단계에 영향을 미치지 않고 사용자 지정하고 발전시킬 수 있습니다.

WAP 패턴의 주요 과제는 지연 시간이 증가한다는 것입니다. 다단계 프로세스는 필연적으로 다운스트림 소비자의 데이터 가용성을 지연시키며, 이는 거의 실시간 사용 사례에 문제가 될 수 있습니다. 더욱이 이 패턴을 구현하려면 DLQ 접근 방식에 비해 더 정교한 오케스트레이션이 필요하여 개발 시간과 복잡성이 증가할 가능성이 있습니다.

WAP 패턴이 Iceberg에서 작동하는 방식

다음 섹션에서는 WAP 패턴이 Iceberg에서 어떻게 작동하는지 살펴봅니다.

Iceberg의 분기 기능

Iceberg는 데이터 수명 주기 관리를 위한 분기 기능을 제공하는데, 이는 특히 WAP 패턴을 효율적으로 구현하는 데 유용합니다. Iceberg 테이블의 메타데이터는 스냅샷의 기록을 저장합니다. 테이블의 각 변경 사항에 대해 생성된 이러한 스냅샷은 동시 액세스 제어 및 테이블 버전 관리에 기본이 됩니다. 분기는 다른 분기에서 분기된 스냅샷의 독립적인 기록이며, 각 분기는 별도로 참조하고 업데이트할 수 있습니다.

테이블이 생성되면 메인 브랜치로만 시작하고 모든 트랜잭션은 처음에 여기에 기록됩니다. 감사 브랜치와 같은 추가 브랜치를 만들고 엔진을 구성하여 여기에 쓸 수 있습니다. 한 브랜치의 변경 사항은 Spark의 fast_forward 다음 스크린샷에 표시된 대로 절차를 수행합니다.

bdb4341_0_3_iceberg-지점

Iceberg 지점을 관리하는 방법

이 섹션에서는 SparkSQL을 사용하여 Iceberg 브랜치를 관리하는 데 필수적인 작업을 다룹니다. 특히 브랜치를 사용하여 새 브랜치를 만들고, 특정 브랜치에 쓰고 읽고, Spark 세션에 대한 기본 브랜치를 설정하는 방법을 보여드리겠습니다. 이러한 작업은 Iceberg로 WAP 패턴을 구현하기 위한 기반을 형성합니다.

브랜치를 생성하려면 다음 SparkSQL 쿼리를 실행하세요.

ALTER TABLE glue_catalog.db.tbl CREATE BRANCH audit

업데이트할 브랜치를 지정하려면 다음을 사용하십시오. glue_catalog.<database_name>.<table_name>.branch_<branch_name> 통사론:

INSERT INTO glue_catalog.db.tbl.branch_audit VALUES (1, 'a'), (2, 'b');

쿼리할 지점을 지정하려면 다음을 사용하십시오. glue_catalog.<database_name>.<table_name>.branch_<branch_name> 통사론:

SELECT * FROM glue_catalog.db.tbl.branch_audit;

Spark 세션 범위 전체에 대한 브랜치를 지정하려면 브랜치 이름을 Spark 매개변수 spark.wap.branch로 설정합니다. 이 매개변수가 설정된 후 모든 쿼리는 명시적 표현식 없이 지정된 브랜치를 참조합니다.

SET spark.wap.branch = audit

-- audit branch will be updated
INSERT INTO glue_catalog.db.tbl VALUES (3, 'c');

Iceberg 브랜치를 사용하여 WAP 패턴을 구현하는 방법

Iceberg의 분기 기능을 사용하면 단일 Iceberg 테이블로 WAP 패턴을 효율적으로 구현할 수 있습니다. 또한 ACID 트랜잭션 및 스키마 진화와 같은 Iceberg 특성은 여러 동시 작성자와 다양한 데이터를 처리하는 데 유용합니다.

  1. 쓰다- 데이터 수집 프로세스는 메인에서 브랜치를 전환하고 메인 브랜치 대신 감사 브랜치에 업데이트를 커밋합니다. 이 시점에서 이러한 업데이트는 메인 브랜치에만 액세스할 수 있는 다운스트림 사용자에게는 액세스할 수 없습니다.
  2. 감사 – 감사 프로세스는 감사 브랜치의 데이터에 대한 데이터 품질 검사를 실행합니다. 어떤 데이터가 깨끗하고 제공될 준비가 되었는지 지정합니다.
  3. 게시 – 감사 프로세스는 Iceberg를 통해 검증된 데이터를 본점에 게시합니다. fast_forward 절차를 통해 하위 사용자가 사용할 수 있도록 합니다.

이 흐름은 다음 스크린샷에 나와 있습니다.

bdb4341_0_4_wap-w-iceberg-branch

Iceberg로 WAP 패턴을 구현하면 다음과 같은 여러 가지 이점을 얻을 수 있습니다.

  • 단순성 – Iceberg 브랜치는 하나의 테이블 내에서 감사 및 메인과 같은 테이블의 여러 상태를 표현할 수 있습니다. 여러 데이터 컨텍스트를 별도로 그리고 균일하게 처리할 때에도 통합된 데이터 관리를 할 수 있습니다.
  • 동시 작성자 처리 – Iceberg 테이블은 ACID를 준수하므로 여러 개의 판독기 및 작성기 프로세스가 동시에 실행되는 경우에도 일관된 읽기 및 쓰기가 보장됩니다.
  • 스키마 진화 – 수집되는 데이터에 문제가 있는 경우 스키마가 테이블 정의와 다를 수 있습니다. Spark는 Iceberg 테이블에 대한 동적 스키마 병합을 지원합니다. Iceberg 테이블은 일관되지 않은 스키마로 데이터를 쓰기 위해 스키마를 유연하게 발전시킬 수 있습니다. 다음 매개변수를 구성하면 스키마가 변경되면 소스의 새 열이 기존 행에 대해 NULL 값으로 대상 테이블에 추가됩니다. 대상에만 있는 열은 새 삽입에 대해 값이 NULL로 설정되거나 업데이트 중에 변경되지 않습니다.
SET `spark.sql.iceberg.check-ordering` = false

ALTER TABLE glue_catalog.db.tbl SET TBLPROPERTIES (
    'write.spark.accept-any-schema'='true'
)
df.writeTo("glue_catalog.db.tbl").option("merge-schema","true").append()

중간 정리로, WAP 패턴은 데이터 품질과 대기 시간 간의 균형을 관리하는 강력한 접근 방식을 제공합니다. Iceberg 브랜치를 사용하면 동시 작성자와 스키마 진화를 처리하면서 단일 Iceberg 테이블에서 WAP 패턴을 간단히 구현할 수 있습니다.

사용 사례 예시

가정 모니터링 시스템이 실내 온도와 습도를 추적한다고 가정해 보겠습니다. 이 시스템은 데이터를 캡처하여 Iceberg 기반 데이터 레이크에 전송합니다. 아마존 단순 스토리지 서비스 (Amazon S3). 데이터는 대화형 데이터 분석을 위해 matplotlib를 사용하여 시각화됩니다. 시스템의 경우 장치 오작동이나 네트워크 문제와 같은 문제로 인해 일부 또는 잘못된 데이터가 작성되어 잘못된 통찰력이 발생할 수 있습니다. 많은 경우 이러한 문제는 데이터가 데이터 레이크로 전송된 후에만 감지됩니다. 또한 이러한 데이터의 정확성은 일반적으로 복잡합니다.

이러한 문제를 해결하기 위해 이 게시물의 시스템에 Iceberg 브랜치를 사용하는 WAP 패턴을 적용합니다. 이 접근 방식을 통해 데이터 레이크로 들어오는 객실 데이터는 시각화되기 전에 품질이 평가되고, 추가 데이터 분석에 적격한 객실 데이터만 사용되도록 합니다. 브랜치를 사용하는 WAP 패턴을 사용하면 효과적인 데이터 관리를 달성하고 다운스트림 프로세스에서 데이터 품질을 향상시킬 수 있습니다. 이 솔루션은 Apache Spark와 상호 작용하기 위한 관리형 Jupyter Notebook인 AWS Glue Studio Notebook을 사용하여 시연됩니다.

사전 조건

이 사용 사례에는 다음과 같은 전제 조건이 필요합니다.

AWS CloudFormation으로 리소스 설정

먼저 제공된 AWS CloudFormation 템플릿을 사용하여 Iceberg 환경을 빌드하기 위한 리소스를 설정합니다. 템플릿은 다음 리소스를 만듭니다.

  • Iceberg 테이블의 메타데이터 및 데이터 파일을 위한 S3 버킷
  • AWS Glue Data Catalog의 Iceberg 테이블을 위한 데이터베이스
  • An AWS 자격 증명 및 액세스 관리 AWS Glue 작업에 대한 (IAM) 역할

리소스를 배포하려면 다음 단계를 완료하세요.

  1. 왼쪽 메뉴에서 스택 시작.

실행 버튼

  1. 다음 파라미터, Iceberg데이터베이스 이름 기본적으로 설정되어 있습니다. 기본값을 변경할 수도 있습니다. 그런 다음 선택하세요. 다음 보기.
  2. 왼쪽 메뉴에서 다음 보기.
  3. 왼쪽 메뉴에서 AWS CloudFormation이 사용자 지정 이름으로 IAM 리소스를 생성 할 수 있음을 인정합니다.
  4. 왼쪽 메뉴에서 문의하기.
  5. 스택 생성이 완료된 후 다음을 확인하세요. 출력 리소스 값은 다음 섹션에서 사용됩니다.

다음으로, Iceberg JAR 파일을 세션에 구성하여 Iceberg 브랜치 기능을 사용합니다. 다음 단계를 완료하세요.

  1. 다음 JAR 파일을 선택하세요 Iceberg 릴리스 페이지 로컬 컴퓨터에 다음 JAR 파일을 다운로드하세요.
    1. 1.6.1 Spark 3.3_with Scala 2.12 런타임 Jar
    2. 1.6.1 aws-bundle Jar
  2. 열기 Amazon S3 콘솔 그리고 CloudFormation 스택을 통해 생성한 S3 버킷을 선택합니다. S3 버킷 이름은 CloudFormation에서 찾을 수 있습니다. 출력 탭.
  3. 왼쪽 메뉴에서 폴더 만들기 S3 버킷에 jar 경로를 생성합니다.
  4. 다운로드한 두 개의 JAR 파일을 업로드하세요 s3://<IcebergS3Bucket>/jars/ S3 콘솔에서.

AWS Glue Studio에 Jupyter Notebook 업로드

CloudFormation 스택을 시작한 후 AWS Glue와 함께 Iceberg를 사용하기 위해 AWS Glue Studio 노트북을 만듭니다. 다음 단계를 완료합니다.

  1. 다운로드 wap.ipynb.
  2. 엽니다 AWS Glue 스튜디오 콘솔.
  3. $XNUMX Million 미만 직업 만들기, 고르다 수첩.
  4. 선택 노트북 업로드선택한다. 파일 선택, 다운로드한 노트북을 업로드하세요.
  5. IAM 역할 이름을 선택하세요. 아이스버그WAPGlueJobRole, CloudFormation 스택을 통해 생성한 것입니다. 그런 다음, 선택 노트 만들기.
  6. 럭셔리 직업 이름 페이지 왼쪽 상단에 다음을 입력하세요. iceberg_wap.
  7. 왼쪽 메뉴에서 찜하기.

Iceberg 지점 구성

먼저 실내 온도 및 습도 데이터 세트를 포함하는 Iceberg 테이블을 만듭니다. Iceberg 테이블을 만든 후 WAP 연습을 수행하는 데 사용되는 브랜치를 만듭니다. 다음 단계를 완료합니다.

  1. Jupyter Notebook에서 생성한 AWS Glue Studio에 Jupyter Notebook 업로드, 다음 셀을 실행하여 Iceberg를 Glue와 함께 사용합니다. %additional_python_modules pandas==2.2 pandas를 사용하여 노트북의 온도 및 습도 데이터를 시각화하는 데 사용됩니다. 셀을 실행하기 전에 다음을 대체합니다. <IcebergS3Bucket> Iceberg JAR 파일을 업로드한 S3 버킷 이름을 입력합니다.

bdb4341_1_세션-구성

  1. 다음 셀을 실행하여 SparkSession을 초기화합니다. 처음 세 가지 설정은 다음으로 시작합니다. spark.sql, Glue와 함께 Iceberg를 사용하려면 필요합니다. 기본 카탈로그 이름은 다음과 같이 설정됩니다. glue_catalog 사용 spark.sql.defaultCatalog. 구성 spark.sql.execution.arrow.pyspark.enabled 가 true 그리고 판다스를 이용한 데이터 시각화에 사용됩니다.

bdb4341_2_sparksession-초기화

  1. 세션이 생성된 후(알림) Session <Session Id> has been created. 노트북에 표시됩니다. 다음 명령을 실행하여 CloudFormation 스택을 통해 만든 S3 버킷에 온도 및 습도 데이터 세트를 복사합니다. 셀을 실행하기 전에 다음을 대체합니다. <IcebergS3Bucket> CloudFormation에서 찾을 수 있는 Iceberg의 S3 버킷 이름을 사용합니다. 출력 탭.
!aws s3 cp s3://aws-blogs-artifacts-public/artifacts/BDB-4341/data/part-00000-fa08487a-43c2-4398-bae9-9cb912f8843c-c000.snappy.parquet s3://<IcebergS3Bucket>/src-data/current/ 
!aws s3 cp s3://aws-blogs-artifacts-public/artifacts/BDB-4341/data/new-part-00000-e8a06ab0-f33d-4b3b-bd0a-f04d366f067e-c000.snappy.parquet s3://<IcebergS3Bucket>/src-data/new/

  1. 데이터 소스 버킷 이름과 경로를 구성합니다.DATA_SRC), Iceberg 데이터웨어하우스 경로(ICEBERG_LOC), 그리고 Iceberg 테이블에 대한 데이터베이스 및 테이블 이름(DB_TBL). 바꾸다 <IcebergS3Bucket> CloudFormation의 S3 버킷을 사용하여 출력 탭.
  2. 데이터 세트를 읽고 CTAS(Create Table As Select) 쿼리를 사용하여 데이터 세트를 사용하여 Iceberg 테이블을 만듭니다.

bdb4341_3_ctas

  1. 다음 코드를 실행하여 Iceberg 테이블의 각 객실에 대한 온도 및 습도 데이터를 표시합니다. Pandas와 matplotlib을 사용하여 각 객실의 데이터를 시각화합니다. 다음 스크린샷에서 볼 수 있듯이 10:05에서 10:30까지의 데이터가 노트북에 표시되며, 각 객실은 온도(파란색 선으로 표시)가 약 25°C, 습도(주황색 선으로 표시)가 52%입니다.
import matplotlib.pyplot as plt
import pandas as pd

CONF = [
    {'room_type': 'myroom', 'cols':['current_temperature', 'current_humidity']},
    {'room_type': 'living', 'cols':['current_temperature', 'current_humidity']},
    {'room_type': 'kitchen', 'cols':['current_temperature', 'current_humidity']}
]

fig, axes = plt.subplots(nrows=3, ncols=1, sharex=True, sharey=True)
for ax, conf in zip(axes.ravel(), CONF):
    df_room = spark.sql(f"""
        SELECT current_time, current_temperature, current_humidity, room_type
        FROM {DB_TBL} WHERE room_type = '{conf['room_type']}'
        ORDER BY current_time ASC
        """)
    pdf = df_room.toPandas()
    pdf.set_index(pdf['current_time'], inplace=True)
    plt.xlabel('time')
    plt.ylabel('temperature/humidity')
    plt.ylim(10, 60)
    plt.yticks([tick for tick in range(10, 60, 10)])
    pdf[conf['cols']].plot.line(ax=ax, grid=True, figsize=(8, 6), title=conf['room_type'], legend=False, marker=".", markersize=2, linewidth=0)

plt.legend(['temperature', 'humidity'], loc='center', bbox_to_anchor=(0, 1, 1, 5.5), ncol=2)

%matplot plt

bdb4341_4_비스-1

  1. Iceberg 테이블에 데이터를 쓰기 전에 다음 쿼리를 실행하여 Iceberg 브랜치를 만듭니다. 다음을 통해 Iceberg 브랜치를 만들 수 있습니다. ALTER TABLE db.table CREATE BRANCH <branch_name> 질문.
ALTER TABLE iceberg_wap_db.room_data CREATE BRANCH stg
ALTER TABLE iceberg_wap_db.room_data CREATE BRANCH audit

이제 Iceberg를 사용하여 WAP 패턴을 구축할 준비가 되었습니다.

Iceberg로 WAP 패턴 구축

이전에 만든 Iceberg 브랜치를 사용하여 WAP 패턴을 구현합니다. 새로 들어오는 온도 및 습도 데이터를 잘못된 값을 포함하여 쓰기 시작합니다. stg 아이스버그 테이블의 분기.

쓰기 단계: 들어오는 데이터를 Iceberg에 쓰기 stg 지사

들어오는 데이터를 쓰려면 stg Iceberg 테이블의 분기를 완료하려면 다음 단계를 완료하세요.

  1. 다음 셀을 실행하고 Iceberg 테이블에 데이터를 쓰세요.

bdb4341_5_쓰기

  1. 기록이 작성된 후 다음 코드를 실행하여 현재 온도 및 습도 데이터를 시각화합니다. stg 다음 스크린샷에서 10:30 이후에 새로운 데이터가 추가된 것을 알 수 있습니다. 출력은 잘못된 판독값을 보여줍니다. 예를 들어 거실의 100:10~35:10 사이의 온도는 약 52°C입니다.
fig, axes = plt.subplots(nrows=3, ncols=1, sharex=True, sharey=True)
for ax, conf in zip(axes.ravel(), CONF):
    df_room_stg = spark.sql(f"""
        SELECT current_time, current_temperature, current_humidity, room_type
        FROM {DB_TBL}.branch_stg WHERE room_type = '{conf['room_type']}'
        ORDER BY current_time ASC
        """)
    pdf = df_room_stg.toPandas()
    pdf.set_index(pdf['current_time'], inplace=True)
    plt.xlabel('time')
    plt.ylabel('temperature/humidity')
    plt.ylim(10, 110)
    plt.yticks([tick for tick in range(10, 110, 30)])
    pdf[conf['cols']].plot.line(ax=ax, grid=True, figsize=(8, 6), title=conf['room_type'], legend=False, marker=".", markersize=2, linewidth=0)

plt.legend(['temperature', 'humidity'], loc='center', bbox_to_anchor=(0, 1, 1, 5.5), ncol=2)

%matplot plt

bdb4341_6_비스-2

오류 기록을 포함한 새로운 온도 데이터가 기록되었습니다. stg 브랜치. 이 데이터는 메인 브랜치에 게시되지 않았기 때문에 다운스트림 측에서는 볼 수 없습니다. 다음으로, 데이터 품질을 평가합니다. stg 분기.

감사 단계: 데이터 품질을 평가합니다. stg 지사

이 단계에서는 온도 및 습도 데이터의 품질을 평가합니다. stg AWS Glue Data Quality를 사용하여 브랜치를 만듭니다. 그런 다음 기준을 충족하지 않는 데이터는 데이터 품질 규칙에 따라 필터링되고, 적격 데이터는 최신 스냅샷을 업데이트하는 데 사용됩니다. audit 지점. 데이터 품질 평가부터 시작하세요:

  1. AWS Glue Data Quality를 사용하여 현재 데이터 품질을 평가하려면 다음 코드를 실행하세요. 평가 규칙은 다음에 정의되어 있습니다. DQ_RULESET, 여기서 정상 온도 범위는 장치 사양에 따라 -10~50°C 사이로 설정됩니다. 이 범위를 벗어나는 값은 이 시나리오에서 오류로 간주됩니다.
from awsglue.context import GlueContext
from awsglue.transforms import SelectFromCollection
from awsglue.dynamicframe import DynamicFrame
from awsgluedq.transforms import EvaluateDataQuality
DQ_RULESET = """Rules = [ ColumnValues "current_temperature" between -10 and 50 ]"""


dyf = DynamicFrame.fromDF(
    dataframe=spark.sql(f"SELECT * FROM {DB_TBL}.branch_stg"),
    glue_ctx=GlueContext(spark.sparkContext),
    name='dyf')

dyfc_eval_dq = EvaluateDataQuality().process_rows(
    frame=dyf,
    ruleset=DQ_RULESET,
    publishing_options={
        "dataQualityEvaluationContext": "dyfc_eval_dq",
        "enableDataQualityCloudWatchMetrics": False,
        "enableDataQualityResultsPublishing": False,
    },
    additional_options={"performanceTuning.caching": "CACHE_NOTHING"},
)

# Show DQ results
dyfc_rule_outcomes = SelectFromCollection.apply(
    dfc=dyfc_eval_dq,
    key="ruleOutcomes")
dyfc_rule_outcomes.toDF().select('Outcome', 'FailureReason').show(truncate=False)

  1. 출력은 평가 결과를 보여줍니다. 105°C와 같은 일부 온도 데이터가 정상 온도 범위인 −10~50°C를 벗어났기 때문에 실패를 표시합니다.
+-------+------------------------------------------------------+
|Outcome|FailureReason                                         |
+-------+------------------------------------------------------+
|Failed |Value: 105.0 does not meet the constraint requirement!|
+-------+------------------------------------------------------+

  1. 평가 후 잘못된 온도 데이터를 필터링합니다. stg 지점을 선택한 다음 최신 스냅샷을 업데이트합니다. audit 유효한 온도 데이터가 있는 분기입니다.

bdb4341_7_감사에 쓰기

데이터 품질 평가를 통해 audit 이제 Iceberg 테이블의 분기에는 다운스트림에서 사용할 수 있는 유효한 데이터가 포함되었습니다.

게시 단계: 유효한 데이터를 다운스트림 측에 게시합니다.

감사 브랜치의 유효한 데이터를 메인에 게시하려면 다음 단계를 완료하세요.

  1. 실행 fast_forward 감사 분기의 유효한 데이터를 다운스트림 측에 게시하는 아이스버그 절차입니다.

bdb4341_8_게시

  1. 절차가 완료되면 Iceberg 테이블의 메인 브랜치를 쿼리하여 게시된 데이터를 검토하고 다운스트림 측에서 쿼리를 시뮬레이션합니다.
fig, axes = plt.subplots(nrows=3, ncols=1, sharex=True, sharey=True)
for ax, conf in zip(axes.ravel(), CONF):
    df_room_main = spark.sql(f"""
        SELECT current_time, current_temperature, current_humidity, room_type
        FROM {DB_TBL} WHERE room_type = '{conf['room_type']}'
        ORDER BY current_time ASC
        """)
    pdf = df_room_main.toPandas()
    pdf.set_index(pdf['current_time'], inplace=True)
    plt.xlabel('time')
    plt.ylabel('temperature/humidity')
    plt.ylim(10, 60)
    plt.yticks([tick for tick in range(10, 60, 10)])
    pdf[conf['cols']].plot.line(ax=ax, grid=True, figsize=(8, 6), title=conf['room_type'], legend=False, marker=".", markersize=2, linewidth=0)

plt.legend(['temperature', 'humidity'], loc='center', bbox_to_anchor=(0, 1, 1, 5.5), ncol=2)

%matplot plt

쿼리 결과에는 데이터 품질 평가를 통과한 유효한 온도 및 습도 데이터만 표시됩니다.

bdb4341_9_비스-3

이 시나리오에서는 Iceberg 브랜치와 함께 WAP 패턴을 적용하여 데이터 품질을 성공적으로 관리했습니다. 오류가 있는 레코드를 포함한 실내 온도 및 습도 데이터는 품질 평가를 위해 먼저 스테이징 브랜치에 기록되었습니다. 이 접근 방식은 오류가 있는 데이터가 시각화되어 잘못된 통찰력으로 이어지는 것을 방지했습니다. AWS Glue Data Quality에서 데이터가 검증된 후 유효한 데이터만 메인 브랜치에 게시되고 노트북에서 시각화되었습니다. Iceberg 브랜치와 함께 WAP 패턴을 사용하면 검증된 데이터만 추가 분석을 위해 다운스트림 측으로 전달되도록 할 수 있습니다.

자원 정리

리소스를 정리하려면 다음 단계를 완료하세요.

  1. 에 Amazon S3 콘솔, S3 버킷을 선택하세요 aws-glue-assets-<ACCOUNT_ID>-<REGION> 노트북 파일(iceberg_wap.ipynb)이 저장되어 있습니다. 해당 위치에 있는 Notebook 파일을 삭제합니다. notebook 통로.
  2. CloudFormation 템플릿을 통해 만든 S3 버킷을 선택하세요. 버킷 이름은 다음에서 얻을 수 있습니다. IcebergS3Bucket CloudFormation의 키 출력 탭. 버킷을 선택한 후 선택하세요 빈 모든 객체를 삭제합니다.
  3. 버킷이 비어 있는지 확인한 후 CloudFormation 스택을 삭제합니다. iceberg-wap-baseline-resources.

결론

이 게시물에서는 Apache Iceberg 테이블에 데이터를 수집할 때 데이터 품질을 유지하기 위한 일반적인 전략을 살펴보았습니다. 단계별 지침은 Iceberg 브랜치로 WAP 패턴을 구현하는 방법을 보여주었습니다. 데이터 품질 검증이 필요한 사용 사례의 경우 WAP 패턴은 다운스트림 애플리케이션에 영향을 주지 않고 동시 작성자 애플리케이션에서도 데이터 대기 시간을 관리할 수 있는 유연성을 제공합니다.

저자에 관하여

다나카 토모히로 Amazon Web Services의 Senior Cloud Support Engineer입니다. 그는 고객이 AWS에서 데이터 레이크에 Apache Iceberg를 사용하도록 돕는 데 열정적입니다. 여가 시간에는 동료들과 커피 브레이크를 즐기고 집에서 커피를 만듭니다.

히키타 소타로 솔루션 아키텍트입니다. 그는 다양한 산업, 특히 금융 산업의 고객이 더 나은 솔루션을 구축할 수 있도록 지원합니다. 그는 특히 빅데이터 기술과 오픈소스 소프트웨어에 열정을 갖고 있습니다.

노리 타카 세키 야마 AWS Glue 팀의 수석 빅 데이터 설계자입니다. 그는 일본 도쿄에서 활동하고 있다. 그는 고객을 돕기 위해 소프트웨어 아티팩트를 구축하는 일을 담당하고 있습니다. 여가 시간에는 로드 바이크를 타고 자전거를 즐깁니다.

관련 기사

spot_img

최근 기사

spot_img
프리미엄 콘텐츠

저작권 @ 2024 Plato Technologies Inc.