Delta Sharingオープン共有を使用して共有されたデータを読み取る（受信者向け）

Delta Sharing Python コネクタと Spark コネクタをインストールする

共有データに関連するメタデータ (共有されているテーブルのリストなど) にアクセスするには、次の操作を行います。 この例では Python を使用します。

1. delta-sharing Python コネクタをインストールします。

   コピー

   Bash

   pip install delta-sharing
   

2. Apache Spark コネクタをインストールします。

Sparkを使用した共有テーブルの一覧表示

共有内のテーブルを一覧表示します。 次の例では、 <profile-path> を資格情報ファイルの場所に置き換えます。

コピー

Python

import delta_sharing

client = delta_sharing.SharingClient(f"<profile-path>/config.share")

client.list_all_tables()

結果は、テーブルの配列と各テーブルのメタデータです。 次の出力は、2 つのテーブルを示しています。

コピー

Console

Out[10]: [Table(name='example_table', share='example_share_0', schema='default'), Table(name='other_example_table', share='example_share_0', schema='default')]

出力が空の場合、または期待するテーブルが含まれていない場合は、データ プロバイダーにお問い合わせください。

Sparkを使用した共有データへのアクセス

以下の変数を置き換えて、次のコマンドを実行します。

• <profile-path>: 資格情報ファイルの場所。

• <share-name>: テーブルの share= の値。

• <schema-name>: テーブルの schema= の値。

• <table-name>: テーブルの name= の値。

• <version-as-of>：オプション。 データを読み込むテーブルのバージョン。 データ プロバイダーがテーブルの履歴を共有している場合にのみ機能します。 delta-sharing-spark 0.5.0 以上が必要です。

• <timestamp-as-of>：オプション。 指定されたタイムスタンプより前または指定されたタイムスタンプのバージョンでデータをロードします。 データ プロバイダーがテーブルの履歴を共有している場合にのみ機能します。 0.6.0 以上の delta-sharing-spark が必要です。

• Python
• Scala

コピー

delta_sharing.load_as_spark(f"<profile-path>#<share-name>.<schema-name>.<table-name>", version=<version-as-of>)

spark.read.format("deltaSharing")\
.option("versionAsOf", <version-as-of>)\
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")\
.limit(10))

delta_sharing.load_as_spark(f"<profile-path>#<share-name>.<schema-name>.<table-name>", timestamp=<timestamp-as-of>)

spark.read.format("deltaSharing")\
.option("timestampAsOf", <timestamp-as-of>)\
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")\
.limit(10))

以下の変数を置き換えて、次のコマンドを実行します。

• <profile-path>: 資格情報ファイルの場所。

• <share-name>: テーブルの share= の値。

• <schema-name>: テーブルの schema= の値。

• <table-name>: テーブルの name= の値。

• <version-as-of>：オプション。 データを読み込むテーブルのバージョン。 データ プロバイダーがテーブルの履歴を共有している場合にのみ機能します。 delta-sharing-spark 0.5.0 以上が必要です。

• <timestamp-as-of>：オプション。 指定されたタイムスタンプより前または指定されたタイムスタンプのバージョンでデータをロードします。 データ プロバイダーがテーブルの履歴を共有している場合にのみ機能します。 0.6.0 以上の delta-sharing-spark が必要です。

コピー

spark.read.format("deltaSharing")
.option("versionAsOf", <version-as-of>)
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")
.limit(10)

spark.read.format("deltaSharing")
.option("timestampAsOf", <version-as-of>)
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")
.limit(10)

Sparkを使用した共有変更データ フィードへのアクセス

テーブル履歴が共有されており、ソース テーブルで変更データフィード (CDF) が有効になっている場合は、これらの変数を置き換えて次のコマンドを実行することで変更データフィードにアクセスできます。 delta-sharing-spark 0.5.0 以降が必要です。

開始点は 1 つだけ指定する必要があります。

• <profile-path>: 資格情報ファイルの場所。

• <share-name>: テーブルの share= の値。

• <schema-name>: テーブルの schema= の値。

• <table-name>: テーブルの name= の値。

• <starting-version>：オプション。 クエリーの開始バージョン。 長整数型として指定します。

• <ending-version>：オプション。 クエリの終了バージョン。 終了バージョンが指定されていない場合、API は最新のテーブル バージョンを使用します。

• <starting-timestamp>：オプション。 クエリーの開始タイムスタンプは、このタイムスタンプ以上で作成されたバージョンに変換されます。 yyyy-mm-dd hh:mm:ss[.fffffffff]の形式の文字列として指定します。

• <ending-timestamp>：オプション。 クエリーの終了タイムスタンプは、このタイムスタンプ以前に作成されたバージョンに変換されます。 次の形式の文字列として指定します。 yyyy-mm-dd hh:mm:ss[.fffffffff]

• Python
• Scala

コピー

delta_sharing.load_table_changes_as_spark(f"<profile-path>#<share-name>.<schema-name>.<table-name>",
  starting_version=<starting-version>,
  ending_version=<ending-version>)

delta_sharing.load_table_changes_as_spark(f"<profile-path>#<share-name>.<schema-name>.<table-name>",
  starting_timestamp=<starting-timestamp>,
  ending_timestamp=<ending-timestamp>)

spark.read.format("deltaSharing").option("readChangeFeed", "true")\
.option("statingVersion", <starting-version>)\
.option("endingVersion", <ending-version>)\
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")

spark.read.format("deltaSharing").option("readChangeFeed", "true")\
.option("startingTimestamp", <starting-timestamp>)\
.option("endingTimestamp", <ending-timestamp>)\
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")

コピー

spark.read.format("deltaSharing").option("readChangeFeed", "true")
.option("statingVersion", <starting-version>)
.option("endingVersion", <ending-version>)
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")

spark.read.format("deltaSharing").option("readChangeFeed", "true")
.option("startingTimestamp", <starting-timestamp>)
.option("endingTimestamp", <ending-timestamp>)
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")

出力が空の場合、または期待するデータが含まれていない場合は、データ プロバイダーにお問い合わせください。

Spark 構造化ストリーミングを使用した共有テーブルへのアクセス

テーブル履歴が共有されている場合は、共有データをストリーム読み取りできます。 delta-sharing-spark 0.6.0 以降が必要です。

サポートされているオプション:

• ignoreDeletes: データを削除するトランザクションを無視します。

• ignoreChanges: UPDATE 、 MERGE INTO 、 DELETE （パーティション内）、 OVERWRITEなどのデータ変更操作によりソース テーブルでファイルが書き換えられた場合は、更新を再処理します。 変更されていない行は引き続き出力できます。 したがって、ダウンストリームの消費者は重複を処理できる必要があります。 削除はダウンストリームに反映されません。 ignoreChanges は ignoreDeletesを包含します。 したがって、 ignoreChangesを使用すると、ソース テーブルの削除や更新によってストリームが中断されることはありません。

• startingVersion: 開始する共有テーブルのバージョン。 このバージョン (含む) 以降のすべてのテーブルの変更は、ストリーミング ソースによって読み取られます。

• startingTimestamp: 開始するタイムスタンプ。 タイムスタンプ（含む）以降にコミットされたすべてのテーブル変更は、ストリーミング ソースによって読み取られます。 例: "2023-01-01 00:00:00.0"。

• maxFilesPerTrigger: 各マイクロバッチで考慮される新しいファイルの数。

• maxBytesPerTrigger: 各マイクロバッチで処理されるデータの量。 このオプションは「ソフト最大値」を設定します。つまり、バッチはほぼこの量のデータを処理し、最小の入力単位がこの制限より大きい場合にはストリーミング クエリを進めるために制限を超えて処理する場合があります。

• readChangeFeed: 共有テーブルの変更データフィードをストリーム読み取りします。

サポートされていないオプション:

• Trigger.availableNow

構造化ストリーミングクエリのサンプル

• Scala
• Python

コピー

spark.readStream.format("deltaSharing")
.option("startingVersion", 0)
.option("ignoreChanges", true)
.option("maxFilesPerTrigger", 10)
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")

コピー

spark.readStream.format("deltaSharing")\
.option("startingVersion", 0)\
.option("ignoreDeletes", true)\
.option("maxBytesPerTrigger", 10000)\
.load("<profile-path>#<share-name>.<schema-name>.<table-name>")

「Databricksでのストリーミング」も参照してください。

削除ベクトルまたは列マッピングが有効になっているテーブルを読み取る

削除ベクトルは、プロバイダーが共有 Delta テーブルで有効にできるストレージ最適化機能です。 「削除とは何ですか?」を参照してください。 。

Databricks は Delta テーブルの列マッピングもサポートしています。 「Delta Lake 列マッピングを使用した列の名前変更と削除」を参照してください。

プロバイダが削除または列マッピングを有効にしてテーブルを共有している場合は、 delta-sharing-spark 3.1 以降を実行しているコンピュートを使用してテーブルを読み取ることができます。 Databricksクラスターを使用している場合は、 Databricks Runtime 14.1 以降を実行しているクラスターを使用してバッチ読み取りを実行できます。 CDF およびストリーミング クエリには、Databricks Runtime 14.2 以上が必要です。

共有テーブルのテーブル機能に基づいてresponseFormatを自動的に解決できるため、バッチ クエリをそのまま実行できます。

変更データフィード (CDF) を読み取るか、削除または列マッピングが有効になっている共有テーブルでストリーミング クエリを実行するには、追加オプションresponseFormat=deltaを設定する必要があります。

次の例は、バッチ、CDF、およびストリーミング クエリを示しています。

コピー

Scala

import org.apache.spark.sql.SparkSession

val spark = SparkSession
        .builder()
        .appName("...")
        .master("...")
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        .getOrCreate()

val tablePath = "<profile-file-path>#<share-name>.<schema-name>.<table-name>"

// Batch query
spark.read.format("deltaSharing").load(tablePath)

// CDF query
spark.read.format("deltaSharing")
  .option("readChangeFeed", "true")
  .option("responseFormat", "delta")
  .option("startingVersion", 1)
  .load(tablePath)

// Streaming query
spark.readStream.format("deltaSharing").option("responseFormat", "delta").load(tablePath)

Delta Sharing Python コネクタをインストールする

共有されているテーブルの一覧など、共有データに関連するメタデータにアクセスするには、 delta-sharing Python コネクタをインストールする必要があります。

コピー

Bash

pip install delta-sharing

pandas を使用した共有テーブルの一覧表示

共有内のテーブルを一覧表示するには、 <profile-path>/config.share資格情報ファイルの場所に置き換えて、次のコマンドを実行します。

コピー

Python

import delta_sharing

client = delta_sharing.SharingClient(f"<profile-path>/config.share")

client.list_all_tables()

出力が空の場合、または期待するテーブルが含まれていない場合は、データ プロバイダーにお問い合わせください。

pandas を使用して共有データにアクセスする

を使用してPandas Pythonの共有データにアクセスするには、変数を次のように置き換えて以下を実行します。

• <profile-path>: 資格情報ファイルの場所。

• <share-name>: テーブルの share= の値。

• <schema-name>: テーブルの schema= の値。

• <table-name>: テーブルの name= の値。

コピー

Python

import delta_sharing
delta_sharing.load_as_pandas(f"<profile-path>#<share-name>.<schema-name>.<table-name>")

pandas を使用した共有チェンジデータフィードへのアクセス

実行を使用して の共有テーブルの変更データフィードにアクセスするには、次のように変数を置き換えます。PandasPythonデータプロバイダーがテーブルの変更データフィードを共有したかどうかによっては、変更データフィードを利用できない場合があります。

• <starting-version>：オプション。 クエリーの開始バージョン。

• <ending-version>：オプション。 クエリの終了バージョン。

• <starting-timestamp>：オプション。 クエリーの開始タイムスタンプ。 これは、このタイムスタンプ以降に作成されたバージョンに変換されます。

• <ending-timestamp>：オプション。 クエリーの終了タイムスタンプ。 これは、このタイムスタンプ以前に作成されたバージョンに変換されます。

コピー

Python

import delta_sharing
delta_sharing.load_table_changes_as_pandas(
  f"<profile-path>#<share-name>.<schema-name>.<table-name>",
  starting_version=<starting-version>,
  ending_version=<starting-version>)

delta_sharing.load_table_changes_as_pandas(
  f"<profile-path>#<share-name>.<schema-name>.<table-name>",
  starting_timestamp=<starting-timestamp>,
  ending_timestamp=<ending-timestamp>)

出力が空の場合、または期待するデータが含まれていない場合は、データ プロバイダーにお問い合わせください。

要件

• Power BI Desktop 2.99.621.0 以上。

• データ プロバイダーによって共有された資格情報ファイルへのアクセス。 「オープン共有モデルでアクセスを取得する」を参照してください。

Databricksに接続する

Delta Sharing コネクタを使用して Databricks に接続するには、次の手順を実行します。

1. 共有資格情報ファイルをテキスト エディターで開き、エンドポイント URL とトークンを取得します。

2. Power BI Desktop を開きます。

3. [データの取得]メニューで、 Delta Sharingを検索します。

4. コネクタを選択し、[ 接続] をクリックします。

5. 資格情報ファイルからコピーしたエンドポイント URL をDelta Sharing Server URLフィールドに入力します。

6. 必要に応じて、[ 詳細オプション ] タブで、ダウンロードできる行の最大数の [行制限 ] を設定します。 これは、デフォルトで 100 万行に設定されます。

7. 「OK」をクリックします。

8. Authenticationの場合は、資格情報ファイルから取得した Windows をBearer ウイルスにコピーします。

9. 「接続」をクリックします。

Power BI Delta Sharing コネクタの制限事項

Power BI Delta Sharingコネクタには次の制限があります。

• コネクタがロードするデータは、マシンのメモリに収まる必要があります。 この要件を管理するために、コネクタは、インポートされる行の数を、Power BI Desktop の [詳細オプション] タブで設定した行制限に制限します。

要件

• Tableau Desktop および Tableau Server 2024.1 以上

• データ プロバイダーによって共有された資格情報ファイルへのアクセス。 「オープン共有モデルでアクセスを取得する」を参照してください。

Databricksに接続する

Delta Sharing コネクタを使用して Databricks に接続するには、次の手順を実行します。

1. Tableau Exchangeにアクセスし、指示に従って Delta Sharing Connector をダウンロードし、適切なデスクトップ フォルダーに配置します。

2. Tableau Desktop を開きます。

3. コネクタページで、「Delta Sharing by Databricks」を検索します。

4. [ 共有ファイルのアップロード] を選択し、プロバイダーによって共有された資格情報ファイルを選択します。

5. [ データを取得] をクリックします。

6. Data Explorerでテーブルを選択します。

7. 必要に応じて、SQL フィルターまたは行制限を追加します。

8. 「 テーブル・データの取得」をクリックします。

Tableau Delta Sharing コネクタの制限

Tableau Delta Sharingコネクタには次の制限があります。

• コネクタがロードするデータは、マシンのメモリに収まる必要があります。 この要件を管理するために、コネクタはインポートされる行の数を Tableau で設定した行制限に制限します。

• すべての列は String型として返されます。

• SQL フィルターは、Delta Sharing サーバーがpredicateHintをサポートしている場合にのみ機能します。