Databricks 연결, 외부 저장소 접근과 OIDC

Databricks 커넥터는 Databricks의 데이터, 연산과 모델을 Foundry에서 사용하도록 연결합니다. 원천 탐색, 대량 가져오기, 증분 수집, 가상 테이블, Python Transforms 및 Pipeline Builder의 연산 위임, 외부 모델 연결을 일반 제공(GA)합니다.

Unity Catalog에서 external access를 활성화하면 가상 테이블이 Delta Lake와 Apache Iceberg의 추가 기능을 사용할 수 있습니다. 일반 JDBC 연결과 저장소에 직접 접근하는 연결의 조건을 구분해야 합니다.

소스와 연산 리소스

Data Connection의 +New Source에서 Databricks를 선택합니다.

SQL warehouse 경로는 /sql/<version>/warehouses/<warehouseId>, 연산 클러스터 경로는 sql/protocolv1/o/<workspaceId>/<clusterId> 형식입니다. 실제 리소스의 연결 정보에서 값을 확인합니다.

인증 방식

Azure Databricks의 OAuth M2M에서도 Databricks 서비스 주체의 client ID와 secret을 사용해야 합니다. Microsoft Entra 서비스 주체 인증 정보를 대신 사용하는 방식은 지원하지 않습니다.

서비스 주체 연합 정책

Workload identity federation에서 Foundry가 발급한 OIDC 신원 토큰을 Databricks가 신뢰하도록 다음 값을 설정합니다.

계정 전체 정책이 아니라 서비스 주체의 federation policy를 만들어야 합니다. 계정 전체 정책은 sub에 Databricks 사용자 이름을 기대하지만 Foundry는 소스 RID를 넣으므로 일치하지 않습니다.

코드에서는 source_configuration.authentication의 issuer_url, audience, subject와 service_principal_application_id를 확인할 수 있습니다.

네트워크

작업 공간 Hostname의 443 포트를 허용하면 Databricks 작업 공간과 Unity Catalog REST API에 접근할 수 있습니다. 에이전트 프록시는 동시에 많은 연결이 열려 풀이 소진될 수 있습니다. 연결 풀 오류가 발생하면 프록시의 maxConnections, coreConnections 설정을 조정합니다.

Cloud Fetch

Cloud Fetch는 JDBC 드라이버가 질의 결과를 임시 클라우드 저장소에 두고 병렬로 추출하는 기능입니다. 단일 스레드 전송보다 최대 10배 수준의 성능 개선이 가능하지만 실제 전송 시간은 구성에 따라 달라집니다.

AWS S3, Azure Data Lake Storage 또는 GCS의 작업 공간 저장소 버킷에 대한 추가 외부 통신 정책이 필요할 수 있습니다. 작업 공간 호스트만 허용해서는 이 경로를 모두 사용할 수 없습니다.

테이블 저장소 직접 접근

external access에는 관리형 또는 외부 테이블의 실제 저장소에 대한 연결이 필요합니다. 저장소마다 Foundry의 외부 통신 정책을 추가하고 저장소 측 네트워크 통제도 Foundry 요청을 허용해야 합니다.

호스트 값은 예시입니다. Azure에서는 DFS와 Blob 도메인을 모두 포함하며 cloud 환경에 따라 도메인의 endpoint 부분이 달라질 수 있습니다. Foundry와 같은 리전의 S3 버킷은 추가 버킷 정책 조건도 확인합니다.

Foundry와 Databricks가 같은 클라우드에 있는 일부 환경에서는 PrivateLink가 필요할 수 있으므로 해당 연결 경로를 확인합니다.

인증서와 검증 우회 설정

서버는 전체 인증서 체인을 제공해야 합니다. 공인 체인은 기본 신뢰하지만 자체 서명 인증서나 TLS 중간 검사 환경에는 필요한 인증서를 추가해야 합니다. 인증서 체인을 확인하는 명령 형식은 다음과 같습니다.

openssl s_client -connect {hostname}:{port} -showcerts

Foundry 워커에서 AllowSelfSignedCerts=1을 설정하는 경로도 있습니다. Connection details의 Connection settings, More options, JDBC properties에서 Add property, New property를 선택해 값을 입력합니다.

이 값은 특정 자체 서명 인증서만 신뢰하는 설정이 아닙니다. 1이면 truststore에 대한 서버 인증서 검증과 호스트 이름의 common name/subject alternative name 일치 검증을 비활성화합니다. 적용 범위를 이해하고 보안 요구에 맞게 선택해야 합니다.

가상 테이블의 두 연결 경로

다음 조건을 모두 충족하면 Delta Lake 또는 Iceberg 기능을 자동으로 노출합니다.

1. Unity Catalog의 external access를 활성화합니다.
2. Foundry가 실제 테이블 저장소에 도달하도록 외부 통신 정책을 구성합니다.
3. 소스 인증 정보에 Unity Catalog의 저장소 인증 정보 발급(Credential vending) 권한을 부여합니다.

이 경우 Delta 또는 Iceberg 클라이언트가 저장소를 직접 읽고 쓰며 데이터 처리에 Databricks 연산을 사용하지 않습니다. Unity Catalog REST API는 테이블 유형 확인 같은 메타데이터 작업에 사용합니다.

Credential vending은 모든 테이블 기능을 지원하지 않습니다. 뷰나 행 필터가 있는 테이블은 이 인증 정보 발급 경로를 사용할 수 없습니다. 조건을 충족하지 못하면 일반 동기화와 같은 JDBC 경로로 연결합니다.

기능과 테이블 형식

대량 등록과 자동 등록을 일반 제공합니다. Code Repositories와 Pipeline Builder에서 테이블, 뷰와 구체화 뷰를 입력으로 사용할 수 있습니다. 출력과 증분 기능은 실제 연결 경로와 테이블 형식의 제약을 따릅니다.

Spark 파이프라인의 출력에는 external access가 필요하며 Databricks의 관리형 Iceberg 또는 외부 Delta 테이블을 사용해야 합니다.

Spark 증분 연산에도 external access가 필요합니다. Delta는 Change Data Feed, Iceberg는 Incremental Reads에 의존합니다.

권한

원하는 상속 범위에 따라 카탈로그, 스키마 또는 테이블에 적용합니다. 외부 위치에는 BROWSE, CREATE EXTERNAL TABLE, EXTERNAL USE LOCATION을 부여하는 것이 권장되며 가상 테이블 출력으로 외부 테이블을 만들 때 필요합니다. 지정한 warehouse 또는 연산 클러스터의 사용 권한도 필요합니다.

가상 테이블은 Foundry 워커를 사용해야 하며 에이전트 워커는 지원하지 않습니다. HTTP Path로 warehouse 또는 클러스터를 지정하고 Code Repositories에서 사용할 소스와 테이블의 코드 설정도 수행합니다.

연산 위임

같은 Databricks 소스에 등록한 가상 테이블을 파이프라인 입력과 출력으로 사용하면 연산을 Databricks로 위임할 수 있습니다. Python Transforms와 Pipeline Builder의 외부 파이프라인에서 구성합니다. Spark가 저장소를 직접 읽고 쓰는 경로의 형식 제약과 원천 연산 위임을 구분합니다.

코드에서 OIDC 인증

Databricks 소스의 get_session_credentials().get().access_token은 Foundry가 발급한 OIDC 신원 토큰입니다. 이를 Databricks API의 Bearer 토큰으로 바로 보내면 401이 발생합니다. OAuth 2.0 token exchange로 Databricks OAuth 토큰을 먼저 받아야 합니다.

databricks-sdk의 IdTokenSource에 갱신 가능한 Foundry 토큰을 제공하면 SDK가 교환과 갱신을 수행합니다. Foundry 신원 토큰은 1시간 뒤 만료되고 Databricks 토큰도 그 만료에 영향을 받으므로 장시간 작업이 시작 시 받은 토큰 하나를 계속 재사용해서는 안 됩니다.

workspace hostname, HTTP path와 서비스 주체는 코드에 고정하지 않고 source_configuration에서 읽습니다. 인증 유형의 wire 값은 UI 이름과 달리 정확히 workflowIdentityFederation입니다.

이 소스는 get_https_connection()의 내장 HTTPS 클라이언트를 제공하지 않습니다. 클라이언트를 만들기 전에 get_https_proxy_uri()와 server_certificates_bundle_path를 이용해 프록시 및 인증서 환경 변수를 설정합니다. 에이전트 프록시를 사용하는 소스에서는 이 값이 필요하며 사용하지 않는 소스에는 영향이 없습니다.

신원과 카탈로그 조회 예제

databricks-sdk, databricks-sql-connector, polars를 설치하고 소스 및 출력 RID와 질의의 catalog/schema를 실제 값으로 바꿉니다. 예제는 서비스 주체 신원을 확인하고 교환한 토큰으로 보이는 테이블 수를 질의합니다. 신원 상세값을 로그나 출력에 저장하지 않습니다.

SQL 연결에는 토큰 하나를 전달하며 연결 중 자동 갱신하지 않으므로 만료 전에 끝나는 짧은 질의에 사용합니다. 장시간 REST API 작업에는 WorkspaceClient를 직접 사용해 SDK의 갱신을 유지합니다.

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):
    """Supply refreshable Foundry identity tokens to the Databricks SDK."""

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

    def id_token(self) -> oidc.IdToken:
        # This field holds the Foundry identity token, not a Databricks token.
        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,
    )

    # Validate authentication without logging or persisting identity details.
    client.current_user.me()

    # The SDK exchanges the Foundry identity token for a Databricks OAuth token.
    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],
        })
    )

REST API만 필요하면 get_exchanged_access_token과 SQL 연결을 생략하고 WorkspaceClient 메서드를 호출합니다. SDK가 각 호출에 교환된 토큰을 적용하고 갱신합니다.

외부 모델

Unity Catalog에 등록한 Databricks 모델은 외부 호스팅 모델 연결이나 외부 변환으로 통합할 수 있습니다. 모델을 Unity Catalog에서 사용할 수 있게 등록한 뒤 모델 제공 경로에 맞는 연결을 구성합니다.