トップ>テクノロジー>実践！ Azure Databricks のバックエンド・データ通信を閉域化する【Bicep (+α)】

この記事は、NTT docomo Business Advent Calendar 2025 10日目の記事です。

Microsoft の IaC 言語である Bicep (+ Azure CLI/Databricks CLI) を使って、Azure Databricks ワークスペースをデプロイし、そのバックエンド通信や Azure データサービスへの通信を閉域化する方法を紹介します。また、その環境を使ったデータ収集の一例として、Azure Event Hubs を使ったプライベートなデータストリーミングを試します。

はじめに

こんにちは、C&A部の吉仲です。初期配属からメール系システムや文書要約 API の開発・運用業務を担当しており、現在は主にシステムログ分析のためのデータ基盤の企画～開発業務に取り組んでいます。

昨年のアドベントカレンダーでの投稿記事では、Azure Databricks を使ったログ分析を試しました。今年はよりインフラに近い部分を扱います。具体的には、Azure Databricks を中心としたプライベートなデータ基盤・データストリーミングを、Microsoft 純正の IaC 言語であるBicep を使って構築する方法を紹介します。そして、Azure Event Hubs からデータを取り込み、Azure Data Lake Storage Gen2 (ADLS2) へ保存するまでの一連のフローを、パブリックネットワークを経由しないセキュアな経路で実現する実装例をコードと共に解説します。

なぜこの構成を記事にするのか

エンタープライズ環境でのデータ基盤において、「セキュリティ」は避けて通れない要件です。特に Azure Databricks を採用する場合、VNet へのデプロイ (VNet 統合) に加えて、ストレージやイベントソースへのアクセスもパブリックネットワークを経由させずに閉域化したいという要望は一般的だと思います。

しかし、実際にこれを構築しようとすると、次のような壁に当たりませんか？ (少なくとも私は苦戦しました)

公式ドキュメントはリファレンスアーキテクチャを提示しているが、具体的な設定値が分散していて全体像を把握するのが難しい
ポータルでのポチポチ作業は解説されているが、IaC (特に Microsoft 公式言語である Bicep) での実装例が少ない
コントロールプレーン/データプレーン、サーバレス/クラスターごとにネットワーク要件が複雑で、「疎通できない」トラブルが起きがち

そこで本記事では、私が実際にプライベートなデータ基盤・データストリーミングを構築する中で苦戦した部分や得られた知見を踏まえて、具体的な構築方法をコードと共に解説したいと思います。

今回の構成と前提

Azure Databricks のバックエンド通信や Azure Databricks からデータサービスへの通信の閉域化については、以下の公式ブログなどでリファレンスアーキテクチャが紹介されています。

www.databricks.com

今回はこの構成に倣い、以下のような環境を構築します。

ハブ&スポーク構成、インターネット向き通信の Firewall 強制トンネリング
Databricks の VNet 統合とSecure Cluster Connectivity 設定 (パブリック IP 無効化)
コントロールプレーン (バックエンド) との Private Link
サーバレスプレーン/データプレーンそれぞれに対する Azure データサービスの Private Link

なお、ハブ VNet (リファレンスアーキテクチャの図で言う "Customer transit VNet") 上に作る Gateway については、対向拠点依存の部分が多いため本記事のスコープ外とします。実際のユースケースでは、VPN Gateway や ExpressRoute Gateway をハブ VNet に構築して、オンプレミス環境等とのプライベート接続を実現します。(【参考】Azure Databricks ワークスペースをオンプレミスネットワークに接続する)

また、上記の構成では Databricks の Web UI へのアクセスまでは閉域化できません。Web UI まで閉域化する「フロントエンドの Private Link」を実現するには、リファレンスアーキテクチャに記載の通り、より複雑な構成になります。本記事では、構成を簡単にするためフロントエンドの閉域化を対象外とし、バックエンドの閉域化だけにフォーカスします。

※本構成は Azure Firewall や多数の Private Link を使用するため、検証環境であっても一定のコスト (時間課金) が発生します。検証後は速やかにリソースを削除することを推奨します。

前提

構築の要件は以下の通りです。

Azure で「グローバル管理者」とサブスクリプションの「所有者」権限を持つアカウント(※「グローバル管理者」は Databricks アカウントコンソールへのログインに必要)
Azure CLI (>=2.81.0) およびDatabricks CLI (>=0.259.0) の実行環境

以降は、基本的には以下の公式ドキュメントに倣った内容・設定で構築を進めます。

learn.microsoft.com

Bicep +α による構築

ここからはハンズオン形式で Bicep と各種 CLI ツールを使って必要なリソースを順に構築していきます。

ネットワーク基盤

ハブ&スポーク構成の VNet と、各 VNet 内のリソースを定義していきます。

VNet はハブ/スポーク用にそれぞれ作成するためモジュール化します。サブネットについては、VNet 用モジュールのsubnets パラメータでまとめて定義する形にしています。

modules/vnet.bicep

param name stringparam location stringparam tags object = {}param addressPrefixes array@description('''e.g. [{name:'default',properties:{addressPrefix:'10.0.0.0/24',networkSecurityGroup:null}}]''')param subnets array = []resource vnet 'Microsoft.Network/virtualNetworks@2024-05-01' = {  name: name  location: location  tags: tags  properties: {    addressSpace: {      addressPrefixes: addressPrefixes    }    encryption: {      enabled: true      enforcement: 'AllowUnencrypted'    }  }  @batchSize(1)  resource snet 'subnets' = [    for snet in subnets: if (!empty(subnets)) {      name: snet.name      properties: snet.properties    }  ]}output id string = vnet.idoutput name string = vnet.name

Databricks を VNet 統合する場合、ネットワークセキュリティグループ (NSG) のルールが自動で設定されますが、これらも事前に Bicep で定義しておきます。今回は、パブリック IP の無効化 + コントロールプレーンとの Private Link 構成のため、以下のような定義になります (= "No Azure Databricks Rules" 設定)。なお、この NSG ルールの変更や削除は非推奨です。

modules/nsgDbw.bicep

param name stringparam location stringparam tags object = {}resource nsg 'Microsoft.Network/networkSecurityGroups@2024-05-01' = {  name: name  location: location  tags: tags  properties: {    securityRules: [      {        name: 'Microsoft.Databricks-workspaces_UseOnly_databricks-worker-to-worker-inbound'        properties: {          description: 'Required for worker nodes communication within a cluster.'          protocol: '*'          sourcePortRange: '*'          destinationPortRange: '*'          sourceAddressPrefix: 'VirtualNetwork'          destinationAddressPrefix: 'VirtualNetwork'          access: 'Allow'          priority: 100          direction: 'Inbound'        }      }      {        name: 'Microsoft.Databricks-workspaces_UseOnly_databricks-worker-to-worker-outbound'        properties: {          description: 'Required for worker nodes communication within a cluster.'          protocol: '*'          sourcePortRange: '*'          destinationPortRange: '*'          sourceAddressPrefix: 'VirtualNetwork'          destinationAddressPrefix: 'VirtualNetwork'          access: 'Allow'          priority: 100          direction: 'Outbound'        }      }      {        name: 'Microsoft.Databricks-workspaces_UseOnly_databricks-worker-to-sql'        properties: {          description: 'Required for workers communication with Azure SQL services.'          protocol: 'tcp'          sourcePortRange: '*'          destinationPortRange: '3306'          sourceAddressPrefix: 'VirtualNetwork'          destinationAddressPrefix: 'Sql'          access: 'Allow'          priority: 101          direction: 'Outbound'        }      }      {        name: 'Microsoft.Databricks-workspaces_UseOnly_databricks-worker-to-storage'        properties: {          description: 'Required for workers communication with Azure Storage services.'          protocol: 'tcp'          sourcePortRange: '*'          destinationPortRange: '443'          sourceAddressPrefix: 'VirtualNetwork'          destinationAddressPrefix: 'Storage'          access: 'Allow'          priority: 102          direction: 'Outbound'        }      }      {        name: 'Microsoft.Databricks-workspaces_UseOnly_databricks-worker-to-eventhub'        properties: {          description: 'Required for worker communication with Azure Eventhub services.'          protocol: 'tcp'          sourcePortRange: '*'          destinationPortRange: '9093'          sourceAddressPrefix: 'VirtualNetwork'          destinationAddressPrefix: 'EventHub'          access: 'Allow'          priority: 103          direction: 'Outbound'        }      }    ]  }}output id string = nsg.idoutput name string = nsg.name

その他、Firewall やそれに割り当てるパブリック IP 、スポーク VNet から Firewall へ強制トンネリングするためのユーザー定義ルート (UDR)、ハブとスポークの VNet ピアリングもモジュール化します。

modules/afw.bicep

param name stringparam policyName stringparam location stringparam zones array = []param tags object = {}param tier string = 'Basic'param vnetName stringparam afwPipId stringparam afwManagementPipId stringresource afwp 'Microsoft.Network/firewallPolicies@2024-05-01' = {  name: policyName  location: location  tags: tags  properties: {    sku: {      tier: tier    }    threatIntelMode: 'Alert'  }  resource rcg 'ruleCollectionGroups' = {    name: 'default'    properties: {      priority: 100      ruleCollections: [        {          name: 'allow-rules'          ruleCollectionType: 'FirewallPolicyFilterRuleCollection'          action: {            type: 'Allow'          }          priority: 1000          rules: [            {              name: 'Allow-InternetOutBound'              ruleType: 'NetworkRule'              ipProtocols: ['Any']              sourceAddresses: ['10.0.0.0/8']              destinationAddresses: ['*'] // 検証のため全許可. 本来は厳密に許可する通信先だけを列挙すべき.              destinationPorts: ['*']            }          ]        }      ]    }  }}resource afw 'Microsoft.Network/azureFirewalls@2024-05-01' = {  name: name  location: location  zones: zones  tags: tags  properties: {    sku: {      name: 'AZFW_VNet'      tier: tier    }    firewallPolicy: {      id: afwp.id    }    ipConfigurations: [      {        name: 'afwIPConf'        properties: {          subnet: {            id: resourceId('Microsoft.Network/virtualNetworks/subnets', vnetName, 'AzureFirewallSubnet')          }          publicIPAddress: {            id: afwPipId          }        }      }    ]    managementIpConfiguration: {      name: 'afwManagementIPConf'      properties: {        subnet: {          id: resourceId('Microsoft.Network/virtualNetworks/subnets', vnetName, 'AzureFirewallManagementSubnet')        }        publicIPAddress: {          id: afwManagementPipId        }      }    }    threatIntelMode: 'Alert'  }}output id string = afw.idoutput name string = afw.nameoutput ipAddress string = afw.properties.ipConfigurations[0].properties.privateIPAddress

modules/pip.bicep

param name stringparam location stringparam zones array = []param tags object = {}param sku string = 'Standard'param tier string = 'Regional'resource pip 'Microsoft.Network/publicIPAddresses@2024-05-01' = {  name: name  location: location  zones: zones  tags: tags  sku: {    name: sku    tier: tier  }  properties: {    publicIPAddressVersion: 'IPv4'    publicIPAllocationMethod: 'Static'  }}output id string = pip.idoutput name string = pip.nameoutput ipAddress string = pip.properties.ipAddress

modules/rt.bicep

param name stringparam udrName stringparam location stringparam tags object = {}param afwIpAddress stringresource rt 'Microsoft.Network/routeTables@2024-05-01' = {  name: name  location: location  tags: tags  resource udr 'routes' = {    name: udrName    properties: {      addressPrefix: '0.0.0.0/0'      nextHopType: 'VirtualAppliance'      nextHopIpAddress: afwIpAddress    }  }}output id string = rt.idoutput name string = rt.name

modules/peer.bicep

param vnetHubName stringparam vnetSpokeName stringresource vnetHub 'Microsoft.Network/virtualNetworks@2024-05-01' existing = {  name: vnetHubName  resource peer 'virtualNetworkPeerings' = {    name: 'peer-hub-to-spoke'    properties: {      allowVirtualNetworkAccess: true      allowForwardedTraffic: true      allowGatewayTransit: true      useRemoteGateways: false      remoteVirtualNetwork: {        id: resourceId('Microsoft.Network/virtualNetworks', vnetSpokeName)      }    }  }}resource vnetSpoke 'Microsoft.Network/virtualNetworks@2024-05-01' existing = {  name: vnetSpokeName  resource peer 'virtualNetworkPeerings' = {    name: 'peer-spoke-to-hub'    properties: {      allowVirtualNetworkAccess: true      allowForwardedTraffic: true      allowGatewayTransit: false      useRemoteGateways: false // VPN/ExpressRoute Gateway を作成する場合は true      remoteVirtualNetwork: {        id: resourceId('Microsoft.Network/virtualNetworks', vnetHubName)      }    }  }}

ここまでの各モジュールを組み合わせて、main.bicep で各リソースを定義していきます。まずはハブ VNet の定義からです。

targetScope = 'resourceGroup'param location stringparam tags object = {}param vnetHubName stringparam pipAfwName stringparam pipAfwManagementName stringparam afwName stringparam afwpName stringmodule vnetHub './modules/vnet.bicep' = {  params: {    name: vnetHubName    location: location    tags: tags    addressPrefixes: ['10.1.0.0/16']    subnets: [      {        name: 'GatewaySubnet' // 名前固定. VPN/ExpressRoute Gateway 用        properties: {          addressPrefix: '10.1.1.0/26'        }      }      {        name: 'AzureFirewallSubnet' // 名前固定        properties: {          addressPrefix: '10.1.1.64/26'        }      }      {        name: 'AzureFirewallManagementSubnet' // 名前固定        properties: {          addressPrefix: '10.1.1.128/26'        }      }    ]  }}module pipAfw './modules/pip.bicep' = {  params: {    name: pipAfwName    location: location    tags: tags  }}module pipAfwManagement './modules/pip.bicep' = {  params: {    name: pipAfwManagementName    location: location    tags: tags  }}module afw './modules/afw.bicep' = {  params: {    name: afwName    policyName: afwpName    location: location    tags: tags    vnetName: vnetHub.outputs.name    afwPipId: pipAfw.outputs.id    afwManagementPipId: pipAfwManagement.outputs.id  }}

【説明】

ハブ VNet には Firewall と Gateway をデプロイするための3つのサブネットを作成
VNet 内に、インターネット向きアウトバウンド通信を担う (SNAT や監視) Firewall を作成

次にスポーク VNet、つまり Databricks データプレーンや各種プライベートエンドポイントを配置する VNet を定義します。

param vnetSpokeName stringparam nsgPepName stringparam nsgDbwName stringparam rtName stringparam udrName stringmodule vnetSpoke './modules/vnet.bicep' = {  params: {    name: vnetSpokeName    location: location    tags: tags    addressPrefixes: ['10.2.0.0/16']    subnets: [      {        name: 'snet-host'        properties: {          addressPrefix: '10.2.1.0/24'          networkSecurityGroup: {            id: nsgDbw.outputs.id          }          routeTable: {            id: rt.outputs.id          }          delegations: delegations // Databricks への委任        }      }      {        name: 'snet-container'        properties: {          addressPrefix: '10.2.2.0/24'          networkSecurityGroup: {            id: nsgDbw.outputs.id          }          routeTable: {            id: rt.outputs.id          }          delegations: delegations // Databricks への委任        }      }      {        name: 'snet-pep'        properties: {          addressPrefix: '10.2.3.0/27'          networkSecurityGroup: {            id: nsgPep.id          }          routeTable: {            id: rt.outputs.id          }        }      }    ]  }}resource nsgPep 'Microsoft.Network/networkSecurityGroups@2024-05-01' = {  name: nsgPepName  location: location  tags: tags}module nsgDbw './modules/nsgDbw.bicep' = {  params: {    name: nsgDbwName    location: location    tags: tags  }}module rt './modules/rt.bicep' = {  params: {    name: rtName    udrName: udrName    location: location    tags: tags    afwIpAddress: afw.outputs.ipAddress // 強制トンネリング先の Firewall の IP アドレス  }}var delegations = [  {    name: 'delegation-dbw'    properties: {      serviceName: 'Microsoft.Databricks/workspaces'    }  }]

【説明】

スポーク VNet には Databricks クラスターのホスト/コンテナ用、各種プライベートエンドポイント用の3つのサブネットを作成
クラスター (ホスト/コンテナ) 用サブネットには、"No Azure Databricks Rules" 設定の NSG を割り当て、Microsoft.Databricks/workspaces への委任を設定
各サブネットでは、UDR によりハブ VNet 上の Firewall へ強制トンネリング

最後にハブ VNet とスポーク VNet 間のピアリングを定義します。

module peer './modules/peer.bicep' = {  params: {    vnetHubName: vnetHub.outputs.name    vnetSpokeName: vnetSpoke.outputs.name  }}

データサービスとその Private Link

Databricks から接続する Azure データサービスの Private Link を定義していきます。

Databricks の外部ロケーションとして使う ADLS2 の定義と、データストリーミング用の Event Hubs の定義をそれぞれモジュール化します。(なお、以降も同様ですが、SKU やスペックは必要最低限のものに固定しています)

modules/dls.bicep

param name stringparam containerNames array = []param location stringparam tags object = {}param sku string = 'Standard_LRS'param accessTier string = 'Hot'param resourceAccessRules array = []resource dls 'Microsoft.Storage/storageAccounts@2025-01-01' = {  name: name  location: location  tags: tags  sku: {    name: sku  }  kind: 'StorageV2'  properties: {    accessTier: accessTier    allowBlobPublicAccess: false    allowedCopyScope: 'PrivateLink'    encryption: {      keySource: 'Microsoft.Storage'      services: {        blob: {          enabled: true        }      }    }    isHnsEnabled: true // 必須 (Data Lake Storage Gen2化)    largeFileSharesState: 'Disabled'    minimumTlsVersion: 'TLS1_2'    networkAcls: {      defaultAction: 'Deny' // ファイアウォールを有効化 (デフォルトで拒否)      resourceAccessRules: resourceAccessRules      bypass: 'Logging, Metrics'    }    publicNetworkAccess: 'Enabled' // Databricks アクセスコネクタからのアクセスを許可するため ('Disabled'だと全遮断)    supportsHttpsTrafficOnly: true  }  resource blob 'blobServices' = {    name: 'default'    properties: {}    resource container 'containers' = [      for containerName in containerNames: {        name: containerName        properties: {          publicAccess: 'None'        }      }    ]  }}output id string = dls.idoutput name string = dls.name

modules/evh.bicep

param name stringparam instanceName stringparam location stringparam tags object = {}param sku string = 'Standard'param capacity int = 1param isAutoInflateEnabled bool = falseparam maximumThroughputUnits int = 0param partitionCount int = 1resource evhns 'Microsoft.EventHub/namespaces@2024-01-01' = {  name: name  location: location  tags: tags  sku: {    name: sku    tier: sku    capacity: capacity  }  properties: {    isAutoInflateEnabled: sku == 'Standard' ? isAutoInflateEnabled : null    maximumThroughputUnits: sku == 'Standard' ? maximumThroughputUnits : null    publicNetworkAccess: 'Disabled'    minimumTlsVersion: '1.2'    kafkaEnabled: true  }  resource evh 'eventhubs' = {    name: instanceName    properties: {      partitionCount: partitionCount      retentionDescription: {        cleanupPolicy: 'Delete'      }      messageRetentionInDays: 1    }    // データストリーミングの収集元で使うため SAS キーを事前に用意    resource sas 'authorizationRules' = {      name: '${instanceName}Send'      properties: {        rights: ['Send']      }    }  }}output id string = evhns.idoutput name string = evhns.name

最後に、Private Link を作成するためのプライベート DNS ゾーンとその VNet 接続、プライベートエンドポイントを定義するモジュールを作ります。

modules/pep.bicep

param name stringparam zoneName stringparam location stringparam tags object = {}param vnetName stringparam snetName stringparam privateLinkServiceId stringparam groupIds arrayresource zone 'Microsoft.Network/privateDnsZones@2024-06-01' = {  name: zoneName  location: 'global'  tags: tags  resource vnetLink 'virtualNetworkLinks' = {    name: 'pl-${vnetName}'    location: 'global'    properties: {      virtualNetwork: {        id: resourceId('Microsoft.Network/virtualNetworks', vnetName)      }    }  }}resource pep 'Microsoft.Network/privateEndpoints@2024-05-01' = {  name: name  location: location  tags: tags  properties: {    subnet: {      id: resourceId('Microsoft.Network/virtualNetworks/subnets', vnetName, snetName)    }    privateLinkServiceConnections: [      {        name: name        properties: {          privateLinkServiceId: privateLinkServiceId          groupIds: groupIds        }      }    ]  }  resource dnsZoneGroup 'privateDnsZoneGroups@2024-05-01' = {    name: 'default'    properties: {      privateDnsZoneConfigs: [        {          name: 'default'          properties: {            privateDnsZoneId: zone.id          }        }      ]    }  }}output id string = pep.idoutput name string = pep.name

ここまでの各モジュールを組み合わせて、main.bicep に各リソースの定義を追記します。なお、順番が前後してしまいますが、後述の Databricks アクセスコネクタをストレージアカウントのファイアウォールで許可しています。

param dlsName stringparam evhName stringmodule dls './modules/dls.bicep' = {  params: {    name: dlsName    containerNames: ['lake']    location: location    tags: tags    resourceAccessRules: [      {        resourceId: dbw.outputs.acId // Databricks アクセスコネクタからのアクセスを許可        tenantId: tenant().tenantId      }    ]  }}module pepDfs './modules/pep.bicep' = {  params: {    name: 'pep-${dls.outputs.name}-dfs'    zoneName: 'privatelink.dfs.core.windows.net'    location: location    tags: tags    vnetName: vnetSpoke.outputs.name    snetName: 'snet-pep'    privateLinkServiceId: dls.outputs.id    groupIds: ['dfs']  }}module pepBlob './modules/pep.bicep' = {  params: {    name: 'pep-${dls.outputs.name}-blob'    zoneName: 'privatelink.blob.core.windows.net'    location: location    tags: tags    vnetName: vnetSpoke.outputs.name    snetName: 'snet-pep'    privateLinkServiceId: dls.outputs.id    groupIds: ['blob']  }  dependsOn: [    pepDfs  ]}module evh './modules/evh.bicep' = {  params: {    name: evhName    instanceName: 'topic1'    location: location    tags: tags  }}module pepEvh './modules/pep.bicep' = {  params: {    name: 'pep-${evh.outputs.name}'    zoneName: 'privatelink.servicebus.windows.net'    location: location    tags: tags    vnetName: vnetSpoke.outputs.name    snetName: 'snet-pep'    privateLinkServiceId: evh.outputs.id    groupIds: ['namespace']  }  dependsOn: [    pepBlob  ]}

プライベートエンドポイントの定義では、ADLS2 はdfs/blob、Event Hubs はnamespace を識別子 (groupIds) に指定します。なお、ADLS2 へのアクセスが Unity Catalog 経由のみの場合、blob エンドポイントはおそらく不要だと思います。

Azure Databricks ワークスペース

Databricks ワークスペースとコントロールプレーンへの Private Link を定義していきます。

Databricks ワークスペースと、そこから Azure データサービスへアクセスする際に使われる Databricks アクセスコネクタの定義をモジュール化します。

modules/dbw.bicep

param name stringparam connectorName stringparam location stringparam tags object = {}param sku string = 'premium'param managedRgName string = 'mrg-${name}'param vnetName stringparam snetHostName stringparam snetContainerName stringparam storageAccountSkuName string = 'Standard_LRS'resource dbac 'Microsoft.Databricks/accessConnectors@2024-05-01' = {  name: connectorName  location: location  tags: tags  identity: {    type: 'SystemAssigned'  }  properties: {}}resource dbw 'Microsoft.Databricks/workspaces@2024-05-01' = {  name: name  location: location  tags: tags  sku: {    name: sku  }  properties: {    managedResourceGroupId: subscriptionResourceId('Microsoft.Resources/resourceGroups', managedRgName)    accessConnector: {      id: dbac.id      identityType: 'SystemAssigned'    }    defaultStorageFirewall: 'Enabled'    publicNetworkAccess: 'Enabled'    parameters: {      customVirtualNetworkId: {        value: resourceId('Microsoft.Network/virtualNetworks', vnetName)      }      customPublicSubnetName: {        value: snetHostName      }      customPrivateSubnetName: {        value: snetContainerName      }      enableNoPublicIp: {        value: true      }      storageAccountSkuName: {        value: storageAccountSkuName      }    }    requiredNsgRules: 'NoAzureDatabricksRules'  }}output id string = dbw.idoutput name string = dbw.nameoutput acId string = dbac.idoutput acName string = dbac.nameoutput acPrincipalId string = dbac.identity.principalId

上記のモジュールを使ってmain.bicep に定義を追記します。

param dbwName stringparam dbacName stringmodule dbw './modules/dbw.bicep' = {  params: {    name: dbwName    connectorName: dbacName    location: location    tags: tags    vnetName: vnetSpoke.outputs.name    snetHostName: 'snet-host'    snetContainerName: 'snet-container'  }}module pepDbw './modules/pep.bicep' = {  params: {    name: 'pep-${dbw.outputs.name}'    zoneName: 'privatelink.azuredatabricks.net'    location: location    tags: tags    vnetName: vnetSpoke.outputs.name    snetName: 'snet-pep'    privateLinkServiceId: dbw.outputs.id    groupIds: ['databricks_ui_api']  }  dependsOn: [    pepEvh  ]}

データサービスに対する RBAC

Databricks から Azure データサービスへのアクセスは、前述のとおり Databricks アクセスコネクタで行います。したがって、このアクセスコネクタに対して ADLS2 や Event Hubs の権限を付与する必要があります。

ADLS2 と Event Hubs の RBAC ロールを付与するモジュールを作ります。

modules/dlsRbac.bicep

param name stringparam dlsName stringparam roleId stringparam principalId stringresource dls 'Microsoft.Storage/storageAccounts@2025-01-01' existing = {  name: dlsName}resource assign 'Microsoft.Authorization/roleAssignments@2022-04-01' = {  name: name  scope: dls  properties: {    principalId: principalId    roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', roleId)  }}output id string = assign.idoutput name string = assign.name

modules/evhRbac.bicep

param name stringparam evhName stringparam roleId stringparam principalId stringresource evhns 'Microsoft.EventHub/namespaces@2024-01-01' existing = {  name: evhName}resource assign 'Microsoft.Authorization/roleAssignments@2022-04-01' = {  name: name  scope: evhns  properties: {    principalId: principalId    roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', roleId)  }}output id string = assign.idoutput name string = assign.name

これらを使ってmain.bicep にて以下の RBAC ロールを Databricks アクセスコネクタに付与します。

ADLS2 (ストレージアカウント): "Azure Storage Blob Contributor" (データの読み書き用)
Event Hubs: "Azure Event Hubs Data Receiver" (イベントデータの読み取り用)

module dlsRbac './modules/dlsRbac.bicep' = {  params: {    name: guid(dlsName, dbacName, 'Storage Blob Data Contributor')    dlsName: dls.outputs.name    roleId: 'ba92f5b4-2d11-453d-a403-e96b0029c9fe' // Storage Blob Data Contributor    principalId: dbw.outputs.acPrincipalId  }}module evhRbac './modules/evhRbac.bicep' = {  params: {    name: guid(evhName, dbacName, 'Azure Event Hubs Data Receiver')    evhName: evh.outputs.name    roleId: 'a638d3c7-ab3a-418d-83e6-5f17a39d4fde' // Azure Event Hubs Data Receiver    principalId: dbw.outputs.acPrincipalId  }}

Bicep でデプロイ

ここまでかなり長くなってしまいましたが、Bicep でのリソース定義は以上です。ここからは実際にデプロイしていきます。

最初に Azure CLI でリソースグループを作成します。

az login# 「所有者」ロールのアカウントでログインaz group create--location japaneast--name rg-azuredatabricks-demo

次に Bicep パラメータファイルを用意してリソース名などのパラメータを指定します。今回は SKU やスペックを固定しているので、ここではほぼリソース名の指定だけになっています。

Bicep パラメータの例 (environments/demo.bicepparam)

using '../main.bicep'var project string = 'adbdemo'param location = 'japaneast'param vnetHubName = 'vnet-hub-${project}-${location}'param pipAfwName = 'pip-${project}-${location}-001'param pipAfwManagementName = 'pip-${project}-${location}-002'param afwName = 'afw-${project}-${location}'param afwpName = 'afwp-${project}-${location}'param vnetSpokeName = 'vnet-spoke-${project}-${location}'param nsgPepName = 'nsg-${project}-pep'param nsgDbwName = 'nsg-${project}-dbw'param rtName = 'rt-${project}'param udrName = 'udr-default-gateway'param dlsName = 'dls${project}001' // Globally uniqueparam evhName = 'evhns-${project}-001' // Globally uniqueparam dbwName = 'dbw-${project}-${location}'param dbacName = 'dbac-${project}'

それではmain.bicep と上記のパラメータファイルを使ってリソースをデプロイします。

# デプロイ後の推定状態を確認az deployment group what-if-g rg-azuredatabricks-demo--template-file main.bicep--parameters environments/demo.bicepparam# デプロイ実行az deployment group create-c-g rg-azuredatabricks-demo--template-file main.bicep--parameters environments/demo.bicepparam

デプロイ成功後、以下のようにリソースグループ内に各種リソースが表示されていると思います。

ワークスペースストレージの Private Link

さて、Bicep でのデプロイは完了しましたが、ここからが肝になります。

Databricks ワークスペースの作成箇所では説明を省きましたが、今回は既定のマネージドストレージ ("ワークスペースストレージ") でもファイアウォールを有効にし、パブリックアクセスを禁止する設定を入れています。

ワークスペースストレージとは、Databricks のシステムデータやDBFS ルートなどに使われるストレージです。公式ドキュメントでも説明されていますが、ワークスペースストレージのファイアウォールを有効にしているときは、その Private Link も作成する必要があります。

learn.microsoft.com

【余談】私は当初、このファイアウォールを有効にしたことを忘れたまま構築を進めてしまいました。そして、クラスターでパイプラインを作成しようとしたところで疎通不可となり、原因特定に随分と時間を費やしました。

ワークスペースストレージの Private Link の構築は Bicep だけでは完結しません。というのも、デプロイ後に作成されるマネージドリソースグループ内を見てもらうと分かるように、ストレージ名がdbstorage<ランダム文字列> になっています。これは動的に決まるリソース名であるため、前述までの Bicep コード内で参照することが難しいです。そこで、main.bicep とは別にpostprocess.bicep を用意し、ワークスペースストレージの Private Link のみを個別に定義する形とします。

targetScope = 'resourceGroup'param location stringparam tags object = {}param vnetName stringparam storageId stringvar stName string = last(split(storageId, '/'))module pepStBlob './modules/pep.bicep' = {  params: {    name: 'pep-${stName}-blob'    zoneName: 'privatelink.blob.core.windows.net'    location: location    tags: tags    vnetName: vnetName    snetName: 'snet-pep'    privateLinkServiceId: storageId    groupIds: ['blob']  }}module pepStDfs './modules/pep.bicep' = {  params: {    name: 'pep-${stName}-dfs'    zoneName: 'privatelink.dfs.core.windows.net'    location: location    tags: tags    vnetName: vnetName    snetName: 'snet-pep'    privateLinkServiceId: storageId    groupIds: ['dfs']  }}

この Bicep コードのデプロイ時に、すでにデプロイ済みのワークスペースストレージの ID をパラメータとして指定します。

# マネージドリソースグループ内の ADLS2 (ワークスペースストレージ) の ID を取得# マネージドリソースグループ名は本記事のBicepコードでは "mrg-dbw-adbdemo-japaneast"az storage account list-g<マネージドリソースグループ名>--query'[].id'-o tsv# 上記のコマンドで表示された ID をパラメータに指定してデプロイ実行az deployment group create-c-g rg-azuredatabricks-demo\--template-file bicep/postprocess.bicep\--parameterslocation=japaneast\--parametersvnetName=vnet-spoke-adbdemo-japaneast\--parametersstorageId=<ワークスペースストレージのID>

以上で、Databricks クラスターからワークスペースストレージへのプライベート通信が可能になりました。

サーバレスプレーンに対する Private Link

ここまで作成してきた Private Link は、全てデータプレーン内にあるクラスターからの通信用の接続構成です。サーバレスプレーンからの通信用には、以下の公式ドキュメントに記載されている設定:Network Connectivity Configuration (NCC) が必要です。

learn.microsoft.com

残念ながら現状はこの設定も Bicep で実施できません。Azure CLI および Databricks CLI を使って、ADLS2 と Event Hubs の Private Link を作成します。

事前準備:

# アカウントレベルのログイン (アカウント ID は下記 URL のコンソールで確認)databricks auth login--host https://accounts.azuredatabricks.net--account-id<アカウントID># ワークスペースレベルのログイン (ワークスペース URL は Azure Portal で確認)databricks auth login--host https://adb-<ワークスペース識別子>.azuredatabricks.net

NCC 作成とワークスペースへの割り当て:

# NCC の作成databricks account network-connectivity create-network-connectivity-configuration\--json'{"name":"ncc-adbdemo-japaneast","region":"japaneast"}'# 上記のコマンドで表示された NCC の ID "network_connectivity_config_id" を指定databricks account workspaces update<ワークスペースID>--network-connectivity-config-id<NCCID>

ADLS2 の Private Link 作成 (dfsエンドポイントの例):

# Private Link を作成する ADLS2 の ID を取得az storage account list-g rg-azuredatabricks-demo--query'[].id'-o tsv# プライベートエンドポイントの作成databricks account network-connectivity create-private-endpoint-rule<NCCID>\--json'{"resource_id":"<ストレージID>","group_id":"dfs"}'# Azure 側で承認保留中のプライベートエンドポイントを確認az network private-endpoint-connection list--id<ストレージID>\  | jq-r'.[]|select(.properties.privateLinkServiceConnectionState.status =="Pending").id'# 上記のコマンドで表示されたプライベートエンドポイントの ID を指定して、接続を承認az network private-endpoint-connection approve--id<プライベートエンドポイントID>

Event Hubs の Private Link 作成:

# Private Link を作成する Event Hubs の ID を取得az eventhubs namespace list-g rg-azuredatabricks-demo--query'[].id'-o tsv# プライベートエンドポイントの作成databricks account network-connectivity create-private-endpoint-rule<NCCID>\--json'{"resource_id":"<EventHubsID>","group_id":"namespace"}'# Azure 側で承認保留中のプライベートエンドポイントを確認az network private-endpoint-connection list--id<EventHubsID>\  | jq-r'.[]|select(.properties.privateLinkServiceConnectionState.status =="Pending").id'# 上記のコマンドで表示されたプライベートエンドポイントの ID を指定して、接続を承認az network private-endpoint-connection approve--id<プライベートエンドポイントID>

以上で、サーバレスプレーン向けの Azure データサービスの Private Link が作成されました。Databricks のアカウントコンソールで、以下のように各エンドポイントの接続がESTABLISHED になっていれば完了です。

なお、NCC を設定すると、ワークスペースストレージのファイアウォールにおいてサーバレスプレーンの VNet が自動的に許可されます。そのため、今回はワークスペースストレージの Private Link は省略します。Private Link でのアクセスとしたい場合は、上記と同じ手順で作成します。

Unity Catalog 外部ロケーション

Databricks から ADLS2 へのプライベート接続が可能になったので、その ADLS2 で Unity Catalog の外部ロケーションを作成してみます。

以降は、Bicep ではなく Azure CLI/Databricks CLI を使った作成になります。

# アクセスコネクタの ID を確認az databricks access-connector list-g rg-azuredatabricks-demo--query'[].id'-o tsv# 上記のアクセスコネクタの ID を指定して、資格情報を作成databricks storage-credentials create\--json'{"name":"adbdemo_storage","azure_managed_identity":{"<Databricks アクセスコネクタID>"}}'# 上記の資格情報を指定して、外部ロケーションを作成databricks external-locations create adbdemo_storage\  abfss://<コンテナ名>@<ストレージアカウント名>.dfs.core.windows.net/ adbdemo_storage

Databricks ワークスペースにログインし、[カタログエクスプローラー]>[外部ロケーション] から作成した外部ロケーションを開き、右上の [接続テスト] を実行します。全て「成功」であれば完了です。

なお、Private Link に不備がある場合は外部ロケーションの作成自体が失敗し、Databricks アクセスコネクタの権限不足の場合は接続テストで失敗すると思います。

以上で、プライベート接続のためのインフラ構築・設定は完了です。お疲れ様でした！

プライベートなデータストリーミングの実践

最後は、構築したプライベート接続環境を使ってデータストリーミングの実装例を紹介します。

構築した VNet とプライベート接続された環境にあるサーバをデータソースとして、Fluent Bit から Event Hubs へデータを送信し、Event Hubs からの受信データを Databricks のパイプラインでストレージに書き込む、という構成です。

まずは、Unity Catalog のカタログとスキーマを作成します。今回は、カタログ/スキーマ用のストレージは同じ ADLS2 コンテナ内でパスを分ける形で分離します(※実際のユースケースでは、メダリオンアーキテクチャの各レイヤーごとにコンテナもしくはストレージアカウントレベルで分離する方が良いと思います)。また、あわせて Databricks パイプラインから Event Hubs へ接続するための資格情報も作成します。

# カタログの作成databricks catalogs create adbdemo--storage-root abfss://<コンテナ名>@<ストレージアカウント名>.dfs.core.windows.net/catalog# スキーマの作成databricks schemas create bronze adbdemo--storage-root abfss://<コンテナ名>@<ストレージアカウント名>.dfs.core.windows.net/bronze# Event Hubs 接続用にサービス資格情報を作成databricks credentials create-credential--purpose SERVICE\--json'{"name":"adbdemo_service","azure_managed_identity":{"<Databricks アクセスコネクタID>"}}'

次に、Event Hubs をソースとするLakeflow (旧: Delta Live Tables) パイプラインを作成します。

pipelines/bronze_ingest_eventhubs_raw.py:

from pysparkimport pipelinesas dpfrom pyspark.sqlimport SparkSessionfrom pyspark.sql.functionsimport col, exprspark = SparkSession.builder.getOrCreate()# Event Hubs の Kafka モードでデータ受信するための設定# ここでは SAS キーではなく Databricks アクセスコネクタで認証KAFKA_OPTIONS = {"databricks.serviceCredential": spark.conf.get("streaming.dbw.serviceCredential"),"kafka.bootstrap.servers": spark.conf.get("streaming.evh.namespace"),"subscribe": spark.conf.get("streaming.evh.name"),"kafka.request.timeout.ms": spark.conf.get("streaming.kafka.requestTimeout"),"kafka.session.timeout.ms": spark.conf.get("streaming.kafka.sessionTimeout"),"maxOffsetsPerTrigger": spark.conf.get("streaming.spark.maxOffsetsPerTrigger"),"failOnDataLoss": spark.conf.get("streaming.spark.failOnDataLoss"),"startingOffsets": spark.conf.get("streaming.spark.startingOffsets"),}defparse(df):return (        df.withColumn("records", col("value").cast("string"))        .withColumn("eventhub_timestamp", expr("timestamp"))        .withColumn("ingested_timestamp", col("current_timestamp"))        .withColumn("date", expr("to_date(ingested_timestamp)"))        .withColumn("hash", expr("md5(records)"))        .withWatermark("eventhub_timestamp","10 minutes")        .dropDuplicatesWithinWatermark(["hash"])        .drop("key","value","partition","offset","timestamp","timestampType")    )@dp.table(    comment="Raw Logs aggregated from FluentBit-EventHubs",    partition_cols=["date"],    spark_conf={"pipelines.trigger.interval":"5 seconds"},    table_properties={"quality":"bronze","pipelines.reset.allowed":"false"},)defcommon_logs_raw():# テーブル名 (topic=インスタンスを区別していないので "common" にした)return spark.readStream.format("kafka").options(**KAFKA_OPTIONS).load().transform(parse)

このパイプラインの定義をDatabricks アセットバンドルとして用意します。Python コード内で参照する各種パラメータもここで定義します。

databricks.yml:

bundle:name: adbdemodatabricks_cli_version:">=0.259.0"targets:demo:workspace:host: https://<ワークスペース識別子>.azuredatabricks.netmode: production # 連続モードをオンにするためresources:pipelines:bronze_ingest_eventhubs_raw:name: bronze_ingest_eventhubs_rawcatalog: <カタログ名>schema: bronzetags:quality: Bronzecontinuous:true # 連続モードをオン (ストリーミングなので常時実行にする)channel: CURRENTedition: COREphoton:trueclusters: # 今回はサーバレスではなくクラスターで実行-label: defaultapply_policy_default_values:truenode_type_id: Standard_D4ds_v5custom_tags:quality: Bronzelibraries:-file:path: ./pipelines/bronze_ingest_eventhubs_raw.pyconfiguration:pipelines.clusterShutdown.delay: 60sstreaming.dbw.serviceCredential: <サービス資格情報名>streaming.evh.namespace: <EventHubs名>.servicebus.windows.net:9093streaming.evh.name: <EventHubsインスタンス名>streaming.kafka.requestTimeout:"60000"streaming.kafka.sessionTimeout:"30000"streaming.spark.maxOffsetsPerTrigger:"50000"streaming.spark.failOnDataLoss:"false"streaming.spark.startingOffsets: earliest

上記を使ってパイプラインをデプロイします。

databricks bundle validatedatabricks bundle deploy

デプロイ完了後しばらく待ち、グラフが表示されて「実行中...」となれば成功です。

最後に、Event Hubs 経由で Databricks にデータ収集するソースとして、Fluent Bit が動作する環境を用意します。この環境は前述の通り、VNet 内にある Event Hubs のプライベートエンドポイントの IP アドレスへ疎通できる場所に作成します。

Event Hubs へログ (/var/log/system.log) をストリーミングするコンフィグを作成します。

/etc/fluent-bit/fluent-bit.conf:

[INPUT]  Name  tail  Tag   systemlog  Path  /var/log/system.log# 収集するログ[OUTPUT]  Name  kafka# Event Hubs の Kafka エンドポイントへ送信  Match systemlog  timestamp_key     timestamp  timestamp_format  iso8601  format            json  brokers           <EventHubs名>.servicebus.windows.net:9093# Event Hubs エンドポイント  topics            <EventHubsインスタンス名>  rdkafka.security.protocol SASL_SSL  rdkafka.sasl.mechanisms   PLAIN  rdkafka.sasl.username     $ConnectionString  rdkafka.sasl.password     <EventHubsのSASポリシー接続文字列>

接続文字列はすでに Bicep で作成済みで、Event Hubs の共有アクセス (SAS) ポリシーの画面から取得できます。

なお、簡単のために Fluent Bit が動作する環境では、プライベートエンドポイントの FQDN (<EventHubs名>.servicebus.windows.net) を/etc/hosts で名前解決させます。実際にはAzure DNS Private Resolver を使うなどして、Azure 外からでもプライベート DNS ゾーンを参照できるようにするのがよいと思います。

それでは、実際に Fluent Bit が動作するサーバで、収集対象の/var/log/system.log にログを追記してみます。すると、Fluent Bit が収集したログがストリーミング処理によってテーブルに追記されました。