기본적으로 파이프라인 흐름은 Unity 카탈로그에서 관리하는 델타 테이블(일반적으로 스트리밍 테이블 또는 구체화된 뷰)에 결과를 씁니다. 싱크는 이벤트 스트리밍 서비스 또는 사용자 지정 데이터 저장소와 같이 Databricks 관리 스토리지 외부의 대상에 변환된 데이터를 쓸 수 있는 대체 출력 대상입니다.
싱크는 추가 흐름과 함께 사용됩니다. 싱크 API 중 하나를 사용하여 싱크를 정의한 다음, append_flow 정의에서 이를 target로 참조합니다.
싱크를 사용하는 경우
Databricks는 다음을 수행해야 하는 경우 싱크를 사용하는 것이 좋습니다.
- 사기 감지, 실시간 분석 또는 고객 권장 사항과 같이 대기 시간이 짧은 운영 사용 사례를 빌드합니다. 여기서 데이터는 클라우드 스토리지가 아닌 메시지 버스로 전달되어야 합니다. 밀리초 대기 시간이 필요한 워크로드의 경우 Lakeflow Spark 선언적 파이프라인에서 실시간 모드 사용을 참조하세요.
- Unity 카탈로그 관리 테이블 및 외부 테이블을 포함하여 외부 델타 인스턴스에서 관리하는 테이블에 변환된 데이터를 씁니다.
- Azure Databricks 외부에서 사용할 수 있도록 처리된 데이터를 Apache Kafka 토픽에 다시 쓰는 것과 같은 역방향 ETL을 외부 시스템에 수행합니다.
- Python 사용자 지정 데이터 원본을 사용하여 Azure Databricks 기본적으로 지원되지 않는 형식에 씁니다.
싱크 형식
파이프라인은 다음 싱크 유형을 지원합니다.
| 싱크 형식 | Description |
|---|---|
| 델타 테이블 싱크 | Unity 카탈로그의 관리형 Delta 테이블 또는 외부 Delta 테이블에 기록합니다. 파일 경로 또는 정규화된 테이블 이름을 지정합니다. |
| Apache Kafka 데이터 수신점 | 파이프라인 런타임에 포함된 Kafka 커넥터를 사용하여 Apache Kafka 토픽에 씁니다. |
| Azure Event Hubs 데이터 수신 지점 | Kafka 인터페이스를 사용하여 Azure Event Hubs에 쓰기 Kafka 싱크와 동일한 옵션을 사용합니다. |
| Python용 사용자 지정 싱크 |
spark.dataSource.register에 등록된 Python 사용자 지정 데이터 원본을 사용하여 모든 데이터 저장소에 쓸 수 있습니다. |
| ForEachBatch 싱크 | 스트리밍 데이터의 각 마이크로 배치에 사용자 지정 Python 논리를 적용합니다. 여러 대상에 쓰거나, upsert를 수행하거나, 기본적으로 스트리밍 쓰기를 지원하지 않는 대상을 사용해야 하는 경우에 사용합니다. |
수신 API
파이프라인은 싱크를 만들기 위한 두 가지 API를 제공합니다.
-
create_sink(): 지원되는 형식(Delta, Kafka, AEH 또는 Python 사용자 지정 데이터 원본)의 명명된 싱크를 만듭니다. Python만 사용할 수 있습니다. 파이프라인에서 싱크 사용을 참조하세요. -
foreach_batch_sink(): 스트리밍 데이터의 각 마이크로 배치에 대해 실행되는 Python 함수를 데코레이트합니다. 사용자 지정 쓰기 논리에 대한 최대 유연성을 제공합니다. ForEachBatch를 사용하여 파이프라인의 임의 데이터 싱크에 쓰기를 참조하세요.
두 싱크 유형은 모두 append_flow의 target로 참조됩니다.
Limitations
- 싱크는 Python에서만 사용할 수 있습니다. SQL은 지원되지 않습니다.
- 스트리밍 쿼리만 지원됩니다. 일괄 처리 쿼리는 지원되지 않습니다.
-
append_flow만 싱크에 쓸 수 있으며,create_auto_cdc_flow및 기타 흐름 유형은 지원되지 않습니다. - 싱크에는 파이프라인 기대치가 지원되지 않습니다.
- 전체 새로 고침을 실행해도 싱크에서 이전에 작성된 데이터가 정리되지 않습니다.