【问题标题】:KSQL - INSERT INTO a Stream yields no dataKSQL - INSERT INTO a Stream 不产生数据
【发布时间】:2018-11-24 05:00:14
【问题描述】:

我在从使用INSERT INTO KSQL 操作填充的流中读取消息时遇到问题。

我遵循的步骤是:

我有一个流 event_stream,它是从一个 kafka 主题创建的。

CREATE STREAM event_stream (eventType varchar, eventTime varchar, 
sourceHostName varchar) WITH (kafka_topic='events', value_format='json');

SELECT * FROM event_stream; 显示消息正确进入。

我想将其中一些消息发送到我已经创建的 kafka 中的另一个主题 output_events

然后我在 KSQL 中创建第二个流:

CREATE STREAM output_stream (eventTime varchar, extraColumn varchar, 
sourceHostName varchar) WITH (kafka_topic='output_events', value_format='json');

最后,我将输入链接到输出:

INSERT INTO output_stream SELECT eventTime, 'Extra info' as extraColumn,    
sourceHostName FROM event_stream WHERE eventType = 'MatchingValue';

以上所有似乎都没有错误地完成,但是如果我运行 SELECT * FROM output_stream;我没有得到任何数据。这是为什么呢?

运行上述查询的 SELECT 部分工作正常,所以我可以看到匹配的结果到达主题。

奇怪的是,如果我运行 DESCRIBE EXTENDED output_stream,消息计数确实表明消息正在到达流:

Local runtime statistics                                                                        
------------------------                                                                        
messages-per-sec:      0.33   total-messages:        86     last-message: 11/9/18 1:15:43 PM UTC
 failed-messages:         0 failed-messages-per-sec:         0      last-failed:       n/a      
(Statistics of the local KSQL server interaction with the Kafka topic output_events) 

我还检查了 ksql-server 日志,但没有看到任何错误。

【问题讨论】:

  • 您使用的是什么版本的 Confluent 平台?
  • 这是版本 5.0.0
  • 当你看题目时会发生什么:PRINT 'output_events' from beginning;
  • 从头开始打印 output_events 没有结果,设置 'auto.offset.reset'='earliest';然后重新运行 SELECT 语句。

标签: apache-kafka ksqldb


【解决方案1】:

这是a bug,由于无意中以错误的语法误用了CREATE STREAM。您正在使用该变体针对现有主题“注册”一个 KSQL 流。要使INSERT INTO 工作,它必须是CREATE STREAM target AS SELECT(“CSAS”)。

让我们解决它。在这里,我使用this docker-compose 进行测试设置。

填充一些虚拟数据:

docker run --rm --interactive --network cos_default confluentinc/cp-kafkacat kafkacat -b kafka:29092 -t events -P <<EOF
{"eventType":"1", "eventTime" :"2018-11-13-06:34:57", "sourceHostName":"asgard"}
{"eventType":"2", "eventTime" :"2018-11-13-06:35:57", "sourceHostName":"asgard"}
{"eventType":"MatchingValue", "eventTime" :"2018-11-13-06:35:58", "sourceHostName":"asgard"}
EOF

用KSQL注册源主题:

CREATE STREAM event_stream (eventType varchar, eventTime varchar, sourceHostName varchar) WITH (kafka_topic='events', value_format='json');

查询流:

ksql> SET 'auto.offset.reset' = 'earliest';
Successfully changed local property 'auto.offset.reset' from 'null' to 'earliest'
ksql> SELECT * FROM event_stream;
1542091084660 | null | 1 | 2018-11-13-06:34:57 | asgard
1542091084660 | null | 2 | 2018-11-13-06:35:57 | asgard
1542091785207 | null | MatchingValue | 2018-11-13-06:35:58 | asgard

所以看看你引用的CREATE STREAM

CREATE STREAM output_stream (eventTime varchar, extraColumn varchar, sourceHostName varchar) WITH (kafka_topic='output_events', value_format='json');

我的猜测是,如果您运行LIST TOPICS;,您会发现该主题已经存在于您的 Kafka 代理上?

ksql> LIST TOPICS;

Kafka Topic            | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups
----------------------------------------------------------------------------------------------------
_confluent-metrics     | false      | 12         | 1                  | 0         | 0
_schemas               | false      | 1          | 1                  | 0         | 0
docker-connect-configs | false      | 1          | 1                  | 0         | 0
docker-connect-offsets | false      | 25         | 1                  | 0         | 0
docker-connect-status  | false      | 5          | 1                  | 0         | 0
events                 | true       | 1          | 1                  | 0         | 0
output_events          | false      | 4          | 1                  | 0         | 0
----------------------------------------------------------------------------------------------------
ksql>

因为如果不是这样,这个CREATE STREAM 就会失败:

ksql> CREATE STREAM output_stream (eventTime varchar, extraColumn varchar, sourceHostName varchar) WITH (kafka_topic='output_events', value_format='json');
Kafka topic does not exist: output_events
ksql>

所以做这个假设我也在我的测试集群上创建这个主题:

$ docker-compose exec kafka bash -c "kafka-topics --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 4 --topic output_events"

然后创建流:

ksql> CREATE STREAM output_stream (eventTime varchar, extraColumn varchar, sourceHostName varchar) WITH (kafka_topic='output_events', value_format='json');

Message
----------------
Stream created
----------------

注意上面写的是Stream created,而不是Stream created and running

现在让我们运行INSERT INTO

ksql> INSERT INTO output_stream SELECT eventTime, 'Extra info' as extraColumn, sourceHostName FROM event_stream WHERE eventType = 'MatchingValue';

Message
-------------------------------
Insert Into query is running.
-------------------------------

如您所见,DESCRIBE EXTENDED 输出确实显示了正在处理的消息:

ksql> DESCRIBE EXTENDED output_stream;

Name                 : OUTPUT_STREAM
Type                 : STREAM
Key field            :
Key format           : STRING
Timestamp field      : Not set - using <ROWTIME>
Value format         : JSON
Kafka topic          : output_events (partitions: 4, replication: 1)

Field          | Type
--------------------------------------------
ROWTIME        | BIGINT           (system)
ROWKEY         | VARCHAR(STRING)  (system)
EVENTTIME      | VARCHAR(STRING)
EXTRACOLUMN    | VARCHAR(STRING)
SOURCEHOSTNAME | VARCHAR(STRING)
--------------------------------------------

Queries that write into this STREAM
-----------------------------------
InsertQuery_0 : INSERT INTO output_stream SELECT eventTime, 'Extra info' as extraColumn, sourceHostName FROM event_stream WHERE eventType = 'MatchingValue';

For query topology and execution plan please run: EXPLAIN <QueryId>

Local runtime statistics
------------------------
messages-per-sec:      0.01   total-messages:         1     last-message: 11/13/18 6:49:46 AM UTC
failed-messages:         0 failed-messages-per-sec:         0      last-failed:       n/a
(Statistics of the local KSQL server interaction with the Kafka topic output_events)

但主题本身没有消息:

ksql> print 'output_events' from beginning;
^C

也不是 KSQL 流:

ksql> SELECT * FROM OUTPUT_STREAM;
^CQuery terminated

因此,INSERT INTO 命令旨在针对现有的 CSAS/CTAS 目标流运行,而不是针对现有主题注册的源 STREAM。

让我们换个方式试试吧。首先,我们需要删除现有的流定义,并终止INSERT INTO 查询:

ksql> DROP STREAM OUTPUT_STREAM;
Cannot drop OUTPUT_STREAM.
The following queries read from this source: [].
The following queries write into this source: [InsertQuery_0].
You need to terminate them before dropping OUTPUT_STREAM.
ksql> TERMINATE InsertQuery_0;

Message
-------------------
Query terminated.
-------------------
ksql> DROP STREAM OUTPUT_STREAM;

Message
------------------------------------
Source OUTPUT_STREAM was dropped.
------------------------------------

现在创建目标流:

ksql> CREATE STREAM output_stream WITH (kafka_topic='output_events') AS SELECT eventTime, 'Extra info' as extraColumn, sourceHostName FROM event_stream WHERE eventType = 'MatchingValue';

Message
----------------------------
Stream created and running
----------------------------

请注意,在创建流时它也是running(之前它只是created)。现在查询流:

ksql> SELECT * FROM OUTPUT_STREAM;
1542091785207 | null | 2018-11-13-06:35:58 | Extra info | asgard

并检查基础主题:

ksql> PRINT 'output_events' FROM BEGINNING;
Format:JSON
{"ROWTIME":1542091785207,"ROWKEY":"null","EVENTTIME":"2018-11-13-06:35:58","EXTRACOLUMN":"Extra info","SOURCEHOSTNAME":"asgard"}

所以,您在 KSQL (raised here) 中遇到了一个错误,但幸运的是,可以通过完全使用更简单的 KSQL 语法来避免一个错误,将您的 CREATE STREAMINSERT INTO 查询合二为一。

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2018-06-15
    • 2022-12-02
    • 1970-01-01
    • 2017-12-20
    • 2015-11-05
    • 1970-01-01
    • 2012-09-28
    相关资源
    最近更新 更多