0

我正在尝试使用 Siddhi json Execution API 从文本中提取有效的 json。我已经从 wso2 商店下载了 siddhi-execution-json-1.1.1.jar并按照文档中提到的示例进行操作。但是上面的相同语法没有给出“SiddhiiQl 中的语法错误，不匹配的输入 'input' 期望 {',',GROUP,ORDER,LIMIT....}”的错误。下面是我的语法：

@info(name='query_name') 
from transact#window.length(1)
select json:group("json",true) as groupedJSONArray
input transact2;

我正在使用来自事务流的以下文本：

data: "" {
    "_id": {
        "$oid": "fr4wfwe4"
    },
    "code": "fesfsce",
    "name": "NAME1",
    "desc": "DECRIPTION",
    "transRefId": "FESFCSEFCS",
    "amount": 1000,
    "currency": "USD",
    "sender": {
        "id": "FRESGVSVDVEFE2333",
        "name": "rose",
        "phone": "123456789"
    },
    "message": "",
    "lockedBy": {},
    "activatedBy": {},
    "statusChangedAt": "",
    "linkBankTrans": null,
    "devGrp": 0,
    "requestId": "",
    "codeStatus": null,
    "codeTransRefId": null,
    "extOriginTransId": null
}
""

作为参考，我正在通过以下查询生成交易流：

@info(name = 'clean payload with replaceall')
from transactionstream1 
select str:replaceAll(payload,"\\","") as data
insert into transact;

我想在 WSO2 流处理器中提取有效的 json 数据：“”。我应该使用其他扩展名还是我执行的方式有一些错误？我需要上面的这个查询：@info(name='query_name') 从上面的 text 中获取 json。

4

1 回答 1

1

这里有几点需要注意，

  1. 您正在尝试使用 json:group() 函数，但是，来自API 文档，v1.1.1 不支持此功能，这是语法错误的原因。json:group() 是一个聚合函数，假设我想每 3 个事件将 JSON 元素组合成一个 json，在这种情况下，可以使用 group()，但不能在你的情况下使用。json:group() 仅在 siddhi-execution-json 的 2.xx 版本中可用，它与 WSO2 流处理器不兼容。它与 siddhi runner 兼容（一个完全开源的产品，专注于为 siddhi lib 本机提供云原生能力）。wso2 sp的下一代。

  2. 从您之前的问题中，您似乎已经提取了payload元素内的 json 字符串。但是，这是字符串类型，您需要将其作为 JSON 对象来使用 siddhi-execution-json 进行操作，这就是您使用clean payload with replace all查询去除反斜杠的原因。但是，您可以通过使用 json:toObject

    from transactionstream1
select json:toObject(payload) as payloadJson  
insert into transact;

    为此，我们将使用 Siddhiobject数据类型，它可以包含任何在查询之间传递或使用扩展进行操作的对象。json:toObject() 也是一种函数扩展类型，用于在一个事件中转换属性。

  3. 将字符串转换为 json 对象后，可以使用 getString()/getFloat/getBool() 函数从 json 对象中提取值

    from transact 
select json:getString(payloadJson, '$.code') as code, 
json:getString(payloadJson, '$.name') as name
insert into LogStream;

  4. 顺便说一句，这是通过 Debezium 检索到的 MongoDB 更改吗？

于 2019-12-12T04:28:02.770 回答