Databricks

Foundry を Databricks に接続すると、Databricks 内で利用可能なデータ、コンピューティング、モデルに対してさまざまな機能を活用できます。

サポートされる機能

Unity Catalog で外部アクセスを有効にすると、Databricks コネクターで仮想テーブルを通じて Delta Lake と Apache Iceberg の追加機能を利用できるようになります。設定の詳細については、仮想テーブルセクションを参照してください。

セットアップ

1. Data Connection アプリケーションを開き、画面右上の + 新規ソースを選択します。
2. 利用可能なコネクターの種類から Databricks を選択します。
3. 以下のセクションの情報を使用し、追加の設定プロンプトに従ってコネクターのセットアップを進めます。

Foundry でのコネクターのセットアップについて詳しくはこちら。

接続情報

Databricks コネクターでは、以下の設定オプションを利用できます。

これらの値の取得方法については、Databricks の公式ドキュメント ↗ を参照してください。

認証

Databricks では、以下の方法で認証できます。

すべての機能を利用するには、指定した資格情報に、関連するカタログおよびコンピューティングリソースに対する必要な権限が付与されていることを確認してください。

フェデレーションポリシーの要件

ワークロード ID フェデレーションを使用する場合、Foundry は、フェデレーションポリシーを通じて Databricks が信頼する OIDC 属性トークンを発行します。Databricks で以下の値を使用してポリシーを作成します。これらの値は、この認証方法を選択したときにソース設定パネルにも表示されます。

フェデレーションポリシーの作成方法の詳細については、Databricks の公式ドキュメント ↗ を参照してください。

これらの値は、コード内でも source_configuration.authentication の issuer_url、audience、subject プロパティとして利用できます。また、ポリシーが関連付けられているサービスプリンシパルの service_principal_application_id も利用できます。例については、コードで Databricks ソースを使用するを参照してください。

ネットワーク

Databricks 接続では、Data Connection アプリケーションでソースを設定する際に、適切なエグレスポリシーを追加します。

Databricks コネクターには、設定オプションで指定した Hostname へのポート443でのネットワークアクセスが必要です。これにより、Foundry から Databricks ワークスペースおよび Unity Catalog REST API に接続できるようになります。

Cloud Fetch

Cloud Fetch は Databricks JDBC ドライバーの機能です。Cloud Fetch を使用すると、クラウドストレージを介して Databricks から Foundry にデータを並列で抽出でき、従来のシングルスレッド転送と比較して最大10倍のパフォーマンスを実現します。

有効にすると、Databricks がクエリ結果を一時的に保存するクラウドストレージサービス（AWS S3、Azure Data Lake Storage、または Google Cloud Storage）へのアウトバウンド接続を許可するために、追加のネットワークポリシーが必要になる場合があります。Foundry ワーカー接続を使用している場合は、ワークスペースのストレージバケット用にエグレスポリシーを作成する必要があります。これは、Cloud Fetch が使用するクラウドストレージの場所です。

詳細については、Databricks の公式ドキュメント ↗を参照してください。

ストレージの場所への外部アクセス（仮想テーブルのみ）

このドキュメントの仮想テーブルセクションでは、Unity Catalog の外部アクセスと、それによって利用可能になる機能について詳しく説明しています。外部アクセスには、テーブルのストレージの場所（マネージドまたは外部）へのネットワーク接続が必要です。外部アクセスによって有効になる機能を利用するには、ストレージの場所ごとにエグレスポリシーを作成する必要があります。

エグレスポリシーが対象とするのは Foundry から出るトラフィックのみです。ストレージの場所に適用されているネットワーク制御でも、Foundry からのトラフィックが許可されていることを確認してください。これらの制御は、クラウドプロバイダーによって異なります。Foundry のトラフィックの送信元 IP の特定について、詳しくはこちら。

外部アクセスおよびテーブルのストレージの場所を特定する方法の詳細については、Databricks の公式ドキュメント ↗を参照してください。

例

以下は、Databricks へのネットワーク接続を確保するために設定が必要になる可能性があるエグレスポリシーの例です。

[1] Azure ストレージの場所へのアクセスを設定する際は、必ず blob.core.<endpoint> と dfs.core.<endpoint> の両方のドメインを含めてください。endpoint は Azure Cloud 環境によって異なる場合があります。

Foundry インスタンスと同じリージョンにある S3 バケットに依存するエグレスポリシーについては、対象のバケットに対して、Amazon S3 バケットポリシーのドキュメントに記載されている追加の設定手順を完了していることを確認してください。

Foundry および Databricks の環境によっては、限られたケースで PrivateLink 経由の接続を確立する必要がある場合があります。これは通常、Foundry と Databricks の両方が同じ CSP でホストされている場合です（たとえば、AWS-AWS または Azure-Azure）。お使いの環境に該当すると思われる場合は、Palantir の担当者にお問い合わせのうえ、追加のガイダンスを受けてください。

追加オプション：SSL とホスト名の検証

自己署名証明書を許可するために、追加で JDBC プロパティを指定する必要がある場合があります。

このプロパティが必要かどうかを判断する方法：

• SSL 接続では、サーバー証明書が検証されます。通常、SSL の検証は証明書チェーンを通じて行われます。デフォルトでは、Agent と Foundry ワーカーの両方が、業界標準の証明書チェーンの大部分を信頼します。
• SSL 検証を機能させるには、サーバーが完全な証明書チェーンを提供する必要があります。Databricks サーバーの証明書チェーンを取得するには、openssl s_client -connect {hostname}:{port} -showcerts コマンドを実行し、OpenSSL コマンドラインユーティリティまたはその他の利用可能なツールを使用してチェーンを検証します。
• 接続先のサーバーに自己署名証明書がある場合、またはファイアウォールが接続に対して TLS インターセプションを実行する場合は、コネクターがその証明書を信頼する必要があります。Agent ベースの接続での証明書の使用について、詳しくはこちら。
• Foundry ワーカー接続を作成し、自己署名証明書を使用する場合は、AllowSelfSignedCerts=1 の JDBC プロパティを追加する必要があります。

自己署名証明書を許可するプロパティを追加する方法：

• 接続情報ページの下部にある接続設定で、追加オプション、JDBC プロパティの順に選択します。
• JDBC プロパティの設定で、プロパティを追加、新しいプロパティの順に選択し、キーに AllowSelfSignedCerts、値に 1 を入力します。

AllowSelfSignedCerts プロパティおよびその他の JDBC プロパティは、Databricks ドライバーのドキュメント ↗で説明されています。このドキュメントで説明されている JDBC プロパティは Databricks ドライバー固有のもので、ほかのソースの種類とは異なります。

仮想テーブル

仮想テーブルを使用すると、Databricks Unity Catalog に登録されたデータに接続できます。これにより、Foundry から Databricks のテーブルの読み取りと書き込みを行えるほか、Foundry のパイプラインから Databricks に計算処理をプッシュダウンできます。このセクションでは、Databricks で仮想テーブルを使用する際の詳細を説明します。このセクションは、Foundry データセットに同期する場合には適用されません。

Databricks コネクターは、仮想テーブルを使用する際に、Delta Lake と Apache Iceberg の機能を利用できる拡張機能を提供します。この機能を使用するには、Unity Catalog で外部アクセスを有効にする必要があります。有効にすると、外部アクセスにより、Foundry は Unity REST API と Iceberg REST カタログを使用してテーブルにアクセスし、基盤となるストレージの場所でデータの読み取りと書き込みを行えます。Unity Catalog の資格情報払い出しを使用して、クラウドオブジェクトストレージへの安全なアクセスを確保します。機能の拡張に加え、これらのテーブルに対する読み取りと書き込みのパフォーマンスも向上する可能性があります。

次の設定を行うと、Databricks コネクターで Delta Lake と Apache Iceberg の機能が自動的に利用できるようになります。

1. Unity Catalog で外部アクセスを有効にします。
2. Foundry からテーブルのストレージの場所への接続を許可するネットワークエグレスポリシーを設定します。
3. Unity Catalog から払い出された資格情報を取得できる資格情報をソースに設定します。

テーブルの読み取りまたは書き込みのための接続は、Delta Lake または Apache Iceberg クライアントを使用して、ストレージの場所に直接行われます。テーブルの読み取りや書き込みに Databricks のコンピューティングは使用されません。アクセスするテーブルの種類の判定など、特定のメタデータ操作には Unity Catalog REST API が使用されます。

外部アクセスの詳細とテーブルのストレージの場所を特定する方法については、Databricks の公式ドキュメント ↗を参照してください。ストレージの場所へのネットワークアクセスを有効にする方法については、このドキュメントのネットワークセクションを参照してください。

外部アクセスを使用するには、Unity Catalog の資格情報払い出しが必要です。資格情報払い出しは、すべてのテーブルの種類やテーブル機能に対応しているわけではありません。たとえば、ビューや行フィルターを持つテーブルは、資格情報払い出しに対応していません。資格情報払い出しとその要件の詳細については、Databricks の公式ドキュメント ↗を参照してください。

上記の要件のいずれかが満たされていない場合、Databricks への接続には JDBC が使用されます。JDBC は、同期に使用されるものと同じ仕組みです。Databricks への JDBC 接続の詳細については、Databricks の公式ドキュメント ↗を参照してください。

以下のテーブルに、Databricks で利用できる仮想テーブルの機能を示します。

Databricks テーブルを入力または出力として使用できる、対応する Foundry ワークフローの詳細については、仮想テーブルのドキュメントを参照してください。外部アクセスが有効かどうかによって、機能が異なる場合があります。

[2] Databricks 仮想テーブルに基づく Spark パイプラインでインクリメンタル処理を利用するには、外部アクセスを有効にする必要があります。インクリメンタル計算には、Delta または Iceberg テーブルを直接操作できることが必要です。Delta テーブルに対するインクリメンタル計算は、Change Data Feed ↗ に依存します。Iceberg テーブルに対するインクリメンタル計算は、インクリメンタル読み取り ↗に依存します。

テーブル形式とストレージの場所

以下のテーブルに、外部アクセスが有効な場合と無効な場合に対応する形式とワークフローの概要を示します。

Spark パイプラインを使用する場合、仮想テーブル出力には外部アクセスを有効にする必要があります。仮想テーブル出力は、Databricks のマネージド Iceberg または外部 Delta テーブルである必要があります。

ソース資格情報の権限

すべての機能を利用するには、ソース接続に指定する資格情報に以下の権限を付与することを推奨します。これらの権限は、目的の継承モデルに応じて、カタログ、スキーマ、またはテーブルに適用する必要があります。

外部テーブルを使用する場合、使用する外部の場所に対して BROWSE、CREATE EXTERNAL TABLE、EXTERNAL USE LOCATION 権限を付与することを推奨します。これらは、仮想テーブル出力を使用して外部テーブルを作成する際に必要です。

また、指定する資格情報には、ソース設定で指定するウェアハウスまたはコンピューティングクラスターの使用権限が必要です。

Unity Catalog での権限管理の詳細については、Databricks の公式ドキュメント ↗を参照してください。

ソース設定の要件

仮想テーブルを使用する場合は、次のソース設定の要件に留意してください。

• Foundry ワーカーのソースを使用する必要があります。仮想テーブルは Agent ワーカー接続の使用をサポートしていません。
• ストレージの場所への推奨ネットワーク設定も含め、このドキュメントのネットワークセクションに記載されているとおりに、双方向接続と許可リスト登録が設定されていることを確認してください。
• Code Repositories で仮想テーブルを使用する場合は、追加で必要なソース設定の詳細について、仮想テーブルのドキュメントを参照してください。
• 接続情報の HTTP パス項目で、ウェアハウスまたはコンピュートクラスターを指定する必要があります。Databricks コンピュートリソースの接続情報の取得については、Databricks の公式ドキュメント ↗を参照してください。

詳細については、上記の接続情報セクションを参照してください。

コンピュートプッシュダウン

Foundry では、仮想テーブルを使用する場合に、計算処理を Databricks にプッシュダウンできます。同じソースに登録された Databricks 仮想テーブルをパイプラインの入力と出力として使用すると、計算処理を完全に Databricks に委譲できます。Python トランスフォームで計算処理を Databricks にプッシュダウンする方法の詳細については、Python のドキュメントを参照してください。Pipeline Builder で計算処理を Databricks にプッシュダウンするには、外部パイプラインのドキュメントを参照してください。

コードで Databricks ソースを使用する

より複雑なシナリオでは、プロコードの代替アプリケーションを使用して Databricks ソースに接続できます。

以下の例では、ワークロード ID フェデレーションで設定されたソースを使用し、Foundry に Databricks のシークレットを保存せずに、外部トランスフォームから Databricks にクエリを実行する方法を示します。これには、Foundry ワーカーのソースと、Databricks のフェデレーションポリシーが必要です。

Databricks SDK for Python ↗（databricks-sdk）が、この交換を代行します。IdTokenSource を通じて、Foundry の ID トークンを SDK に渡します。SDK はそのトークンを Databricks OAuth トークンに交換し、そのトークンの有効期限が切れるたびに交換を繰り返します。Foundry の ID トークンは1時間後に有効期限が切れ、Databricks トークンもその有効期限を引き継ぐため、交換処理を委譲することが重要です。長時間実行されるトランスフォームでは、開始時に1回だけトークンを交換し、実行中ずっと再利用することはできません。

トークンをクライアントに直接渡すには、交換済みのトークンを SDK の設定から読み取ります。以下の例では、Databricks SQL Connector for Python ↗（databricks-sql-connector）でこの方法を使用しています。

ワークスペースのホスト名、HTTP パス、サービスプリンシパルは、コード内で直接設定するのではなく、source_configuration から読み取ります。このプロパティは、ソースの接続設定を公開します。これにより、後からソースが別のワークスペースを使用するように設定されても、トランスフォームは引き続き動作します。

接続設定 API では、ワークロード ID フェデレーションを認証タイプ workflowIdentityFederation で識別します。以下に示すように、source_configuration.authentication.type を確認する際は、この値をそのまま使用してください。

トークン交換とクエリはどちらも Databricks ワークスペースへの HTTPS 呼び出しであるため、ワークスペースのホスト名はソースのエグレスポリシーの対象に含まれている必要があります。Databricks ソースは HTTPS 接続を公開しないため、get_https_connection() が返す組み込み HTTPS クライアントは利用できません。代わりに、クライアントを作成する前に、get_https_proxy_uri() と server_certificates_bundle_path からプロキシと証明書の環境変数を設定します。これらの変数は、Agent プロキシを経由してエグレスをルーティングするソースでは必須であり、それ以外のソースでは効果がありません。

例：アクセスを検証してカタログにクエリを実行する

この例では、サービスプリンシパルとして認証し、属性を解決できることを確認してから、交換済みのトークンを使用してクエリを実行します。アプリケーションデータを読み取るには、クエリをご自身のものに置き換えてください。

この例では、単一のアクセストークンを SQL コネクターに渡します。このトークンは接続が維持されている間は更新されないため、このパターンは、トークンの有効期限が切れる前に完了する短時間のクエリにのみ使用してください。Databricks REST API を使用する長時間実行のワークロードでは、SDK が資格情報を更新できるように、WorkspaceClient を直接使用してください。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
import os

import databricks.sql
import polars as pl
from databricks.sdk import WorkspaceClient, oidc
from databricks.sdk.core import (
    Config,
    credentials_strategy,
    oidc_credentials_provider,
)
from transforms.api import Output, transform
from transforms.external.systems import (
    ResolvedSource,
    Source,
    external_systems,
)


class FoundrySourceIdTokenSource(oidc.IdTokenSource):
    """更新可能な Foundry ID トークンを Databricks SDK に提供します。"""

    def __init__(self, refreshable):
        self._refreshable = refreshable

    def id_token(self) -> oidc.IdToken:
        # このフィールドは Databricks トークンではなく、Foundry ID トークンを保持します。
        return oidc.IdToken(jwt=self._refreshable.get().access_token)


def read_connection_settings(source: ResolvedSource):
    configuration = source.source_configuration

    if configuration.type != "databricks":
        raise ValueError("Expected a Databricks connection")

    authentication = configuration.authentication
    if authentication.type != "workflowIdentityFederation":
        raise ValueError("Expected OIDC federation authentication")

    client_id = authentication.service_principal_application_id
    if not client_id:
        raise ValueError("A service principal application ID is required")

    hostname = configuration.host_name.removeprefix("https://").rstrip("/")

    return hostname, configuration.http_path, client_id


def configure_egress(source: ResolvedSource) -> None:
    proxy = source.get_https_proxy_uri()
    if proxy:
        os.environ["HTTPS_PROXY"] = proxy
        os.environ["HTTP_PROXY"] = proxy

    certificates = source.server_certificates_bundle_path
    if certificates:
        os.environ["REQUESTS_CA_BUNDLE"] = str(certificates)
        os.environ["SSL_CERT_FILE"] = str(certificates)


def create_workspace_client(
    source: ResolvedSource,
    hostname: str,
    client_id: str,
) -> WorkspaceClient:
    refreshable = source.get_session_credentials()

    @credentials_strategy("foundry-source-oidc", [])
    def foundry_oidc(config: Config):
        return oidc_credentials_provider(
            config,
            FoundrySourceIdTokenSource(refreshable),
        )

    return WorkspaceClient(
        host=f"https://{hostname}",
        client_id=client_id,
        credentials_strategy=foundry_oidc,
    )


def get_exchanged_access_token(client: WorkspaceClient) -> str:
    authorization = client.config.authenticate().get("Authorization", "")
    scheme, _, access_token = authorization.partition(" ")

    if scheme.lower() != "bearer" or not access_token:
        raise RuntimeError("Expected a Databricks bearer token")

    return access_token


@external_systems(
    databricks_source=Source("<source_rid>"),
)
@transform.using(
    output=Output("<output_dataset_rid>"),
)
def compute(output, databricks_source: ResolvedSource):
    hostname, http_path, client_id = read_connection_settings(
        databricks_source
    )
    configure_egress(databricks_source)

    client = create_workspace_client(
        databricks_source,
        hostname,
        client_id,
    )

    # 属性の詳細をログに記録したり永続化したりせずに、認証を検証します。
    client.current_user.me()

    # SDK は Foundry ID トークンを Databricks OAuth トークンと交換します。
    access_token = get_exchanged_access_token(client)

    with databricks.sql.connect(
        server_hostname=hostname,
        http_path=http_path,
        access_token=access_token,
    ) as connection:
        with connection.cursor() as cursor:
            cursor.execute("""
                SELECT COUNT(*) AS visible_table_count
                FROM `<catalog_name>`.`information_schema`.`tables`
                WHERE table_schema = '<schema_name>'
            """)
            row = cursor.fetchone()

            if row is None:
                raise RuntimeError("Catalog query returned no result")

            visible_table_count = int(row[0])

    output.write_table(
        pl.DataFrame({
            "check_id": ["databricks_oidc_catalog_access"],
            "identity_check_succeeded": [True],
            "catalog_query_succeeded": [True],
            "visible_table_count": [visible_table_count],
        })
    )

Databricks REST API だけが必要な場合は、WorkspaceClient を直接使用し、get_exchanged_access_token は省略します。SDK が呼び出しのたびに交換済みのトークンを付加し、更新します。

OIDC が有効なソースでセッション資格情報を使用する方法の詳細については、Python でのソースの使用のドキュメントを参照してください。

外部モデル

Unity Catalog に登録された Databricks モデルは、次の方法で Foundry に統合できます。

• 外部ホストモデル
• 外部トランスフォーム

Unity Catalog でモデルを利用可能にする方法の詳細については、Databricks の公式ドキュメント ↗を参照してください。また、Foundry で Databricks 外部モデルを設定する方法については、ガイドを参照してください。