DLT Sink API 소개: Kafka와 외부 Delta 테이블에 파이프라인 작성하기

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

Delta Live Tables에 처음 접하는 경우, 이 블로그를 읽기 전에 Delta Live Tables 시작하기를 읽어보시는 것을 추천드립니다. 이 문서에서는 Delta Live Tables (DLT) 선언형 ETL 정의와 문장을 사용하여 확장 가능하고 신뢰할 수 있는 파이프라인을 어떻게 생성할 수 있는지 설명하고 있습니다.

소개

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

전통적으로, DLT 파이프라인은 스트리밍 테이블 또는 물리화 된 뷰 로 데이터를 효율적으로 수집하고 처리하는 방법을 제공하였습니다. 이는 Unity 카탈로그에 의해 관리되었습니다. 이 접근 방식은 대부분의 데이터 처리 요구 사항을 충족시키지만, 데이터 파이프라인이 외부 시스템과 연결되어야 하거나 스트리밍 테이블이나 머티리얼라이즈드 뷰에 쓰는 대신 구조화된 스트리밍 싱크를 사용해야 하는 경우가 있습니다.

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

이러한 기능들은 현재 Public Preview에 있으며, 우리는 시간이 지남에 따라 Databricks Runtime에서 DLT로 더 많은 싱크를 추가하고, 결국 모든 것을 지원할 계획입니다. 다음으로 작업 중인 것은 foreachBatch 로, 이를 통해 고객이 임의의 데이터 싱크에 쓰고 Delta 테이블에 사용자 정의 병합을 수행할 수 있습니다.

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

API는 싱크를 정의하는 데 세 가지 주요 인수를 받습니다:

• 싱크 이름: 파이프라인 내에서 싱크를 고유하게 식별하는 문자열입니다. 이 이름을 사용하면 싱크를 참조하고 관리할 수 있습니다.
• 형식 명세: 출력 형식을 결정하는 문자열로, "kafka" 또는 "delta"를 지원합니다.
• 싱크 옵션: 키와 값이 모두 문자열인 키-값 쌍의 사전. Kafka 싱크의 경우, 구조화된 스트리밍에서 사용할 수 있는 모든 구성 옵션을 활용할 수 있으며, 인증, 파티셔닝 전략, 기타 설정을 포함합니다. Kafka가 지원하는 구성 옵션의 전체 목록은 문서 를 참조하십시오. Delta 싱크는 path 속성을 사용하여 저장 경로를 정의하거나 tableName 속성을 사용하여 Unity Catalog의 테이블에 직접 쓸 수 있도록 하여 더 간단한 구성을 제공합니다.

싱크에 쓰기

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

파이프라인 구축

이제 클릭스트림 데이터를 처리하는 DLT 파이프라인을 구축해 봅시다. 이 데이터는 Databricks 데이터셋에 패키지화되어 있습니다. 이 파이프라인은 데이터를 파싱하여 Apache Spark 페이지에 연결하는 이벤트를 식별하고, 이 데이터를 Event Hubs와 Delta 싱크 모두에 쓸 것입니다. 우리는 Medallion Architecture를 사용하여 파이프라인을 구조화할 것이며, 이는 데이터를 다른 계층으로 구성하여 품질과 처리 효율성을 향상시킵니다.

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

Azure Event Hubs Sink 설정

이 섹션에서는 create_sink API를 사용하여 Event Hubs sink를 설정할 것입니다. 이는 당신이 운영 중인 Kafka 또는 Event Hubs 스트림을 가지고 있다고 가정합니다. 우리의 파이프라인은 공유 액세스 정책을 사용하여 Kafka가 활성화된 Event Hubs로 데이터를 스트리밍하며, 연결 문자열은 Databricks Secrets에 안전하게 저장됩니다. 대안으로, SAS 정책 대신 서비스 주체를 통합에 사용할 수 있습니다. 연결 속성과 비밀번호를 적절히 업데이트하십시오. Event Hubs 싱크를 설정하는 코드는 다음과 같습니다:

Delta Sink 설정

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

아래는 Delta sink를 설정하는 방법을 보여주는 예입니다:

Kafka와 Delta 싱크를 채우기 위한 플로우 생성

Event Hubs와 Delta 싱크가 설정되면, 다음 단계는 이러한 싱크를 append_flow 데코레이터를 사용하여 채우는 것입니다. 이 과정은 싱크로 데이터를 스트리밍하고, 최신 정보로 계속 업데이트되도록 하는 것을 포함합니다.

Event Hubs sink의 경우, value 매개변수는 필수이며, key, partition, headers, topic 등의 추가 매개변수는 선택적으로 지정할 수 있습니다. 아래는 Kafka 싱크와 Delta 싱크 모두에 대한 플로우를 설정하는 방법의 예입니다:

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

모든 것을 함께 가져오기

위에서 보여준 접근 방식은 데이터를 효율적으로 변환하는 DLT 파이프라인을 구축하는 방법과 새로운 Sink API를 활용하여 결과를 외부 Delta 테이블과 Kafka가 활성화된 Event Hubs에 원활하게 전달하는 방법을 보여줍니다.

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

행동 요령

DLT Sinks 기능이 이제 모든 Databricks 고객에게 Public Preview로 제공됩니다! 이 강력한 새로운 기능을 통해 Kafka와 Delta 테이블과 같은 외부 시스템으로 DLT 파이프라인을 원활하게 확장할 수 있습니다. 이를 통해 실시간 데이터 흐름과 통합이 간소화되어 보장됩니다. 자세한 정보는 다음 리소스를 참조하십시오:

• Databricks 문서 에서 DLT Sinks에 대해
• 데모 보기 DLT에서 Confluent Cloud를 사용한 Kafka 싱크
• DLT에 대한 Python 언어 참조

부록:

파이프라인 코드: