1

我对 Confluent KSql 很陌生，但对 Kafka 并不陌生。我有作为 Avro 序列化数据存在于 Kafka 中的现有主题。我已经启动并运行了 Confluent 模式注册表，并将 KSql 配置为指向注册表。

当我尝试根据我的一个主题创建一个表时，KSql 抱怨它找不到流。当我尝试在 KSql 中创建一个仅在 KSql 中流式传输我的主题的流时，似乎无法指向在注册表中有引用的我的 Avro 序列化主题。

有谁知道如何解决这两个问题？我想使用 KSql 的方式不适合它的功能吗？

更新

这里有更多细节

ksql> show topics;

 Kafka Topic                                                                                 | Registered | Partitions | Partition Replicas | Consumers | Consumer Groups
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 COM_FINDOLOGY_MODEL_REPORTING_OUTGOINGFEEDADVERTISERSEARCHDATA                              | false      | 2          | 2                  | 0         | 0
 COM_FINDOLOGY_MODEL_TRAFFIC_CPATRACKINGCALLBACK                                             | false      | 2          | 2                  | 0         | 0
 COM_FINDOLOGY_MODEL_TRAFFIC_ENTRYPOINTCLICK                                                 | true       | 10         | 3                  | 0         | 0

KSql 配置

#bootstrap.servers=localhost:9092
bootstrap.servers=host1:9092,host2:9092,host3:9092,host4:9092,host5:9092

#listeners=http://localhost:8088
listeners=http://localhost:59093

ksql.server.ui.enabled=true

ksql.schema.registry.url=http://host1:59092

注册表配置

# The host name advertised in ZooKeeper. Make sure to set this if running Schema Registry with multiple nodes.
host.name: x.x.x.x
listeners=http://0.0.0.0:59092

# Zookeeper connection string for the Zookeeper cluster used by your Kafka cluster
# (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
#kafkastore.connection.url=localhost:2181

# Alternatively, Schema Registry can now operate without Zookeeper, handling all coordination via
# Kafka brokers. Use this setting to specify the bootstrap servers for your Kafka cluster and it
# will be used both for selecting the master schema registry instance and for storing the data for
# registered schemas.
# (Note that you cannot mix the two modes; use this mode only on new deployments or by shutting down
# all instances, switching to the new configuration, and then starting the schema registry
# instances again.)
kafkastore.bootstrap.servers=PLAINTEXT://host1:9092,PLAINTEXT://host2:9092,PLAINTEXT://host3:9092,PLAINTEXT://host4:9092,PLAINTEXT://host5:9092

# The name of the topic to store schemas in
kafkastore.topic=_schemas

# If true, API requests that fail will include extra debugging information, including stack traces
debug=false

尝试通过声明外部主题来解决问题

ksql> register  topic xxx with (value_format='avro', kafka_topic='COM_FINDOLOGY_MODEL_REPORTING_OUTGOINGFEEDADVERTISERSEARCHDATA');
You need to provide avro schema file path for topics in avro format.
4

2 回答 2

0

REGISTER TOPIC是不推荐使用的语法。您应该使用CREATE STREAM（或CREATE TABLE，取决于您的数据访问要求）。

所以你的陈述看起来像这样：

CREATE STREAM MY_STREAM_1 \
  WITH (VALUE_FORMAT='AVRO', \
  KAFKA_TOPIC='COM_FINDOLOGY_MODEL_REPORTING_OUTGOINGFEEDADVERTISERSEARCHDATA');

请注意，为了便于阅读，我习惯于\换行；你不必这样做。

于 2018-08-20T14:42:25.950 回答
0

在更改了我从 Kafka 主题中使用的信息后，我解决了我遇到的问题，而不是使用整个主题内容。该主题包含使用创建的 Avro 编码数据（ok）ReflectionData。 KSql处理流中的非标准项有问题，但只要有相应的 KSql 数据类型，就会处理 ReflectionData 项。我通过在 KSql 中创建一个新流来解决这个问题，该流只选择了我需要的也与 KSql 兼容的项目。完成后，我可以从更大的流中处理我需要的东西。

评论我认为 KSql 中存在一定的缺陷，您必须在其中创建新的实际中间主题Kafka来处理数据。我认为更好的解决方案是将中间流视为View实际流。在将其解析为我理解的 KTable 之前，需要中间主题来保存累积和处理的项目。

于 2018-09-05T17:41:08.037 回答