1

我有一个 Azure 流分析 (ASA) 作业，它处理来自事件中心的设备遥测数据。流应该与来自 sql 表的参考数据连接，以使用额外的设备元数据增强每条消息。合并的条目应存储在 CosmosDb 中。

为设备元数据提供服务的 sql 数据库：

CREATE TABLE [dbo].[MyTable]
(
  [DeviceId] NVARCHAR(20) NOT NULL PRIMARY KEY, 
  [MetaData] NVARCHAR(MAX) NULL   /* this stores json, which can vary per record */
)

在 ASA 中，我使用一个简单的查询配置了参考数据输入：

SELECT DeviceId, JSON_QUERY(MetaData) FROM [dbo].[MyTable]

我有执行连接的主要 ASA 查询：

WITH temptable AS (
SELECT * FROM [telemetry-input] TD PARTITION BY PartitionId
LEFT OUTER JOIN [metadata-input] MD
ON TD.DeviceId = MD.DeviceId
)

SELECT TD.*, MD.MetaData 
INTO [cosmos-db-output] 
FROM temptable PARTITION BY PartitionId

这一切正常，合并的数据存储在 CosmosDb 中。但是，来自 sql 的 Metadata 列的值被视为字符串，并使用引号和转义字符存储在 comos 中。例子： 

{ "DeviceId" : "abc1234", … , "MetaData" : "{ \"TestKey\": \"test value\" }" };

有没有办法将元数据中的 json 存储为适当的 Json 对象，即

{ "DeviceId" : "abc1234", … , "MetaData" : { "TestKey": "test value" } };
4

2 回答 2

2

我找到了在 ASA 中实现它的方法 - 你需要创建javascript 用户函数：

function parseJson(strjson){
          return JSON.parse(strjson);
}

并在您的查询中调用它：

...
SELECT TD.*, udf.parseJson(MD.MetaData)
...
于 2020-02-15T10:47:38.077 回答
1

正如您在问题中提到的，参考 json 数据被视为 json 字符串，而不是 json 对象。根据我对 ASA 中查询语法的研究，没有内置函数可以转换它。

但是，我建议您使用Azure Function Cosmos DB 触发器来处理创建的每个文档。请参考我的功能代码：

using System;
using System.Collections.Generic;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Host;
using Newtonsoft.Json.Linq;

namespace ProcessJson
{
    public class Class1
    {
        [FunctionName("DocumentUpdates")]
        public static void Run(
        [CosmosDBTrigger(databaseName:"db",collectionName: "item", ConnectionStringSetting = "CosmosDBConnection",LeaseCollectionName = "leases",
            CreateLeaseCollectionIfNotExists = true)]
        IReadOnlyList<Document> documents,
        TraceWriter log)
        {
            log.Verbose("Start.........");
            String endpointUrl = "https://***.documents.azure.com:443/";
            String authorizationKey = "***";
            String databaseId = "db";
            String collectionId = "import";

            DocumentClient client = new DocumentClient(new Uri(endpointUrl), authorizationKey);

            for (int i = 0; i < documents.Count; i++)
            {
                Document doc = documents[i];
                if((doc.alreadyFormat == Undefined.Value) ||(!doc.alreadyFormat)){
                   String MetaData = doc.GetPropertyValue<String>("MetaData");
                   JObject o = JObject.Parse(MetaData);

                   doc.SetPropertyValue("MetaData", o);
                   doc.SetPropertyValue("alreadyFormat", true);
                   client.ReplaceDocumentAsync(UriFactory.CreateDocumentUri(databaseId, collectionId, doc.Id), doc); 

                   log.Verbose("Update document Id " + doc.Id);

                }

            }
        }
    }
}

另外请参考案例：Azure Cosmos DB SQL - how to unescape inner json property

于 2020-02-14T02:32:15.330 回答