Share via

使用 Apache Spark 数据帧读取 Delta Sharing 共享表

  • 项目

本文提供的语法示例演示如何使用 Apache Spark 来查询通过 Delta Sharing 共享的数据。 使用 deltasharing 关键字作为数据帧操作的格式选项。

用于查询共享数据的其他选项

还可以创建使用元存储中注册的 Delta Sharing 目录中的共享表名的查询，例如以下示例中的查询：

SQL

SELECT * FROM shared_table_name

Python

spark.read.table("shared_table_name")

若要详细了解如何在 Azure Databricks 中配置 Delta Sharing 和使用共享表名来查询数据，请参阅使用 Databricks 到 Databricks 的 Delta Sharing 读取共享数据（面向收件人）。

可以使用结构化流式处理以增量方式处理共享表中的记录。 若要使用结构化流式处理，必须为表启用历史记录共享。 请参阅更改共享。 历史记录共享需要 Databricks Runtime 12.1 或更高版本。

如果共享表在源 Delta 表上启用了更改数据馈送，并在共享上启用了历史记录，则可在通过结构化流式处理或批处理操作读取 Delta 共享时使用更改数据馈送。 请参阅在 Azure Databricks 上使用 Delta Lake 更改数据馈送。

使用 Delta Sharing 格式关键字进行读取

Apache Spark 数据帧读取操作支持 deltasharing 关键字，如以下示例所示：

df = (spark.read
  .format("deltasharing")
  .load("<profile-path>#<share-name>.<schema-name>.<table-name>")
)

读取 Delta Sharing 共享表的变更数据馈送

对于启用了历史记录共享和变更数据馈送的表，可以使用 Apache Spark 数据帧读取变更数据馈送记录。 历史记录共享需要 Databricks Runtime 12.1 或更高版本。

df = (spark.read
  .format("deltasharing")
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2021-04-21 05:45:46")
  .option("endingTimestamp", "2021-05-21 12:00:00")
  .load("<profile-path>#<share-name>.<schema-name>.<table-name>")
)

使用结构化流式处理读取 Delta Sharing 共享表

对于共享历史记录的表，可以使用共享表作为结构化流式处理的源。 历史记录共享需要 Databricks Runtime 12.1 或更高版本。

streaming_df = (spark.readStream
  .format("deltasharing")
  .load("<profile-path>#<share-name>.<schema-name>.<table-name>")
)

# If CDF is enabled on the source table
streaming_cdf_df = (spark.readStream
  .format("deltasharing")
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2021-04-21 05:45:46")
  .load("<profile-path>#<share-name>.<schema-name>.<table-name>")
)