ksql-是否可以从多个主题创建流并获得完整的事件负载？

esyap4oy 于 2021-06-06 发布在 Kafka

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

我们需要听多个主题，并在每个主题的事件中寻找特定的领域。每个主题事件都是json格式的，并且json格式的固定字段很少。需要从所有这些多个主题中筛选事件，并在每个事件负载中查找特定字段。如果此字段值与特定格式匹配，则将这些事件从不同主题发送到一个固定主题，该主题可由另一个使用者进一步处理。
我们正在寻找ksql是否能在这个场景中提供帮助—我们从多个主题创建一个流，并根据ksql流中的固定列过滤数据，然后将其推送到新主题。我的问题是：1）是否可以从多个主题创建ksql流？2）是否可以将主题的完整事件负载作为ksql流中的一列获取？
在高层次上（使用错误的ksql语法），我正在寻找

CREATE STREAM my_all_topics (myFixedFiedl1 varchar, eventPayload varchar) WITH (value_format = 'json', kafka_topic_LIST='topic1, topic2, topic3');

CREATE STREAM mytopic_stream (myFixedFiedl1 varchar, eventPayload varchar) with (kafka_topic='my-final-topic-name', value_format='json')
as select myFixedField1, eventPayload from my_all_topics where myFixedField1 like 'myprefix%';

apache-kafka ksqldb

来源：https://stackoverflow.com/questions/53055605/ksql-is-it-possible-to-create-stream-from-multiple-topics-and-get-full-event-p

2条答案

按热度按时间

bfhwhh0e1#

你不能完全按照你的意愿去做-一个ksql STREAM 它来源于一个而且只有一个Kafka主题。
但你可以用ksql的 INSERT INTO 实现你想要的功能。
为源主题建模：

CREATE STREAM source_a (myFixedField1 varchar, eventPayload varchar) WITH (kafka_topic='topic_a', value_format='json')
CREATE STREAM source_b (myFixedField1 varchar, eventPayload varchar) WITH (kafka_topic='topic_b', value_format='json')
CREATE STREAM source_c (myFixedField1 varchar, eventPayload varchar) WITH (kafka_topic='topic_c', value_format='json')

基于第一个源主题创建目标主题：

CREATE STREAM mytopic_stream (myFixedField1 varchar, eventPayload varchar) AS SELECT myFixedField1, eventPayload from source_a where myFixedField1 like 'myprefix%';

指定从其余源主题插入到目标主题：

INSERT INTO mytopic_stream SELECT myFixedField1, eventPayload from source_b where myFixedField1 like 'myprefix%';
INSERT INTO mytopic_stream SELECT myFixedField1, eventPayload from source_c where myFixedField1 like 'myprefix%';

另请参见
https://www.youtube.com/watch?v=z508vddtp_m
https://docs.confluent.io/current/ksql/docs/tutorials/basics-local.html#insert-进入

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

6qqygrtg2#

我不确定，但你似乎能把溪流和水结合起来 JOIN .

CREATE STREAM mytopic_stream AS
    SELECT A.*, B.*, C.*
    FROM stream_A A
        JOIN stream_B B ON A.key = B.key_for_A
        JOIN stream_C C ON A.key = B.key_for_A

如果您还没有向ksql注册kafka主题，那么您必须首先处理好这一步。

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

我来回答

ksql-是否可以从多个主题创建流并获得完整的事件负载？

2条答案

相关问题

热门标签

最新问答