DLT Sink API 소개: Kafka 및 외부 Delta 테이블로 파이프라인 작성

Delta Live Tables를 처음 사용하시는 경우, 이 블로그를 읽기 전에 Getting Started with Delta Live Tables를 읽어보시는 것이 좋습니다. 이 자료에서는 DLT(Delta Live Tables) 선언적 ETL 정의 및 문을 사용하여 확장 가능하고 안정적인 파이프라인을 만드는 방법을 설명합니다.

소개

Delta Live Tables (DLT) 파이프라인은 Databricks 내에서 안정적이고 유지 관리 가능하며 테스트 가능한 데이터 처리 파이프라인을 구축하기 위한 강력한 플랫폼을 제공합니다. 선언적 프레임워크를 활용하고 최적의 서버리스 컴퓨팅을 자동으로 프로비저닝함으로써 DLT는 스트리밍, 데이터 변환 및 관리의 복잡성을 단순화하여 최신 데이터 워크플로우에 대한 확장성과 효율성을 제공합니다.

전통적으로 DLT 파이프라인은 Unity Catalog에서 관리하는 스트리밍 테이블 또는 구체화된 뷰로 데이터를 수집하고 처리하는 효율적인 방법을 제공했습니다. 이 접근 방식은 대부분의 데이터 처리 요구 사항을 충족하지만, 데이터 파이프라인이 외부 시스템과 연결해야 하거나 스트리밍 테이블 또는 구체화된 뷰에 쓰는 대신 Structured Streaming 싱크를 사용해야 하는 경우가 있습니다.

DLT의 새로운 싱크 API의 도입은 사용자가 Apache Kafka, Azure Event Hubs와 같은 외부 이벤트 스트림에 처리된 데이터를 쓰거나 Delta 테이블에 쓸 수 있도록 함으로써 이를 해결합니다. 이 새로운 기능은 DLT 파이프라인의 범위를 확장하여 외부 플랫폼과의 원활한 통합을 가능하게 합니다.

이러한 기능은 현재 공개 미리 보기 상태이며, Databricks Runtime에서 DLT로 더 많은 싱크를 계속 추가하여 궁극적으로 모두 지원할 예정입니다. 현재 작업 중인 다음 기능은 foreachBatch로, 고객이 임의의 데이터 싱크에 쓰고 Delta 테이블에 사용자 지정 병합을 수행할 수 있도록 합니다.

싱크 API는 dlt Python 패키지에서 사용할 수 있으며, 아래와 같이 create_sink()와 함께 사용할 수 있습니다.

API는 싱크를 정의할 때 세 가지 주요 인수를 받습니다.

• 싱크 이름: 파이프라인 내에서 싱크를 고유하게 식별하는 문자열입니다. 이 이름을 통해 싱크를 참조하고 관리할 수 있습니다.
• 형식 지정: 출력 형식을 결정하는 문자열로, "kafka" 또는 "delta"를 지원합니다.
• 싱크 옵션: 키-값 쌍으로 구성된 사전으로, 키와 값 모두 문자열입니다. Kafka 싱크의 경우 Structured Streaming에서 사용할 수 있는 모든 구성 옵션(인증, 파티셔닝 전략 등에 대한 설정 포함)을 활용할 수 있습니다. Kafka 지원 구성 옵션의 전체 목록은 문서를 참조하세요. Delta 싱크는 path 속성을 사용하여 스토리지 경로를 정의하거나 tableName 속성을 사용하여 Unity Catalog의 테이블에 직접 쓰는 것을 허용하여 더 간단한 구성을 제공합니다.

싱크에 쓰기

@append_flow API가 향상되어 싱크 이름으로 식별된 대상 싱크에 데이터를 쓸 수 있게 되었습니다. 전통적으로 이 API는 사용자가 여러 소스에서 단일 스트리밍 테이블로 데이터를 원활하게 로드할 수 있도록 했습니다. 새로운 향상된 기능을 통해 사용자는 이제 특정 싱크에 데이터를 추가할 수도 있습니다. 설정 방법을 보여주는 예는 다음과 같습니다.

파이프라인 구축

이제 Databricks 데이터 세트에 포함된 클릭스트림 데이터를 처리하는 DLT 파이프라인을 구축해 보겠습니다. 이 파이프라인은 Apache Spark 페이지에 연결되는 이벤트를 식별하기 위해 데이터를 구문 분석한 다음 이 데이터를 Event Hubs 및 Delta 싱크 모두에 씁니다. 품질 및 처리 효율성을 향상시키기 위해 데이터를 여러 계층으로 구성하는 메달리온 아키텍처를 사용하여 파이프라인을 구조화할 것입니다.

Auto Loader를 사용하여 Bronze 계층에 원시 JSON 데이터를 로드하는 것으로 시작합니다. 그런 다음 Silver 계층에서 데이터를 정리하고 품질 표준을 적용하여 무결성을 보장합니다. 마지막으로 Gold 계층에서는 현재 페이지 제목이 Apache_Spark인 항목을 필터링하고 spark_referrers라는 테이블에 저장합니다. 이 테이블은 싱크의 소스로 사용됩니다. 전체 코드는 부록을 참조하세요.

Azure Event Hubs 싱크 구성

이 섹션에서는 create_sink API를 사용하여 Event Hubs 싱크를 설정합니다. 이를 위해서는 작동 중인 Kafka 또는 Event Hubs 스트림이 있어야 합니다. 파이프라인은 공유 액세스 정책을 사용하여 Kafka 지원 Event Hubs로 데이터를 스트리밍하며, 연결 문자열은 Databricks Secrets에 안전하게 저장됩니다. 또는 SAS 정책 대신 서비스 주체를 사용하여 통합할 수 있습니다. 연결 속성과 비밀을 적절하게 업데이트해야 합니다. Event Hubs 싱크를 구성하는 코드는 다음과 같습니다.

Delta 싱크 구성

Event Hubs 싱크 외에도 create_sink API를 사용하여 Delta 싱크를 설정할 수 있습니다. 이 싱크는 Databricks 파일 시스템(DBFS)의 지정된 위치에 데이터를 쓰지만, Amazon S3 또는 ADLS와 같은 객체 스토리지 위치에 쓰도록 구성할 수도 있습니다.

Delta 싱크를 구성하는 방법의 예는 다음과 같습니다.

Kafka 및 Delta 싱크를 채우기 위한 흐름 만들기

Event Hubs 및 Delta 싱크가 설정되면 다음 단계는 append_flow 데코레이터를 사용하여 이러한 싱크를 채우는 것입니다. 이 프로세스는 싱크로 데이터를 스트리밍하여 최신 정보로 지속적으로 업데이트되도록 합니다.

Event Hubs 싱크의 경우 value 매개변수는 필수이며, key, partition, headers 및 topic과 같은 추가 매개변수는 선택적으로 지정할 수 있습니다. Kafka 및 Delta 싱크에 대한 흐름을 설정하는 방법의 예는 다음과 같습니다.

applyInPandasWithState 함수도 이제 DLT에서 지원되어 사용자가 DLT 파이프라인 내에서 상태 저장 처리를 위해 Pandas의 강력한 기능을 활용할 수 있습니다. 이 향상된 기능을 통해 익숙한 Pandas API를 사용하여 더 복잡한 데이터 변환 및 집계를 수행할 수 있습니다. DLT 싱크 API를 사용하면 사용자는 이 상태 저장 처리 데이터를 Kafka 토픽으로 쉽게 스트리밍할 수 있습니다. 이 통합은 실시간 분석 및 이벤트 기반 아키텍처에 특히 유용하며, 데이터 파이프라인이 스트리밍 데이터를 외부 시스템으로 효율적으로 처리하고 분배할 수 있도록 합니다.

모두 함께 가져오기

위에서 설명한 접근 방식은 DLT 파이프라인을 구축하여 데이터를 효율적으로 변환하는 동시에 새로운 싱크 API를 사용하여 결과를 외부 Delta 테이블 및 Kafka 지원 Event Hubs로 원활하게 전달하는 방법을 보여줍니다.

이 기능은 실시간 분석 파이프라인에 특히 유용하며, 이상 탐지, 예측 유지 보수 및 기타 시간 민감한 사용 사례와 같은 애플리케이션을 위해 Kafka 스트림으로 데이터를 스트리밍할 수 있습니다. 또한 다운스트림 프로세스를 Kafka 토픽으로 스트리밍 이벤트를 전송하여 즉시 트리거할 수 있는 이벤트 기반 아키텍처를 가능하게 하여 새로 도착한 데이터를 신속하게 처리할 수 있습니다.

Call to Action

DLT Sinks 기능이 모든 Databricks 고객을 위해 Public Preview로 제공됩니다! 이 강력한 새 기능을 사용하면 DLT 파이프라인을 Kafka 및 Delta tables와 같은 외부 시스템으로 원활하게 확장하여 실시간 데이터 흐름과 간소화된 통합을 보장할 수 있습니다. 자세한 내용은 다음 자료를 참조하세요.

• DLT Sinks에 대한 Databricks 문서
• Confluent Cloud를 사용한 DLT의 Kafka Sinks에 대한 데모 시청
• DLT용 Python 언어 참조

Appendix:

Pipeline Code:

(이 글은 AI의 도움을 받아 번역되었습니다. 원문이 궁금하시다면 여기를 클릭해 주세요)