0

我有如下流，我想从中创建另一个流。我正在尝试以下命令，但出现以下错误。我错过了什么吗？

ksql> create stream down_devices_stream as select * from fakedata119 where deviceProperties['status']='false';
Failed to generate code for SqlPredicate.filterExpression: (FAKEDATA119.DEVICEPROPERTIES['status'] = 'false')schema:org.apache.kafka.connect.data.SchemaBuilder@6e18dbbfisWindowedKey:false
Caused by: Line 1, Column 180: Operator "<=" not allowed on reference operands

ksql> select * from fakedata119;
1529505497087 | null | 19 | visibility sensors | Wed Jun 20 16:38:17 CEST 2018 | {visibility=74, status=true}
1529505498087 | null | 7 | fans | Wed Jun 20 16:38:18 CEST 2018 | {temperature=44, rotationSense=1, status=false, frequency=49}
1529505499088 | null | 28 | air quality monitors | Wed Jun 20 16:38:19 CEST 2018 | {coPpm=257, status=false, Co2Ppm=134}
1529505500089 | null | 4 | fans | Wed Jun 20 16:38:20 CEST 2018 | {temperature=42, rotationSense=1, status=true, frequency=51}
1529505501089 | null | 23 | air quality monitors | Wed Jun 20 16:38:21 CEST 2018 | {coPpm=158, status=true, Co2Ppm=215}

sql> describe fakedata119;

     Field            | Type                                 
    ---------------------------------------------------------
     ROWTIME          | BIGINT           (system)            
     ROWKEY           | VARCHAR(STRING)  (system)            
     DEVICEID         | INTEGER                              
     CATEGORY         | VARCHAR(STRING)                      
     TIMESTAMP        | VARCHAR(STRING)                      
     DEVICEPROPERTIES | MAP[VARCHAR(STRING),VARCHAR(STRING)]
4

2 回答 2

0

我添加了一个测试来涵盖这个用例：

https://github.com/confluentinc/ksql/pull/1476/files

有趣的是，这会传递给我们的 master 和即将到来的 5.0 分支，但在 4.1 上失败了。

所以...看起来这是您正在使用的版本的问题，但好消息是它已在即将发布的版本中修复。另外，您现在可以使用 Robin 在上面的工作。

快乐查询！

安迪

于 2018-06-21T12:32:23.260 回答
0

在没有看到您的输入数据的情况下，我猜测它看起来像这样：

{
  "id": "a42",
  "category": "foo",
  "timestamp": "2018-06-21 10:04:57 BST",
  "deviceID": 42,
  "deviceProperties": {
    "status": "false",
    "foo": "bar"
  }
}

如果是这样，您最好使用EXTRACTJSONFIELD它来访问嵌套值并构建谓词。

CREATE STREAM test (Id VARCHAR, category VARCHAR, timeStamp VARCHAR, \
                    deviceID INTEGER, deviceProperties VARCHAR) \
WITH (KAFKA_TOPIC='test_map2', VALUE_FORMAT='JSON');


ksql> SELECT EXTRACTJSONFIELD(DEVICEPROPERTIES,'$.status') AS STATUS FROM fakeData223;
false

ksql> SELECT *  FROM fakeData223 \
      WHERE EXTRACTJSONFIELD(DEVICEPROPERTIES,'$.status')='false';
1529572405759 | null | a42 | foo | 2018-06-21 10:04:57 BST | 42 | {"status":"false","foo":"bar"}

您发现的错误我已在此处记录为要跟踪的错误：https ://github.com/confluentinc/ksql/issues/1474

于 2018-06-21T09:37:41.783 回答