動機
Delta Sharing というオープンソースソフトウェアをご存じでしょうか？
データ共有のためのプロトコル、サーバ機能やクライアント機能の参考実装を提供するものです。
Delta Sharing のサーバ機能の参考実装を用いると、公式README にも記載の通り、パブリッククラウドのストレージに置かれたデータを共有することができます。（以下、引用）
We support sharing Delta Lake tables on S3, Azure Blob Storage and Azure Data Lake Storage Gen2.
一方で、「手元で試したい」、「うちはオンプレ環境を運用していて…」というケースもあるかと思います。
そこでこの記事では、AWS S3互換機能を有したストレージを利用し、パブリッククラウドではない場所に置かれたデータを共有できるようにしてみるTipsを簡単な説明とともに紹介したいと思います。
なお、 Delta Sharing はまだまだ若いオープンソースソフトウェアですし絶賛進化中なので、もし利用される際には最新の公式情報を都度取得し、十分に検証してから用いるようにして下さい。ここで挙げているTipsも必ずしも常に動作するとは限らないものなので、あくまで動作イメージをつかむための参考情報としてご利用ください。
手短な結論（方法だけ知りたい方向け）
Delta Sharing のサーバ機能を立ち上げる際、 conf ディレクトリの中に以下のような設定ファイルを保存します。
conf/core-site.xml
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
<property>
<name>fs.s3a.access.key</name>
<value>ストレージのアクセスキーを記載して下さい</value>
</property>
<property>
<name>fs.s3a.secret.key</name>
<value>ストレージのシークレットキーを記載してください</value>
</property>
<property>
<name>fs.s3a.endpoint</name>
<value>ストレージのエンドポイントを記載して下さい（例：http://localhost:9000）</value>
</property>
<property>
<name>fs.s3a.path.style.access</name>
<value>true</value>
</property>
</configuration>
この状態で通常通り、ストレージ上のデータのURLを指定しながらサーバ機能を起動すればよいです。
conf/delta-sharing.yaml（サーバの設定ファイル）の例
(snip)
shares:
- name: "share"
schemas:
- name: "schema"
tables:
- name: "table"
location: "s3a://test/sample_data"
(snip)
Delta Sharingとは？
「データ共有」と聞いても「データを渡したいのだったら、ただ渡せばよいのではないか」と思われるかもしれません。
しかし、気心知れた同士でデータを見せ合うだけに限らず、例えば組織、さらには企業を跨いでデータ共有したいとなると、その間には異なる文化、ルール、基盤技術が介在することになり、一筋縄でいかないこともあります。
例えば…「限られた相手にだけ公開したい」、「決められた手続き・処理に従って共有したい」、「手元のデータの一部を共有したい」、「自分と同じプラットフォームを利用していない相手にも共有したい」、「手元のデータ構造を維持したまま、共有用に再構造化して見せたい」などなど。挙げはじめると、色々なことが思い浮かびます。
そこでデータ共有のためのプロトコルをオープンに定め、あわせて多くの人が利用しやすいようサーバ機能、クライアント機能の参考実装を公開しているのが Delta Sharing です。
NTTデータのDelta Sharingについての記事 にも簡単なイメージと共に紹介されていますので、よろしければ合わせてご覧くださいませ。
Delta Sharingのサーバ機能について
前述の通り、データを第三者に共有するためのプロトコルを実装したサーバ機能の参考実装が公開されています。
公式READMEのサーバ機能の説明 の通り、あらかじめビルドされたパッケージか、もしくは自分でビルドしてサービスを起動することで、
設定ファイル（例：conf/delta-sharing.yaml）内で指定したデータを第三者に共有できるようになります。
先の「設定ファイル」は以下のような、「共有する対象データの場所に関する情報」や「共有時の論理構造の情報」を含んでいます。
conf/delta-sharing.yaml の例
(snip)
shares:
- name: "share"
schemas:
- name: "schema"
tables:
- name: "table1"
location: "s3a://test/sample_data"
- name: "table2"
location: "s3a://test2/sample_data2"
(snip)
つまり、手元の任意の場所に置いてあるデータを、任意の論理構造で相手に共有できるので、「データ共有する際のコンテキスト」を反映しやすくなると考えられます。
さて、今回の記事のポイントは、location で指定されている箇所です。 公式READMEのストレージ設定に関する説明 に示されている通り、パブリッククラウドのストレージサービスを利用できるようになっていることがわかります。
ここに、手元のストレージやオンプレ環境のストレージを指定できるようにするにはどうしたらよいか？というのが本記事の主旨です。
今回のお題設定
Delta Sharing はオープンソースですし、分かりやすい構造をしているため、言ってしまえばストレージ利用する部分を自分で作りこめば、もちろん様々なストレージに対応できます。例えば、参考実装の io/delta/sharing/server/CloudFileSigner.scala あたりを見ると、データ共有用にプリサインドURLを発行するための各種ストレージサービス向けの実装が並んでいることがわかります。そのあたりの実装や PR#56 、 PR#59 のプルリクエストあたりを見れば、どういったことを気を付けながら独自のストレージに対応させるか、理解できるのではないかと思います。
一方で、ストレージ製品、ミドルウェアの中には著名なサービスとの互換性を実現したものもあります。
例えば、オープンソースのストレージである MinIO はAWS S3互換のAPIを利用できます。
ということで、今回は手元で MinIO によるストレージがある仮定で、AWS S3互換機能を利用して、お手軽に
MinIO ==S3プロトコル==> Delta Sharingサーバ ==Delta Sharingプロトコル==> Delta Sharingクライアント
という連携をお試ししてみましょう。
準備（データを置くストレージなど）
MinIOの準備
手元やオンプレ環境で MinIO を起動する方法、S3互換機能を利用する手順については、 MinIO公式ドキュメントのベアメタル環境の説明 をご覧ください。
ひとまず試すだけでしたら、 ベアメタル環境のクイックスタート を見て簡単に動かすだけでも、本記事の実験は可能です。
上記のクイックスタートに基づいて、 MinIO を起動する際、
$ export MINIO_ROOT_USER=myminioaccesskey
$ export MINIO_ROOT_PASSWORD=myminiosecretkey
$ minio server /mnt/data
というような手順で起動することになります。ここで、 MINIO_ROOT_USER や MINIO_ROOT_PASSWORD で定めている環境変数の値が、後ほどAWS S3互換ストレージとしてアクセスする際のアクセスキー、シークレットキーに相当しますのでご注意ください。
上記の例では、 /mnt/data 以下をAWS S3互換ストレージとして利用します。
MinIOの中にDelta Lakeテーブルを作っておく準備
Delta Sharingは、Delta Lake 形式のデータを共有できます。
Delta Lake はストレージにデータを置く際に、履歴を残したり、（ある程度の）トランザクション管理できるようにしたりする技術です。
基本的にはテーブル構造のデータを扱えるため、今回のように「企業の中の業務データ、分析対象データをテーブル形式でストレージに逐次更新しながら保存しておく」ことに適しています。
Delta Lake については NTTデータのDelta Lakeについての記事 にも簡単な絵と共に説明が掲載されているので、よろしければご参照ください。
ここでは準備として、Delta Lake 形式のデータを MinIO 内に保存しておきたいのですが、それには Apache Spark を利用するのが簡単です。
Apache Spark のPythonクライアント機能を利用し、 AWS S3互換機能を利用して MinIO に書き込む例が CreateDummyDataOnMinIO.ipynb に掲載されています。
なお余談ですが、このJupyterノートブックは、Apache Spark のPythonクライアントをJupyter経由で利用している例ですが、 Apache Spark とJupyterを組み合わせて利用するには、 Apache Spark公式ドキュメントの環境変数に関する説明 に記載されているPYSPARK_DRIVER_PYTHON や PYSPARK_DRIVER_PYTHON_OPTS （こちらは記載ないかもしれません）といった環境変数を利用し、pyspark起動時のPythonを指定する（jupyter）方法が簡易です。
例）
$ export PYSPARK_DRIVER_PYTHON="jupyter"
$ export PYSPARK_DRIVER_PYTHON_OPTS="lab --ip 0.0.0.0"
$ pyspark --packages io.delta:delta-core_2.12:1.0.0,org.apache.hadoop:hadoop-aws:3.2.0,io.delta --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"
ここでは、 pyspark コマンドを利用し、 Apache Spark を起動する際、 Delta Lake やAWS S3を利用するためのライブラリ、プロパティを合わせて指定しています。
ここで当該Jupyterノートブックを利用し、 s3a://test/sample_data という場所（ただし、MinIOの中）に Delta Lake 形式のデータを保存したものとします。
Delta Sharingのサーバ機能の設定と起動
Delta Sharing のサーバ機能を起動する際には、以下のような設定ファイルを渡すことになっています。
conf/delta-sharing.yaml の例
# The format version of this config file
version: 1
# Config shares/schemas/tables to share
shares:
- name: "share"
schemas:
- name: "schema"
tables:
- name: "table"
location: "s3a://test/sample_data"
# Set the host name that the server will use
host: "localhost"
# Set the port that the server will listen on. Note: using ports below 1024
# may require a privileged user in some operating systems.
port: 8080
# Set the url prefix for the REST APIs
endpoint: "/delta-sharing"
# Set the timeout of S3 presigned url in seconds
preSignedUrlTimeoutSeconds: 900
# How many tables to cache in the server
deltaTableCacheSize: 10
# Whether we can accept working with a stale version of the table. This is useful when sharing
# static tables that will never be changed.
stalenessAcceptable: false
# Whether to evaluate user provided `predicateHints`
evaluatePredicateHints: false
上記の例では share/schema/table という論理構造でデータを共有していますが、このとき location に指定しているのが、データの実体の場所を表すURLです。
この例では、「AWS S3ストレージ上の s3a://test/sample_data」に置いてあるデータを共有する、ということだとわかります。
このときプロトコルには「s3a」を利用していますが、要はこのプロトコルに互換性を持ったストレージ（今回の例でいえば MinIO など）であれば、指定可能だろうと想像できます。
しかしながら、手元やオンプレ環境のAWS S3互換ストレージを利用するには、そのためのアクセスキー、シークレットキー、エンドポイントに関する情報を Delta Sharing に渡さないといけません。どうやって渡すのでしょうか。
実は、 Delta Sharing は先の紹介した Delta Lake に加え、 Apache Hadoop のライブラリを内部で利用することで簡潔に実装されています。
ストレージのプリサインドURLを取得するための実装を引用します。
io/delta/standalone/internal/DeltaSharedTableLoader.scala:81（2021/12/18時点のmainブランチ）
(snip)
private val fileSigner = withClassLoader {
val tablePath = new Path(tableConfig.getLocation)
val fs = tablePath.getFileSystem(conf)
fs match {
case _: S3AFileSystem =>
new S3FileSigner(deltaLog.dataPath.toUri, conf, preSignedUrlTimeoutSeconds)
case wasb: NativeAzureFileSystem =>
WasbFileSigner(wasb, deltaLog.dataPath.toUri, conf, preSignedUrlTimeoutSeconds)
case abfs: AzureBlobFileSystem =>
AbfsFileSigner(abfs, deltaLog.dataPath.toUri, preSignedUrlTimeoutSeconds)
case _ =>
throw new IllegalStateException(s"File system ${fs.getClass} is not supported")
}
}
(snip)
ここで org.apache.hadoop.fs.Path#getFileSystem メソッドを利用し、もともと Apache Hadoop が対応している各種ストレージを利用できるようになっています。
ということは、利用しているストレージに関する設定を渡すには、Apache Hadoop の設定を通じて渡せばよいのではないか？と想像できます。
そこで、 Apache Hadoop の設定ファイルとして conf/core-site.xml を作成し、必要なパラメータを渡しながら Delta Sharing のサーバ機能を起動することで、任意のAWS S3互換ストレージを利用できるようになります。
conf/core-site.xml の例
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
<property>
<name>fs.s3a.access.key</name>
<value>ストレージのアクセスキーを記載して下さい</value>
</property>
<property>
<name>fs.s3a.secret.key</name>
<value>ストレージのシークレットキーを記載してください</value>
</property>
<property>
<name>fs.s3a.endpoint</name>
<value>ストレージのエンドポイントを記載して下さい（例：http://localhost:9000）</value>
</property>
<property>
<name>fs.s3a.path.style.access</name>
<value>true</value>
</property>
</configuration>
今回の例では、 MinIO を利用していました。
上記設定ファイルで fs.s3a.access.key や fs.s3a.secret.key には、MinIO 起動時に設定したアクセスキー、シークレットキーを記載してください。
また合わせてエンドポイントを指定しないといけないので、 fs.s3a.endpoint も設定して下さい。
なお、この例ではプロパティで各種キーを渡すようにしていますが、基本的には Apache Hadoop の機能を利用しているので、例えばあらかじめAWSのプロファイル等に設定したうえでそれを利用してアクセスする（クレデンシャルの渡し方を変える）ことも可能だと考えられます。
詳しくは、 Apache Hadoop公式ドキュメントのS3認証設定の説明 をご覧くださいませ。（例えば、 fs.s3a.aws.credentials.provider あたりです）
Delta Sharingサーバ機能の conf ディレクトリについての補足
conf 以下に置いたファイルは読み込まれるようになっています。
初期の Delta Sharing の実装では、任意の設定を渡す方法が乏しかったのですが、 PR#45 のプルリクエストにより当該ディレクトリがクラスパスに加わるようになりました。
fs.s3a.path.style.accessについての補足
上記設定ファイルでは、 fs.s3a.path.style.access を指定しています。
今回の動作確認ではAWS S3の仮想ホスト形式のURLを用いていなかったため、 MinIO との組み合わせでパススタイルのURLを利用するよう、 Apache Hadoop 側の設定を渡しました。当該プロパティについては Apache Hadoopのcore-default.xml を参照ください。
Delta Sharingクライアント機能からのアクセス
ここまでくれば、通常の Delta Sharing クライアント機能によるアクセスで、データ共有されたデータを読み込めるようになっています。
Delta Sharing クライアント機能に渡すプロファイルは通常通り、
{
"shareCredentialsVersion": 1,
"endpoint": "http://localhost:8080/delta-sharing/",
"bearerToken": ""
}
といった内容でよいです。データ共有を受けるユーザ目線では、Delta Sharing 裏側のストレージは表面上は意識しなくても利用できるようになっています。
通常の使い方と変わりませんが、データ共有を受けてデータを読み込む例が LoadDataViaDeltaSharing.ipynb に記載されています。
まとめ
Delta Sharing はデータ共有のためのプロトコル、その参考実装のオープンソースソフトウェアです。
公式READMEのストレージ設定に関する説明 の通り、パブリッククラウドのストレージを利用することを意識した実装になっています。
もちろん、独自のストレージを利用できるよう実装を改変することでも対応できますが、ストレージへのアクセスに Apache Hadoop や Delta Lake のライブラリが用いられていることを利用し、AWS S3互換ストレージれであれば設定を渡すことで簡易に対応できるのではないか？というお試し手段を紹介しました。