DatabricksでOSS版Delta Sharingを動かしてみる

Delta Sharingとは

Delta Sharingは、データレイク、レイクハウスに格納されているデータを外部の組織とセキュアに共有するためのプロトコルです。すでにOSS版Delta Sharingが公開されています。DatabricksがホストするDelta Sharingも近いうちにリリースされる予定です。

以下のような特徴があります。

• クライアント(データ受領者)に対して、データセットの特定のサブセットに対してのみアクセス権を指定できるので、データ提供者は簡単にテーブル全体、あるいは特定バージョンのテーブル、テーブルの一部を共有することができます。
• Delta LakeのACIDトランザクションを活用することで、データ提供者はリアルタイム、かつ高い信頼性でデータを更新することができ、データ受領者初音に一貫性のある状態でデータを参照できます。
• データ受領者はデータ提供者と同じプラットフォームを持つ必要はありませんし、クラウドを利用している必要もありません。クラウド間で共有できるのはもちろん、クラウド、オンプレミスでも共有可能です。
• Parquetを理解しているのであれば、Delta Sharingのプロトコルをクライアントで実装するのは容易です。オープンソースのエンジン、BIツールにおけるプロトタイプ実装の多くで構築に要した期間は1、2週間でした。
• 基盤となっているクラウドシステムを活用することで高速、安価かつ、高信頼性、並列性を保ったデータ転送が可能です。

OSS版Delta Sharingを動かしてみる

ここでは、テスト用にホスティングされているDelta Sharingサーバーにアクセスします。

こちらのサーバーは認証を必要としませんが、実運用で利用するシナリオにおいては、呼び出しにBearerトークンが必要になります。

REST APIによるアクセス

以下のようにREST API経由でDelta Sharingサーバーにアクセスすることで共有(share)名を取得することができます。

Bash

%sh curl https://sharing.delta.io/delta-sharing/shares -s | jq '.'

{
  "items": [
    {
      "name": "delta_sharing"
    }
  ]
}

さらに共有されているスキーマを確認します。

Bash

%sh curl https://sharing.delta.io/delta-sharing/shares/delta_sharing/schemas -s | jq '.'

{
  "items": [
    {
      "name": "default",
      "share": "delta_sharing"
    }
  ]
}

スキーマに含まれるテーブルを確認します。

Bash

%sh curl https://sharing.delta.io/delta-sharing/shares/delta_sharing/schemas/default/tables -s | jq '.'

{
  "items": [
    {
      "name": "COVID_19_NYT",
      "schema": "default",
      "share": "delta_sharing"
    },
    {
      "name": "boston-housing",
      "schema": "default",
      "share": "delta_sharing"
    },
    {
      "name": "flight-asa_2008",
      "schema": "default",
      "share": "delta_sharing"
    },
    {
      "name": "lending_club",
      "schema": "default",
      "share": "delta_sharing"
    },
    {
      "name": "nyctaxi_2019",
      "schema": "default",
      "share": "delta_sharing"
    },
    {
      "name": "nyctaxi_2019_part",
      "schema": "default",
      "share": "delta_sharing"
    },
    {
      "name": "owid-covid-data",
      "schema": "default",
      "share": "delta_sharing"
    }
  ]
}

テーブルのデータを取得します。

Bash

%sh curl -X POST https://sharing.delta.io/delta-sharing/shares/delta_sharing/schemas/default/tables/boston-housing/query -s -H 'Content-Type: application/json' -d @- << EOF
{
   "predicateHints" : [
      "date >= '2021-01-01'",
      "date <= '2021-01-31'"
   ],
   "limitHint": 1000
}
EOF

{"protocol":{"minReaderVersion":1}}
{"metaData":{"id":"a76e5192-13de-406c-8af0-eb8d7803e80a","format":{"provider":"parquet"},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"ID\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"crim\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"zn\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"indus\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"chas\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"nox\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"rm\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"age\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"dis\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"rad\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"tax\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"ptratio\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"black\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"lstat\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"medv\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[]}}
{"file":{"url":"https://open-delta-sharing.s3.us-west-2.amazonaws.com/samples/boston-housing/part-00000-70e5db64-7a0c-4648-b2e8-58c1a8cd35dc-c000.snappy.parquet?X-Amz-Security-Token=IQoJb3JpZ2luX2VjEGEaCXVzLXdlc3QtMiJHMEUCIQC%2BTdGKsFxjQ9NKOgvqbiYNNfW8bCaMDyEjZqimaFoSDAIgBpxADvkzoq0rja4NuAnWiRs8xKdKBxqdPZB0YbfmRcEq%2BgMIOhACGgwzMDEwNDQzMjAwMDMiDGlQuIBYJ%2BpPLpFVkCrXAwtHW0AtO1PDPPpO3JSxhhsNF7w2pncC6s3D8gTlZqJJHv%2Fy1pJ8Km9mochVSMKxfxOXWOr4I8%2FtOBMoGfSvFH%2B1okC%2BNyoVMJKQKZ2osJ9rzOuw5zRTSGO382hFVqkePqHVX9wUma4q2NJsJOxcAgYZCG2gDd3LWeCEXk7q23Fk3EWy9o5hYSESSnt0ddblc%2F9IBJWgMIdVR43StpCp%2FaeJd477W7benH52DR%2B7wfqdgyEQrCOKWcvxLSLFkRl1BQnKOk9fyaDboEE1LjKinlH12rAusxWr3AxYs5XF3vr5rP65dsK3jLSQhz8bxb0KIxOJ2eGDjHYgzhzuw6HFdwAxF13ukRMlBYn8g5ZS80GpaSGr0Avxi4pZKNA%2B3WiE52rro5dLgCdUFMWlGgjegWxIqI5RRnYk5FQeXChYr5Pf9uOyED9FgfsG2ANJO18S%2FsjgpH%2FKuPqgYcbV%2BeBJXVxN4qxIdlc20l4S3Jxqp2PX1NSqDzYS5IjS%2FdQyYd3kfI4wrGKuHN5TTySVtEa79gk1Es74Wgs5dUxYyDcyPtUga22Jbgviz9Ayc5ecmmB30e5iW5CIITGxjEBlsDsYBvjBbIbOZE7Degu%2BxNK8YtAsbFVOXzJAcjD2x6WNBjqlAaYKap8SX2SPrcuwoMLYiZXCAFCZql%2FwmytfzYVWpbew3IUihElxHvQipX%2BnqhLq%2FwFQrTGuhCTltDQubXaX0NFnsowicYZvaj29ufOOnhyI1NhAzu6jgsB%2B6mb1SrVNDKw7F2RdAaFWmVr0jfqVOd6gywlj1zRn8amTVHvYbKOkNEr%2BuuS%2Fqt0bs52WCHTkJ41DMYUtyEXb9KooCWlX8WeHGW6L2g%3D%3D&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20211203T012447Z&X-Amz-SignedHeaders=host&X-Amz-Expires=900&X-Amz-Credential=ASIAUMF5DY4BZ5EMEYO4%2F20211203%2Fus-west-2%2Fs3%2Faws4_request&X-Amz-Signature=bafd4b94afcc84ae6ef81b6f0aea38619c3499efb662bc28e460e42b402cefb5","id":"a631e8c0413b821312ee7ace0308aec0","partitionValues":{},"size":27348,"stats":"{\"numRecords\":506,\"minValues\":{\"ID\":1,\"crim\":0.00632,\"zn\":0.0,\"indus\":0.46,\"chas\":0,\"nox\":0.385,\"rm\":3.561,\"age\":2.9,\"dis\":1.1296,\"rad\":1,\"tax\":187,\"ptratio\":12.6,\"black\":0.32,\"lstat\":1.73,\"medv\":5.0},\"maxValues\":{\"ID\":506,\"crim\":88.9762,\"zn\":100.0,\"indus\":27.74,\"chas\":1,\"nox\":0.871,\"rm\":8.78,\"age\":100.0,\"dis\":12.1265,\"rad\":24,\"tax\":711,\"ptratio\":22.0,\"black\":396.9,\"lstat\":37.97,\"medv\":50.0},\"nullCount\":{\"ID\":0,\"crim\":0,\"zn\":0,\"indus\":0,\"chas\":0,\"nox\":0,\"rm\":0,\"age\":0,\"dis\":0,\"rad\":0,\"tax\":0,\"ptratio\":0,\"black\":0,\"lstat\":0,\"medv\":173}}"}}

Python APIを使う

ライブラリをインストールします。

%pip install delta-sharing

設定ファイルを作成します。

Bash

%sh 
cat <<EOT > /dbfs/tmp/takaakiyayoi/open-datasets.share
{
  "shareCredentialsVersion": 1,
  "endpoint": "https://sharing.delta.io/delta-sharing/",
  "bearerToken": "faaie590d541265bcab1f2de9813274bf233"
}
EOT

Python API経由でデータにアクセスします。

Python

import os
import delta_sharing

# 上で作成したファイルをポイントします。ファイルはローカルファイルシステム、あるいは、リモートストレージに格納することができます。
profile_file = "/dbfs/tmp/takaakiyayoi/open-datasets.share"

# SharingClientの作成
client = delta_sharing.SharingClient(profile_file)

# 全ての共有テーブルのリスト
print("########### All Available Tables #############")
print(client.list_all_tables())

########### All Available Tables #############
[Table(name='COVID_19_NYT', share='delta_sharing', schema='default'), Table(name='boston-housing', share='delta_sharing', schema='default'), Table(name='flight-asa_2008', share='delta_sharing', schema='default'), Table(name='lending_club', share='delta_sharing', schema='default'), Table(name='nyctaxi_2019', share='delta_sharing', schema='default'), Table(name='nyctaxi_2019_part', share='delta_sharing', schema='default'), Table(name='owid-covid-data', share='delta_sharing', schema='default')]

Sparkでデータを処理する

大規模なデータを処理する際には、Delta Sharing sparkコネクターを使ってデータをロードします。

注意
クラスターライブラリとしてio.delta:delta-sharing-spark_2.12:0.2.0(Maven)をインストールする必要があります。

Python

import os
import delta_sharing
from pyspark.sql import SparkSession

# フォーマット"deltaSharing"を指定してデータを読み込み
spark.read.format("deltaSharing").load("/tmp/takaakiyayoi/open-datasets.share" + "#delta_sharing.default.boston-housing") \
	 .where("age > 18") \
	 .display()

Databricks版Delta Sharingが利用可能になった際には、改めて記事を投稿します。

サンプルノートブック

上で説明したノートブックです。

Databricks 無料トライアル

Databricks 無料トライアル