无法在ksql中创建流

hk8txs48 于 2021-06-08 发布在 Kafka

关注(0)|答案(2)|浏览(363)

我有如下流，我想从这个创建另一个流。我尝试下面的命令，我得到以下错误。我错过什么了吗？

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)]

apache-kafka ksqldb

来源：https://stackoverflow.com/questions/50951239/cannot-create-stream-in-ksql

2条答案

按热度按时间

af7jpaap1#

我添加了一个测试来覆盖这个用例：
https://github.com/confluentinc/ksql/pull/1476/files
有趣的是，这将传递给我们的主分支和即将到来的5.0分支，但在4.1中失败。
所以。。。看起来这是您正在使用的版本中的一个问题，但好消息是它在即将发布的版本中得到了修复。另外，你现在可以利用罗宾的工作。
快乐查询！
安迪

赞(0）回复(0）举报 2021-06-08

eeq64g8w2#

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

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

如果是这样，你最好用 EXTRACTJSONFIELD 访问嵌套值，并生成 predicate 。

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

赞(0）回复(0）举报 2021-06-08

我来回答

无法在ksql中创建流

2条答案

相关问题

热门标签

最新问答