如何将结构化记录直接从KSQL接收到连接器中（例如，InfluxDB)

Question

我正在尝试将数据直接从KSQL接收到InfluxDB（或任何其他需要定义的连接器）中。我能够在简单的情况下使事情正常工作，但当模式需要复杂类型时，我就开始遇到麻烦。（即InfuxDB的标签）。

下面是我的流/模式的一个示例：

 Field    | Type                                                   
-------------------------------------------------------------------
 ROWKEY   | VARCHAR(STRING)  (primary key)
 FIELD_1  | VARCHAR(STRING)                                        
 FIELD_2  | VARCHAR(STRING)                                        
 FIELD_3  | VARCHAR(STRING)                                        
 FIELD_4  | DOUBLE                                                 
 TAGS     | MAP<STRING, VARCHAR(STRING)> 

如果我手动创建一个AVRO模式并从一个简单的生产者填充记录，我就可以通过这里的入门指南并嵌入InfluxDB的标记。

但是，当我转移到KSQL时，如果我试图将AVRO流直接存入InfluxDB中，就会丢失关于复杂类型（标记）的信息。我注意到这篇博客文章中的警告，“警告ksqldb/ksql还不能以与此连接器兼容的Avro格式写入数据”

接下来，我尝试将AVRO流转换为JSON格式，但现在我明白了我必须在每个记录中指定模式，类似于这个问题提出的问题。我还没有能够将AVRO流转换为JSON流并同时嵌入模式和有效负载。

最后，我看到了kafkacat的“摇摆解决方案”，但这将迫使我将记录从KSQL转储到kafkacat中，然后再返回到Kafka中，最后才到达Infux。

是否有一种方法可以直接将JSON或AVRO格式的KSQL中的复杂记录接收到连接器中？

Answer 1

我认为ksqlDB还不能以InfluxDB要求的格式输出AVRO数据的原因是，由于AVRO映射需要非空键，而SQLmap 类型允许空键，所以它不会将tags字段作为AVROmap类型输出。因此，ksqlDB将映射序列化为键值条目的Avro数组。

要获得与Avro一起工作的东西，您需要以下两种方法：

1. 对非空类型的支持:https://github.com/confluentinc/ksql/issues/4436或
2. 支持使用现有Avro架构:https://github.com/confluentinc/ksql/issues/3634

请随时投票/评论这些问题，以提高他们的形象。

以前，基于JSON的解决方案是行不通的，因为正如您所指出的，连接器需要在有效负载中嵌入JSON模式。但是，汇合平台/模式注册中心的最新版本支持模式注册中心中的JSON模式。因此，虽然我还没有尝试过，但升级到最新的CP版本可能意味着基于JSON的解决方案可以工作。如果不是，那么可能需要提出一个JIRA/GitHub票证来升级相应的组件，使其工作。