Databricks と Confluent をつなぐ：データ連携の全体像を徹底解説

## 概要

本記事では、Databricks と Confluent を連携して活用する方法について整理します。近年、Kafka を中心としたストリーミング基盤の需要が高まり、Confluent Cloud と Databricks はどちらも高いスケーラビリティと豊富な機能を提供するため、組み合わせることで強力なデータ活用基盤を実現できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/663c6476-72ba-3876-e020-c55ba3ed3770.png)

引用元：[Databricks Real-Time Apps with Confluent | Databricks Blog](https://www.databricks.com/blog/2022/01/13/confluent-streaming-for-databricks-build-scalable-real-time-applications-on-the-lakehouse.html)

## Databricks と Confluent の概要

**Databricks** は **Apache Spark** をベースにした高性能なデータ分析プラットフォームです。一方、**Confluent** は、Apache Kafka を中核とするデータストリーミングプラットフォームを提供しており、**Confluent Cloud** やオンプレミス向けの **Confluent Platform** など複数の形態が存在します。  
両者を組み合わせることで、リアルタイムのデータ収集・蓄積・分析をスムーズに行うことができるようになります。

Confluent には Kafka の動作を補完する機能として **Confluent Schema Registry** が用意されています。Kafka で扱うデータ形式（Avro、Protobuf、JSON など）のスキーマを管理できるため、データの進化やスキーマの不整合を防ぐ役割を担います。Kafka を使ってデータを運用する場合は、Schema Registry の連携が非常に重要です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4e9b9cd8-550b-e780-61fc-c14d6a78a726.png)  

引用元：[Schema Registry for Confluent Platform | Confluent Documentation](https://docs.confluent.io/platform/current/schema-registry/index.html)

Confluent 社の公式ページでも **Databricks** との連携方法が紹介されており、いわゆるテクニカルパートナーとして協業関係にあります。また、Confluent はフレームワークとして Flink も利用可能なため、Spark Streaming と機能が部分的に競合する面もあります。  

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/01e2f740-96ff-26f7-b220-915d74ec7be1.png)  

引用元：[A complete real-time event streaming and analytics solution that helps companies make faster and better data driven decisions.](https://www.confluent.io/ja-jp/partner/databricks/)

上記図では Confluent → Databricks の一方向の連携イメージが示されていますが、実際には Databricks → Confluent の双方向連携が可能です。例えば「Confluent → Databricks → Confluent」というエンドツーエンドのフローを下記記事で紹介しておありますので、興味のある方は下記記事を参考にしてください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6d0036cf-3638-435a-c6d3-4b3f14a2169d.png)


引用元：[Confluent と Databricks を連携したリアルタイム分析の実践 #Python - Qiita](https://qiita.com/manabian/items/6634eff1ff2e271cfa07)

## Databricks と Confluent の連携方法

### 連携方法の概要

Databricks 上で Confluent と連携する場合、多くは Kafka フォーマットを利用した構造化ストリーミングでの処理となります。必要なコンポーネントは Databricks Runtime にインストール済みのため、追加の準備は比較的少なくて済むようです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ca9b0137-a4a5-53f3-682e-28c5bfc2ed34.png)

引用元：[Apache Kafka と Azure Databricks を使用したストリーム処理 - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/connect/streaming/kafka#write)

Confluent Schema Registry と連携する方法もあり、2024年12月30日時点では **Avro 形式** と **Protobuf　形式**のシリアライズ／デシリアライズがサポートされています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9da6072c-a801-0938-7977-b1d9359b8c71.png)  
引用元：[ストリーミング Avro データの読み取りと書き込み - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/structured-streaming/avro-dataframe#example-with-schema-registry)  

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eb337f18-90c5-f37b-9a37-5645c6180a57.png)  
引用元：[プロトコル バッファーの読み取りと書き込みを行う - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/structured-streaming/protocol-buffers#use-protobuf-with-confluent-schema-registry)

### Databricks で Confluent からデータを取得（Consume）する方法

Databricks から Confluent のデータを取得する際は、Kafka の設定値をどう指定するかが重要です。Kafka には Consumer Group を設定することが多いですが、Spark 側で状態管理するため、Databricks では推奨されない場合もあります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8a38c2be-f88b-02f6-b0cb-9a2e799e8b24.png)  
引用元：[Apache Kafka と Azure Databricks を使用したストリーム処理 - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/connect/streaming/kafka#configurations)

Confluent の Topic を読み込むと、以下のように `key` と `value` が binary 型のスキーマで取得されるため、目的のデータ形式に合わせてデシリアライズが必要です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/37bc2708-4c23-c9d7-e34f-b9e52cd99965.png)  
引用元：[Apache Kafka と Azure Databricks を使用したストリーム処理 - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/connect/streaming/kafka#schema-for-kafka-records)

Databricks 上で Confluent の Topic データを取得する具体的な方法は、投稿済みの下記記事で紹介しています。

- **Confluent Schema Regsitry を利用する方法**  
  - [Avro 形式の Topic のデータ取得をスキーマレジストリを利用して実施する方法](https://qiita.com/manabian/items/9a6ce47b7eebe80ec693)  
  - [Protobuf 形式の Topic のデータ取得をスキーマレジストリを利用して実施する方法](https://qiita.com/manabian/items/60abe202fc496ad6cb22)
- **スキーマを手動で指定する方法**  
  - [Avro 形式の Topic のデータ取得を手動でスキーマを指定して実施する方法](https://qiita.com/manabian/items/1bf8c50fd29406355000)  
  - [JSON 形式の Topic のデータ取得を手動でスキーマを指定して実施する方法](https://qiita.com/manabian/items/deedb99b9647f193f767)  
  - [Protobuf 形式の Topic のデータ取得を手動でスキーマを指定して実施する方法](https://qiita.com/manabian/items/10940ce43a60117cb1ad)

### Databricks で Confluent へデータを送信（Produce）する方法

Databricks から Confluent へデータを書き込む場合も、Kafka の構造化ストリーミングの下記の設定が必要です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5b3b174e-846f-c225-16f7-62c0937da286.png)  
引用元：[Apache Kafka と Azure Databricks を使用したストリーム処理 - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/connect/streaming/kafka#configure-kafka-structured-streaming-writer)

Confluent へ書き込む前に、Spark データフレーム側で次のスキーマを持たせておく必要があります。`value` 列のみ必須で、それ以外は任意となります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2700cb71-8352-59de-827b-46823e43f71e.png)  
引用元：[Apache Kafka と Azure Databricks を使用したストリーム処理 - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/connect/streaming/kafka#configure-kafka-structured-streaming-writer)

具体的な実装方法は、以下の記事で紹介しています。

- **Confluent Schema Regsitry を利用する方法**  
  - [Avro 形式の Topic データを書き込む際にスキーマレジストリを利用して実施する方法](https://qiita.com/manabian/items/dea4584e418a94ba9549)  
  - [Protobuf 形式の Topic データを書き込む際にスキーマレジストリを利用して実施する方法](https://qiita.com/manabian/items/afa8d8bc75df665d0fe2)
- **スキーマを手動で指定する方法**  
  - [Avro 形式の Topic データを書き込む際に手動でスキーマを指定して実施する方法](https://qiita.com/manabian/items/b6d3c0f596e878371c77)  
  - [JSON 形式の Topic データを書き込む際に手動でスキーマを指定して実施する方法](https://qiita.com/manabian/items/f9ad74831a14a171f940)  
  - [Protobuf 形式の Topic データを書き込む際に手動でスキーマを指定して実施する方法](https://qiita.com/manabian/items/01221449ee43cd5a33ab)

### Databricks から Confluent のオブジェクトを管理する方法

Confluent のオブジェクトを管理するには、Confluent 社が提供する `confluent-kafka` という Python ライブラリを利用できます。Confluent Schema Registry に対して各種操作を行うサンプルも公開していますので、詳しくは以下をご確認ください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3a558750-21a1-a09b-5706-9fdde7c068bc.png)  
引用元：[confluent-kafka · PyPI](https://pypi.org/project/confluent-kafka/)  

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/80f6aa8c-b399-e96a-d3d7-dd556829efc7.png)  
引用元：[Python × Confluent Schema Registry：Confluent Schema Registry 入門 #Kafka - Qiita](https://qiita.com/manabian/items/2516cac0f0ca109d1cc0)

### Confluent で Databricks と連携する方法

Confluent には **Databricks Delta Lake Sink Connector** が提供されており、Confluent Cloud 上から Databricks への書き込みが可能です。現時点では AWS 環境のみがサポートされていますが、興味のある方は以下の公式ドキュメントを参照してください。

- [Set up Databricks Delta Lake (AWS) Sink Connector for Confluent Cloud | Confluent Documentation](https://docs.confluent.io/cloud/current/connectors/cc-databricks-delta-lake-sink/databricks-aws-setup.html)

## 参考リンク

- **Kafka 連携関連**  
  - [Apache Kafka と Azure Databricks を使用したストリーム処理 - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/connect/streaming/kafka)  
  - [Structured Streaming + Kafka Integration Guide - Spark 3.5.4 Documentation](https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html)
- **Avro 形式関連**  
  - [Avro ファイル - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/query/formats/avro)  
  - [ストリーミング Avro データの読み取りと書き込み - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/structured-streaming/avro-dataframe)  
  - [to_avro 関数 - Azure Databricks - Databricks SQL | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/sql/language-manual/functions/to_avro)  
  - [from_avro 関数 - Azure Databricks - Databricks SQL | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/sql/language-manual/functions/from_avro)  
  - [pyspark.sql.avro.functions.from_avro — PySpark master documentation](https://api-docs.databricks.com/python/pyspark/latest/pyspark.sql/api/pyspark.sql.avro.functions.from_avro.html#)  
  - [pyspark.sql.avro.functions.to_avro — PySpark master documentation](https://api-docs.databricks.com/python/pyspark/latest/pyspark.sql/api/pyspark.sql.avro.functions.to_avro.html)
- **JSON 形式関連**  
  - [JSON ファイル - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/query/formats/json)  
  - [to_json 関数 - Azure Databricks - Databricks SQL | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/sql/language-manual/functions/to_json)  
  - [from_json 関数 - Azure Databricks - Databricks SQL | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/sql/language-manual/functions/from_json)  
  - [pyspark.sql.functions.from_json — PySpark master documentation](https://api-docs.databricks.com/python/pyspark/latest/pyspark.sql/api/pyspark.sql.functions.from_json.html)  
  - [pyspark.sql.functions.to_json — PySpark master documentation](https://api-docs.databricks.com/python/pyspark/latest/pyspark.sql/functions/to_json.html)
- **Protobuf 形式関連**  
  - [プロトコル バッファーの読み取りと書き込みを行う - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/structured-streaming/protocol-buffers)  
  - [pyspark.sql.protobuf.functions.from_protobuf — PySpark 3.5.4 documentation](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.protobuf.functions.from_protobuf.html)  
  - [pyspark.sql.protobuf.functions.to_protobuf — PySpark 3.5.4 documentation](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.protobuf.functions.to_protobuf.html)