2

我们正在尝试将 Kafka KSQL 迁移到我们的系统，并希望分享一些我们在此过程中无法解决的问题。我们的集群中有 3 个 Kafka 节点，每个服务器都有： 

8 CORE  
50G+ RAM  
100G ssd

在每台服务器上，我们都有 zookeeper 来管理集群。所有操作系统限制都增加了，因此节点可以使用比它需要的更多的资源： 

Xmx: 10G  
Xms: 10G  
nofiles: 500000

目前，从生产者到集群的流量很小（每秒约 10 条消息）。现在我们只有一个生产者，消息格式是：

{"user_id": <id|INT>, "action_id": <id|INT>, "amount": <amount|FLOAT>}

Kafka 中的主题分为 6 个分区和 1 个复制：

Topic:<some_topic>   PartitionCount:6        ReplicationFactor:1     Configs:
        Topic: <some_topic>  Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: <some_topic>  Partition: 1    Leader: 1       Replicas: 1     Isr: 1
        Topic: <some_topic>  Partition: 2    Leader: 2       Replicas: 2     Isr: 2
        Topic: <some_topic>  Partition: 3    Leader: 0       Replicas: 0     Isr: 0
        Topic: <some_topic>  Partition: 4    Leader: 1       Replicas: 1     Isr: 1
        Topic: <some_topic>  Partition: 5    Leader: 2       Replicas: 2     Isr: 2

现在，当然，节点没有得到充分利用，在 kafka 方面一切都很好）

我们希望在 Kafka 之上使用 KSQL，以便能够使用 SQL 过滤进入我们系统的数据。以下是 KSQL 服务器资源：

32 CORE
100G+ RAM
50G+ ssd

我们只有一张桌子：

 Field   | Type                      
-------------------------------------
 ROWTIME   | BIGINT           (system) 
 ROWKEY    | VARCHAR(STRING)  (system) 
 ACTION_ID | INTEGER                   
 USER_ID   | INTEGER                   
 AMOUNT    | DOUBLE

这是创建表的命令：

create table <some_table> (action_id INT, user_Id INT, amount DOUBLE) with (KAFKA_TOPIC='<some_topic>', VALUE_FORMAT='JSON', KEY = 'user_id');

在我们的应用程序中，我们需要通过 user_id 订阅表，如下所示：

SELECT * FROM <some_table> WHERE USER_ID=<some_user_id>;

对于生产 KSQL 服务器配置，我们使用来自 confluent 的官方推荐： https ://docs.confluent.io/current/ksql/docs/installation/server-config/config-reference.html#recommended-ksql-production-settings

KSQL 服务器的操作系统和软件限制也有所增加：

Xmx: 10G  (we have tried till 50G)
Xms: 10G  (we have tried till 50G)
nofiles: 500000

如果我们只使用一个订阅，我们不会遇到任何问题（在这种情况下一切都很好）。
但我们总共需要超过 200000 个订阅。因此，当我们尝试获得 100-200 个并行订阅时，我们的客户端会出现“读取超时”。在服务器中，我们没有看到任何可能影响 KSQL 的异常负载。
我们假设这个问题只与 KSQL 有关，因为当我们尝试使用另一个 KSQL 服务器（在不同的机器上）时，同时我们可以看到第二个服务器工作正常并且可以处理大约 1-20 个订阅。

我在与 KSQL 服务器连接的 Internet 上找不到任何基准，在文档中，我也找不到任何提及 KSQL 用例的内容，也许它的设计目的只是为少量连接提供大量数据，或者我们的系统配置错误，因此我们应该修复它以使用该软件实现我们的目标。
任何建议都会有所帮助。
提前致谢 ）

4

1 回答 1

0

您在使用 ksqlDB 时遇到可伸缩性问题的原因是您使用推送查询的方式并非旨在使用它们......但是！

推送查询：

SELECT * FROM <some_table> WHERE USER_ID=<some_user_id>;

您使用它来订阅特定用户的更新似乎是一件完全明智的事情。

但是，在 ksql 版本中，您使用的此类推送查询仅适用于在 CLI 上执行命令的人类。每个这样的查询将在内部消耗大量服务器资源并消耗源主题中的所有行。

基本上，推送查询不会扩展。

ksqlDB 团队正在积极致力于增强 ksql 以支持这种确切的用例风格，因为我们认识到这是一件很常见的事情。（见https://github.com/confluentinc/ksql/issues/5517）。

同时，实现这一点的方法是使用您自己的消费者直接从 Kafka 消费数据并在本地进行过滤。

于 2020-06-01T11:38:09.630 回答