Delta Live Tables Sink APIによる外部サービスへのデータの書き込み

見逃してました。2月頭のアップデートです。

[Delta Live Tables シンクを使用してパイプラインから外部サービスにデータを書き込む (パブリック プレビュー)](https://docs.databricks.com/aws/ja/release-notes/product/2025/february#delta-live-tables-%E3%82%B7%E3%83%B3%E3%82%AF%E3%82%92%E4%BD%BF%E7%94%A8%E3%81%97%E3%81%A6%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%8B%E3%82%89%E5%A4%96%E9%83%A8%E3%82%B5%E3%83%BC%E3%83%93%E3%82%B9%E3%81%AB%E3%83%87%E3%83%BC%E3%82%BF%E3%82%92%E6%9B%B8%E3%81%8D%E8%BE%BC%E3%82%80-%E3%83%91%E3%83%96%E3%83%AA%E3%83%83%E3%82%AF-%E3%83%97%E3%83%AC%E3%83%93%E3%83%A5%E3%83%BC)

> Delta Live Tables `sink` API はパブリック プレビュー段階です。 Delta Live Tablesシンクを使用すると、パイプラインによって変換されたデータを、Apache Kafka や Azure Event Hubs などのイベント ストリーミング サービスや、Unity Catalog や Hive metastoreによって管理される外部テーブルなどのターゲットに書き込むことができます。[Delta Live Tables シンクを使用して外部サービスにレコードをストリームする](https://docs.databricks.com/aws/ja/dlt/dlt-sinks)を参照してください。

ブログも公開されていました。

https://www.databricks.com/jp/blog/introducing-dlt-sink-api-write-pipelines-kafka-and-external-delta-tables

要は、これまでのDelta Live Tables(DLT)ではストリーミングテーブルとマテリアライズドビューのみを用いてデータの処理を行っていましたが、それだと外部システムとの連携が困難だったり、構造化ストリーミングのシンクを利用したいと言うニーズに応えられなかったので、今回新たに[Sink API](https://docs.databricks.com/ja/delta-live-tables/dlt-sinks.html)を提供するようにしたとのこと。これによって、外部のイベントストリーミングサービスや外部のDeltaテーブルへの書き込みが可能になります。

と言うことで、動かしてみます。

# DLT Sink APIのウォークスルー

## ブロンズレイヤー

```py
import dlt
from pyspark.sql.functions import *

json_path = "/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/"

@dlt.table(
 comment="生のウィキペディアクリックストリームデータセット、databricks-datasetsから取り込まれました。",
 table_properties={
   "quality": "bronze"
 }
)
def clickstream_raw():         
 return (
   spark.readStream.format("cloudFiles").option("cloudFiles.format", "json").option("inferSchema", "true").load(json_path)
 )
```

## シルバーレイヤー

```py
@dlt.table(
 comment="データ型/列名がクリーンアップされ、品質期待値が設定されたウィキペディアクリックストリームデータセット。",
 table_properties={
   "quality": "silver"
 }
)
@dlt.expect("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")
@dlt.expect_or_fail("valid_count", "click_count > 0")
def clickstream_clean():
  return (
    dlt.readStream("clickstream_raw")
     .withColumn("current_page_id", expr("CAST(curr_id AS INT)"))
     .withColumn("click_count", expr("CAST(n AS INT)"))
     .withColumn("previous_page_id", expr("CAST(prev_id AS INT)"))
     .withColumnRenamed("curr_title", "current_page_title")
     .withColumnRenamed("prev_title", "previous_page_title")
     .select("current_page_id", "current_page_title", "click_count", "previous_page_id", "previous_page_title")
  )
```

## ゴールドレイヤー

```py
@dlt.table(
 comment="Apache Sparkページにリンクする最も一般的なページのテーブル。",
 table_properties={
   "quality": "gold" 
 } 
)
def spark_referrers():
 return (
   dlt.readStream("clickstream_clean")
     .filter(expr("current_page_title == 'Apache_Spark'"))
     .withColumnRenamed("previous_page_title", "referrer")
     .select("referrer", "current_page_id", "current_page_title", "click_count")
 )
```

ここまでは従来と同じですね。

## Delta Live Tables sinkを構成

[Delta Live Tables シンクを構成する](https://docs.databricks.com/aws/ja/delta-live-tables/dlt-sinks#delta-live-tables-%E3%82%B7%E3%83%B3%E3%82%AF%E3%82%92%E6%A7%8B%E6%88%90%E3%81%99%E3%82%8B)を参考に、Deltaテーブルシンクを構成します。

```py
dlt.create_sink(
  name = "delta_sink",
  format = "delta",
  options = { "tableName": "users.takaaki_yayoi.my_table" }
)
```

## Delta Live Tables sinkへの書き込みフロー

シンクへの書き込みフローを定義します。

```py
@dlt.append_flow(name = "delta_sink_flow", target="delta_sink")
def delta_sink_flow():
  return(
  spark.readStream.table("spark_referrers")
  .selectExpr("current_page_id", "referrer", "current_page_title", "click_count")
  )
```

## パイプラインの実行

上のノートブックからパイプラインを構成して実行します。

![Screenshot 2025-03-04 at 16.24.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/678db237-a4b7-4e05-b1e4-3166e4c5edb2.png)

新たに**シンク**がパイプラインの最後に現れます。

![Screenshot 2025-03-04 at 16.41.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1697f4d3-531e-435d-b4e6-b5fdff5b82ff.png)

シンクのDeltaテーブルにデータが永続化されました。

![Screenshot 2025-03-04 at 16.42.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ce1bfac9-fdeb-48a4-8ea7-7145548b0a68.png)

リネージもキャプチャされています。

![Screenshot 2025-03-04 at 16.43.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/93aa12b6-4298-43bb-a966-874cbb3cd450.png)

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)