Azure Databricks Spark SQL Query to CosmosDB 從其他文档獲取列

Question

使用 Azure-Cosmosdb-spark 连接器将 sql 查询传递到 CosmosDB 时，它似乎从集合中的其他文档添加列。集合中有许多文档，例如用户和公司，它们由实体类型分隔。例如，用户设置为

{   "id": "user-000003",
    "email": "someemail",
    "firstName": "firstname",
    "lastName": "lastname",
    "username": "someusername",
    "companyId": "company-000003",
    "entity": "user"
}

本公司成立为：

{   "id": "company-000003",
    "contactName": "namegoes here",
    "addressLine1": "Address line 1",
    "addressLine2": "Address line 2",
    "entity": "company"
}

使用Azure-Cosmosdb火花sdk我创建了我的连接

cosmosConfig = {
                "Endpoint" : "my endpoint goes here",
                "Masterkey" : "my key goes here",
                "Database" : "my database goes here",
                "preferredRegions" : "my region goes here",
                "Collection" : "my collection", 
                "SamplingRatio" : "1.0",
                "schema_samplesize" : "1000",
                "query_pagesize" : "2147483647",
               }

然后呢

将其设置为使用该连接

cosmosdbConnection = spark.read.format("com.microsoft.azure.cosmosdb.spark").options(**cosmosConfig).load()
cosmosdbConnection.createOrReplaceTempView("c")

然后，我运行查询

exampleQuery= 'SELECT c.* FROM c WHERE c.entity = "user"'
users = spark.sql(exampleQuery)

我希望得到一个包含用户文档中定义的列、id、email、firstName、lastName、username、companyId和实体的数据帧。但是，它也从公司文档中提取列名，但都是空值。在Azure Cosmos DB存储资源管理器或Azure门户中运行相同的查询，只会返回用户文档。

我只需要指定我想要的列名，但是如果模式发生变化，我需要添加这些列。

我想是这个问题吧？我正在寻找在sql查询中从文档中获取列的方法。我确实认为它会将sql查询传递给cosmosdb SQL API。

这是我第一次在cosmos数据库中使用数据库，我在谷歌上搜索过，但似乎看不到我做错了什么。

Answer 1

问题在于，在配置设置中，没有指定对集合的SQL查询，它只是读取集合中的所有文档，而不是应按实体/主题类型拆分的不同文档。

通过在配置中添加 SQL 查询

cosmosConfig = {
                "Endpoint" : "my endpoint goes here",
                "Masterkey" : "my key goes here",
                "Database" : "my database goes here",
                "preferredRegions" : "my region goes here",
                "Collection" : "my collection", 
                "SamplingRatio" : "1.0",
                "schema_samplesize" : "1000",
                "Query" : "SELECT * FROM c WHERE C.entity = 'SomeEntity"
                "query_pagesize" : "2147483647",
               }

它将在此基础上创建连接。如果您需要添加多个实体，请创建一个函数，例如：

def createCosmosDBObject(useEntity):
  query = "SELECT * FROM c WHERE c.entity = " + "'" + useEntity + "'"
  # Create connection setting
  cosmosConfig = {
                "Endpoint" : "Kyour endpoint",
                "Masterkey" : "Your Key",
                "Database" : "Your Database",
                "preferredRegions" : "Azure Region",
                "Collection" : "Your Collection", 
                "ConnectionMode": "DirectHttps", 
                "SamplingRatio" : "1.0",
                "schema_samplesize" : "20000",
                "query_pagesize" : "2147483647",
                "query_custom" : query
               }

  createConnection = spark.read.format("com.microsoft.azure.cosmosdb.spark").options(**cosmosConfig).load()
  createConnection.createOrReplaceTempView(useEntity)

然后可以通过文档实体名称调用它：

createCosmosDBObject("customer")

然后，它将在查询中插入“customer ”,然后创建一个可以查询的临时视图，而不会与Cosmos DB中的其他文档实体类型重叠

希望那有帮助