使用 Apache Spark 将数据从 Azure Cosmos DB 复制到专用 SQL 池

Azure Synapse Link for Azure Cosmos DB 使用户能够对 Azure Cosmos DB 中的操作数据运行准实时分析。 但是，有时需要聚合和丰富某些数据，以便为数据仓库用户提供服务。 只需通过笔记本中的几个单元格即可策展和导出 Azure Synapse Link 数据。

先决条件

• 为 Synapse 工作区预配以下内容：
  • 无服务器 Apache Spark 池
  • 专用 SQL 池
• 为 Azure Cosmos DB 帐户预配包含数据的 HTAP 容器
• 将 Azure Cosmos DB HTAP 容器连接到工作区
• 使用正确的设置将数据从 Spark 导入专用 SQL 池

步骤

在本教程中，你将连接到分析存储，因此不会对事务存储产生任何影响（它不会消耗任何请求单位）。 我们将执行以下步骤：

1. 将 Azure Cosmos DB HTAP 容器读入 Spark 数据帧
2. 在新数据帧中聚合结果
3. 将数据引入专用 SQL 池中

数据

在该示例中，我们使用名为“RetailSales”的 HTAP 容器。 它属于名为“ConnectedData”的链接服务，具有以下架构：

• _rid: string (nullable = true)
• _ts: long (nullable = true)
• logQuantity：double（nullable = true）
• productCode：字符串（nullable = true）
• quantity: long (nullable = true)
• price: long (nullable = true)
• id：string（nullable = true）
• advertising: long (nullable = true)
• storeId：long（nullable = true）
• weekStarting: long (nullable = true)
• _etag: string (nullable = true)

为了进行报告，我们将按“productCode”和“weekStarting”聚合销售额（数量、收入 [价格 x 数量]）。 最后，我们会将该数据导入名为 dbo.productsales 的专用 SQL 池表。

配置 Spark 笔记本

创建一个 Spark 笔记本，以 Scala as Spark (Scala) 作为主要语言。 使用笔记本的默认会话设置。

读取 Spark 中的数据

在第一个单元格中，使用 Spark 将 Azure Cosmos DB HTAP 容器读取到数据帧中。

val df_olap = spark.read.format("cosmos.olap").
    option("spark.synapse.linkedService", "ConnectedData").
    option("spark.cosmos.container", "RetailSales").
    load()

在新数据帧中聚合结果

在第二个单元格中，在将新数据帧加载到专用 SQL 池数据库之前，先运行它所需的转换和聚合。

// Select relevant columns and create revenue
val df_olap_step1 = df_olap.select("productCode","weekStarting","quantity","price").withColumn("revenue",col("quantity")*col("price"))
//Aggregate revenue, quantity sold and avg. price by week and product ID
val df_olap_aggr = df_olap_step1.groupBy("productCode","weekStarting").agg(sum("quantity") as "Sum_quantity",sum("revenue") as "Sum_revenue").
    withColumn("AvgPrice",col("Sum_revenue")/col("Sum_quantity"))

将结果加载到专用 SQL 池中

在第三个单元格中，将数据加载到专用 SQL 池中。 它将自动创建一个临时外部表、外部数据源和外部文件格式，作业完成后会删除这些内容。

df_olap_aggr.write.sqlanalytics("userpool.dbo.productsales", Constants.INTERNAL)

通过 SQL 查询结果

可以使用简单的 SQL 查询（例如以下 SQL 脚本）查询结果：

SELECT  [productCode]
,[weekStarting]
,[Sum_quantity]
,[Sum_revenue]
,[AvgPrice]
 FROM [dbo].[productsales]

查询会以图表模式显示以下结果：

后续步骤

• 使用 Apache Spark 3 查询 Azure Cosmos DB 分析存储
• 使用 Apache Spark 2 查询 Azure Cosmos DB 分析存储