将 OneLake 与 Azure Databricks 集成

此方案演示如何通过 Azure Databricks 连接到 OneLake。 完成本教程后，你将能够从 Azure Databricks 工作区读取和写入 Microsoft Fabric lakehouse。

先决条件

在连接之前，必须具备：

• Fabric 工作区和 Lakehouse。
• 高级 Azure Databricks 工作区。 只有高级 Azure Databricks 工作区支持 Microsoft Entra 凭据直通，您在该方案中需要此功能。

设置 Databricks 工作区

1. 打开 Azure Databricks 工作区并选择“ 创建>群集”。

2. 若要使用 Microsoft Entra 标识向 OneLake 进行身份验证，必须在高级选项中对群集启用 Azure Data Lake Storage （ADLS） 凭据传递。

   

   注释

   还可以使用服务主体将 Databricks 连接到 OneLake。 有关使用服务主体对 Azure Databricks 进行身份验证的详细信息，请参阅 管理服务主体。

3. 使用首选参数创建群集。 有关创建 Databricks 群集的详细信息，请参阅 配置群集 - Azure Databricks。

4. 打开笔记本并将其连接到新创建的群集。

撰写您的笔记本

1. 导航到你的 Fabric lakehouse，然后将 Azure Blob 文件系统（ABFS）路径复制到你的 lakehouse。 可以在 “属性 ”窗格中找到它。

   注释

   在读取和写入 ADLS Gen2 和 OneLake 时，Azure Databricks 仅支持 Azure Blob 文件系统 （ABFS） 驱动程序： abfss://myWorkspace@onelake.dfs.fabric.microsoft.com/

2. 在 Databricks 笔记本中保存 lakehouse 的路径。 这个数据湖仓是你准备稍后写入已处理数据的位置。

   oneLakePath = 'abfss://myWorkspace@onelake.dfs.fabric.microsoft.com/myLakehouse.lakehouse/Files/'
   

3. 将数据从 Databricks 公共数据集加载到数据帧中。 还可以从 Fabric 中的其他位置读取文件，或者从已拥有的另一个 ADLS Gen2 帐户中选择文件。

   yellowTaxiDF = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("/databricks-datasets/nyctaxi/tripdata/yellow/yellow_tripdata_2019-12.csv.gz")
   

4. 筛选、转换或准备数据。 对于此方案，可以剪裁数据集，以便更快地加载、与其他数据集联接，或向下筛选到特定结果。

   filteredTaxiDF = yellowTaxiDF.where(yellowTaxiDF.fare_amount<4).where(yellowTaxiDF.passenger_count==4)
   display(filteredTaxiDF)
   

5. 使用 OneLake 路径将筛选后的数据帧写入到您的 Fabric 湖仓。

   filteredTaxiDF.write.format("csv").option("header", "true").mode("overwrite").csv(oneLakePath)
   

6. 通过读取新加载的文件测试数据是否已成功写入。

   lakehouseRead = spark.read.format('csv').option("header", "true").load(oneLakePath)
   display(lakehouseRead.limit(10))
   

这将完成设置，现在可以使用 Azure Databricks 在 Fabric 中读取和写入数据。

使用 Databricks 的无服务器计算能力连接到 OneLake

Databricks 无服务器计算 允许在不预配群集的情况下运行工作负荷。 根据 Databricks 无服务器文档，为了自动配置 Spark 在无服务器计算上的运行，Databricks 无服务器不允许配置在此处列出的受支持属性之外的Spark 属性。

如果尝试在链接到 Databricks 无服务器计算的笔记本中修改或设置不受支持的 Spark 配置，系统将返回CONFIG_NOT_AVAILABLE错误。

OneLake 支持来自 Databricks 无服务器计算的入站连接。 可以连接到 OneLake（通过 Databricks 无服务器计算），前提是你拥有适当的访问权限，并且 Databricks 无服务器计算与 OneLake 之间有网络路径。 使用 Databricks 的无服务器功能时，必须确保代码不会修改任何不受支持的 Spark 属性。

先决条件

在连接之前，必须具备：

• Fabric 工作区和 Lakehouse。
• 高级 Azure Databricks 工作区。
• 具有至少参与者角色分配的工作区服务主体。
• 可以使用数据库机密或 Azure Key Vault (AKV) 来存储和检索机密。 此示例使用 Databricks 机密。

撰写您的笔记本

1. 在 Databricks 工作区中创建笔记本并将其附加到无服务器计算。

   

2. 导入 Python 模块 - 在本示例中，你将使用三个模块：

   • msal 是Microsoft身份验证库（MSAL），旨在帮助开发人员将Microsoft标识平台身份验证集成到其应用程序中。
   • requests 模块用于使用 Python 发出 HTTP 请求。
   • delta Lake 用于使用 Python 读取和写入 Delta Lake 表。

   from msal import ConfidentialClientApplication
   import requests
   from deltalake import DeltaTable
   

3. 声明 Microsoft Entra 租户的变量，包括应用程序 ID。 使用部署 Microsoft Fabric 的租户的租户 ID。

   # Fetch from Databricks secrets.
   tenant_id = dbutils.secrets.get(scope="<replace-scope-name>",key="<replace value with key value for tenant _id>")
   client_id = dbutils.secrets.get(scope="<replace-scope-name>",key="<replace value with key value for client _id>") 
   client_secret = dbutils.secrets.get(scope="<replace-scope-name>",key="<replace value with key value for secret>")
   

4. 声明 Fabric 工作区变量。

   workspace_id = "<replace with workspace name>"
   lakehouse_id = "<replace with lakehouse name>"
   table_to_read = "<name of lakehouse table to read>"
   storage_account_name = workspace_id
   onelake_uri = f"abfss://{workspace_id}@onelake.dfs.fabric.microsoft.com/{lakehouse_id}.lakehouse/Tables/{table_to_read}"
   

5. 初始化客户端以获取令牌。

   authority = f"https://login.microsoftonline.com/{tenant_id}"
   
   app = ConfidentialClientApplication(
    client_id,
    authority=authority,
    client_credential=client_secret
    )
   
    result = app.acquire_token_for_client(scopes=["https://onelake.fabric.microsoft.com/.default"])
   
    if "access_token" in result:
      access_token = result["access_token"]
      print("Access token acquired.")
      token_val = result['access_token']
   

6. 从 OneLake 读取增量表

   dt = DeltaTable(onelake_uri, storage_options={"bearer_token": f"{token_val}", "use_fabric_endpoint": "true"})
   df = dt.to_pandas()
   print(df.head())
   

   注释

   服务主体具有参与者工作区角色分配，你可以使用它将数据写回到 OneLake。

在这一步设置完成后，您现在可以通过附加到无服务器计算上的 Databricks 笔记本从 OneLake 读取数据。

相关内容

• 将 OneLake 与 Azure HDInsight 集成