使用 Spark 的 Cosmos DB 空间查询答案

【问题标题】：Cosmos DB spatial query using Spark使用 Spark 的 Cosmos DB 空间查询
【发布时间】：2021-06-27 12:27:09
【问题描述】：

我想使用空间查询来查询 cosmos db 集合。特别是ST_DISTANCE 查询。此查询使用azure-cosmos Python SDK 按预期工作。

我希望通过 Apache Spark 将此查询用于更复杂的查询模式。但是，在笔记本的 SQL 单元格中使用 ST_DISTANCE 查询会导致以下错误。

Error in SQL statement: AnalysisException: Undefined function: 'ST_DISTANCE'. This function is neither a registered temporary function nor a permanent function registered in the database 'default'.

笔记本初始化如下。

# Configure Catalog Api to be used
spark.conf.set("spark.sql.catalog.cosmosCatalog", "com.azure.cosmos.spark.CosmosCatalog")
spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountEndpoint", cosmosEndpoint)
spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountKey", cosmosMasterKey)

from pyspark.sql.functions import col

df = spark.read.format("cosmos.oltp").options(**cfg)\
 .option("spark.cosmos.read.inferSchema.enabled", "true")\
 .load()

df.createOrReplaceTempView("outlets")

_______________________________________________________________________

%sql
SELECT * FROM outlets f WHERE ST_DISTANCE(f.boundary, POINT(0,0)) < 600

【问题讨论】：

标签： apache-spark pyspark jupyter-notebook azure-cosmosdb azure-cosmosdb-sqlapi

【解决方案1】：

根据我从 Cosmos DB Spark 连接器 github 存储库 [1] 中了解到的情况，并非所有 Cosmos DB 过滤器查询都通过连接器支持（还没有？）。因此，空间系列中的 ST_DISTANCE 和其他过滤器函数将无法正常工作，因为它们不是 Spark 原生支持的谓词，可以下推到数据库中。

找到了一些至少可以暂时解决这个问题的方法。查询配置 [2] 允许将自定义查询直接发送到 Cosmos DB。可以构建和查询临时视图。这不适用于所有用例，但这解决了我需要完成距离过滤的单个视图的问题。休息可以通过 Spark SQL 处理。

请参阅以下示例中的spark.cosmos.read.customQuery[2]。

outlets_cfg = {
  "spark.cosmos.accountEndpoint" : cosmosEndpoint,
  "spark.cosmos.accountKey" : cosmosMasterKey,
  "spark.cosmos.database" : cosmosDatabaseName,
  "spark.cosmos.container" : cosmosContainerName,
  "spark.cosmos.read.customQuery" : "SELECT * FROM c WHERE ST_DISTANCE(c.location,{\"type\":\"Point\",\"coordinates\": [12.832489, 18.9553242]}) < 1000"
}

df = spark.read.format("cosmos.oltp").options(**outlets_cfg)\
 .option("spark.cosmos.read.inferSchema.enabled", "true")\
 .load()

df.createOrReplaceTempView("outlets")

[1]https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/cosmos/azure-cosmos-spark_3-1_2-12/

[2]https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/cosmos/azure-cosmos-spark_3-1_2-12/docs/configuration-reference.md#query-config

【讨论】：