ksqldb-stream不使用主题键生成任何结果

b09cbbtk  于 2021-07-15  发布在  Kafka
关注(0)|答案(0)|浏览(213)

用例：创建一个订单系统来处理移动设备和苹果设备的订单。
收到订单
一旦收到订单，就需要对通过和失败的结果进行3种不同的验证。
一旦所有的验证结果都执行了，我们需要汇总所有通过的订单，并且计数为3。满足上述验证条件的订单需要更新订单状态（从“已创建”到“已验证”），以便进一步处理。
主题：订单->保留所有订单的记录订单验证->保留主题订单中所有订单上发生的所有验证的记录。有3种类型的验证发生，因此每个订单在本主题中将有3条状态为“通过”或“失败”以及订单id的记录。
解决方案需要创建两个主题->
订单->接受所有订单
订单验证。->保留所有订单验证结果
注意：我在一个docker容器中使用ksqldb cli（6.0.0）。
ksql处理：
已经从两个主题s_o（来自ordres）和s_ov（订单验证）创建了2个流。

  1. -- Create Stream of Orders topic as s_o
  2. CREATE stream s_o  (id integer , quantity integer,price integer,customerId integer,OrderState varchar, product varchar )  WITH (KAFKA_TOPIC='orders', VALUE_FORMAT='JSON')
  3. -- Create Stream of Order-validations topic  s_ot
  5. CREATE STREAM s_ov  (OrderValidation STRUCT <OrderValidationResult VARCHAR, OrderValidationType VARCHAR,orderId integer> )  WITH (KAFKA_TOPIC='order-validations', VALUE_FORMAT='JSON');

将订单状态为“created”且窗口滚动时间为60秒的计数进行聚合，并存储在名为t\u s\u o的表中；

  1. -- Create a table to aggregate count based on validation resuts and group by orderId
  2. CREATE TABLE t_s_ov  AS select ORDERVALIDATION->orderid as id,  count(*) as total  from s_ov  window tumbling (size 60 seconds)   where ORDERVALIDATION->ORDERVALIDATIONRESULT ='PASS'   group by ORDERVALIDATION->orderid emit changes;

当计数为3或更多时（如果计数为3，则所有验证都通过），将来自s\u o和t\u s\u ov的订单合并，以将订单状态更新为“已验证”
因为ksql不允许持久表和非持久流之间的连接，所以我从主题“t\u s\u ov”创建了一个流（主题作为步骤2的一部分在内部创建）

  1. CREATE STREAM s_t_s_ov  (id integer KEY, total integer )  WITH (KAFKA_TOPIC='T_S_OV',VALUE_FORMAT='JSON') ;

验证步骤4的结果

  1. Details:
  3. ksql> DROP STREAM IF EXISTS s_t_s_ov;
  4. >CREATE STREAM s_t_s_ov  (id integer KEY, total integer )  WITH (KAFKA_TOPIC='T_S_OV',VALUE_FORMAT='JSON') ;
  6. DROP STREAM IF EXISTS s_t_s_ov;
  7.  Message
  8. ------------------------------------------------
  9.  Source `S_T_S_OV` (topic: T_S_OV) was dropped.
  10. ------------------------------------------------
  12. CREATE STREAM s_t_s_ov  (id integer KEY, total integer )  WITH (KAFKA_TOPIC='T_S_OV',VALUE_FORMAT='JSON') ;
  13.  Message
  14. ----------------
  15.  Stream created
  16. ----------------
  17. ksql> select * from s_t_s_ov emit changes limit 3;
  18. +--------------------------------------------------------+--------------------------------------------------------+
  19. |ID                                                      |TOTAL                                                   |
  20. +--------------------------------------------------------+--------------------------------------------------------+
  22. Press CTRL-C to interrupt

有人知道我做错了什么吗？或任何其他可能的方法来实现这一点。
订单流

  1. ksql> describe extended s_o;
  3. Name                 : S_O
  4. Type                 : STREAM
  5. Timestamp field      : Not set - using <ROWTIME>
  6. Key format           : KAFKA
  7. Value format         : JSON
  8. Kafka topic          : orders (partitions: 1, replication: 1)
  9. Statement            : CREATE stream s_o  (id integer , quantity integer,price integer,customerId integer,OrderState varchar, product varchar )  WITH (KAFKA_TOPIC='orders', VALUE_FORMAT='JSON')
  10. ;
  12.  Field      | Type
  13. ------------------------------
  14.  ID         | INTEGER
  15.  QUANTITY   | INTEGER
  16.  PRICE      | INTEGER
  17.  CUSTOMERID | INTEGER
  18.  ORDERSTATE | VARCHAR(STRING)
  19.  PRODUCT    | VARCHAR(STRING)
  20. ------------------------------
  22. Local runtime statistics
  23. ------------------------
  24. consumer-messages-per-sec:      0.27 consumer-total-bytes:   6086118 consumer-total-messages:     34341     last-message: 2021-04-26T15:08:40.681Z
  25. consumer-failed-messages:      5802 consumer-failed-messages-per-sec:         0      last-failed: 2021-04-23T04:13:57.008Z
  26. (Statistics of the local KSQL server interaction with the Kafka topic orders)

订单验证流

  1. ksql> describe extended s_ov;
  3. Name                 : S_OV
  4. Type                 : STREAM
  5. Timestamp field      : Not set - using <ROWTIME>
  6. Key format           : KAFKA
  7. Value format         : JSON
  8. Kafka topic          : order-validations (partitions: 1, replication: 1)
  9. Statement            : CREATE STREAM s_ov  (OrderValidation STRUCT <OrderValidationResult VARCHAR, OrderValidationType VARCHAR,orderId integer> )  WITH (KAFKA_TOPIC='order-validations', VALUE_FORMAT='JSON');
  11.  Field           | Type                                                                                              
  12. ---------------------------------------------------------------------------------------------------------------------
  13.  ORDERVALIDATION | STRUCT<ORDERVALIDATIONRESULT VARCHAR(STRING), ORDERVALIDATIONTYPE VARCHAR(STRING), ORDERID INTEGER>
  14. ---------------------------------------------------------------------------------------------------------------------
  16. Queries that read from this STREAM
  17. -----------------------------------
  18. CTAS_T_S_OV_1_155 (RUNNING) : CREATE TABLE T_S_OV_1 WITH (KAFKA_TOPIC='T_S_OV_1', PARTITIONS=1, REPLICAS=1) AS SELECT   S_OV.ORDERVALIDATION->ORDERID ID,   COUNT(*) TOTAL FROM S_OV S_OV WINDOW TUMBLING ( SIZE 60 SECONDS )  WHERE (S_OV.ORDERVALIDATION->ORDERVALIDATIONRESULT = 'PASS') GROUP BY S_OV.ORDERVALIDATION->ORDERID EMIT CHANGES;
  19. CTAS_T_S_OV_85 (RUNNING) : CREATE TABLE T_S_OV WITH (KAFKA_TOPIC='T_S_OV', PARTITIONS=1, REPLICAS=1) AS SELECT   S_OV.ORDERVALIDATION->ORDERID ORDERID,   COUNT(*) TOTAL FROM S_OV S_OV WINDOW TUMBLING ( SIZE 60 SECONDS )  WHERE (S_OV.ORDERVALIDATION->ORDERVALIDATIONRESULT = 'PASS') GROUP BY S_OV.ORDERVALIDATION->ORDERID EMIT CHANGES;
  21. For query topology and execution plan please run: EXPLAIN <QueryId>
  23. Local runtime statistics
  24. ------------------------
  25. consumer-messages-per-sec:      1.09 consumer-total-bytes:  10836043 consumer-total-messages:     80796     last-message: 2021-04-27T01:39:21.178Z
  26. consumer-failed-messages:       704 consumer-failed-messages-per-sec:         0      last-failed: 2021-04-26T23:58:47.916Z
  27. (Statistics of the local KSQL server interaction with the Kafka topic order-validations)

订单验证表聚合

  1. ksql> describe extended t_s_ov;
  3. Name                 : T_S_OV
  4. Type                 : TABLE
  5. Timestamp field      : Not set - using <ROWTIME>
  6. Key format           : KAFKA
  7. Value format         : JSON
  8. Kafka topic          : T_S_OV (partitions: 1, replication: 1)
  9. Statement            : CREATE TABLE T_S_OV WITH (KAFKA_TOPIC='T_S_OV', PARTITIONS=1, REPLICAS=1) AS SELECT
  10.   S_OV.ORDERVALIDATION->ORDERID ORDERID,
  11.   COUNT(*) TOTAL
  12. FROM S_OV S_OV
  13. WINDOW TUMBLING ( SIZE 60 SECONDS )
  14. WHERE (S_OV.ORDERVALIDATION->ORDERVALIDATIONRESULT = 'PASS')
  15. GROUP BY S_OV.ORDERVALIDATION->ORDERID
  16. EMIT CHANGES;
  18.  Field   | Type
  19. ------------------------------------------------------------------
  20.  ORDERID | INTEGER          (primary key) (Window type: TUMBLING)
  21.  TOTAL   | BIGINT
  22. ------------------------------------------------------------------
  24. Queries that write from this TABLE
  25. -----------------------------------
  26. CTAS_T_S_OV_85 (RUNNING) : CREATE TABLE T_S_OV WITH (KAFKA_TOPIC='T_S_OV', PARTITIONS=1, REPLICAS=1) AS SELECT   S_OV.ORDERVALIDATION->ORDERID ORDERID,   COUNT(*) TOTAL FROM S_OV S_OV WINDOW TUMBLING ( SIZE 60 SECONDS )  WHERE (S_OV.ORDERVALIDATION->ORDERVALIDATIONRESULT = 'PASS') GROUP BY S_OV.ORDERVALIDATION->ORDERID EMIT CHANGES;
  28. For query topology and execution plan please run: EXPLAIN <QueryId>
  30. Local runtime statistics
  31. ------------------------
  32. messages-per-sec:      0.09   total-messages:      7730     last-message: 2021-04-27T01:40:16.45Z
  34. (Statistics of the local KSQL server interaction with the Kafka topic T_S_OV)

来自订单验证表的流

  1. ksql> describe extended s_t_s_ov;
  3. Name                 : S_T_S_OV
  4. Type                 : STREAM
  5. Timestamp field      : Not set - using <ROWTIME>
  6. Key format           : KAFKA
  7. Value format         : JSON
  8. Kafka topic          : T_S_OV (partitions: 1, replication: 1)
  9. Statement            : CREATE STREAM s_t_s_ov  (id integer KEY, total integer )  WITH (KAFKA_TOPIC='T_S_OV',VALUE_FORMAT='JSON') ;
  11.  Field | Type
  12. --------------------------------
  13.  ID    | INTEGER          (key)
  14.  TOTAL | INTEGER
  15. --------------------------------
  17. Local runtime statistics
  18. ------------------------
  19. messages-per-sec:      0.09   total-messages:      7737     last-message: 2021-04-27T01:41:31.567Z
  21. (Statistics of the local KSQL server interaction with the Kafka topic T_S_OV)

注意：消息总数：7737，但是它不显示任何结果。
我尝试了第3步的另一个版本，即orders id值为空的results。

  1. ksql> CREATE STREAM s_t_s_ov1  (id integer, total integer )  WITH (KAFKA_TOPIC='T_S_OV',VALUE_FORMAT='JSON') ;
  2. >
  4.  Message
  5. ----------------
  6.  Stream created
  7. ----------------
  8. ksql> select * from s_t_s_ov1 emit changes limit 3;
  9. +--------------------------------------------------------+--------------------------------------------------------+
  10. |ID                                                      |TOTAL                                                   |
  11. +--------------------------------------------------------+--------------------------------------------------------+
  12. |null                                                    |1                                                       |
  13. |null                                                    |1                                                       |
  14. |null                                                    |1                                                       |
  15. Limit Reached
  16. Query terminated

注意，我已经从语句中删除了key。

apache-kafkaksqldb

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com