在 ksqlDB 中创建的流显示 NULL 值答案

【问题标题】：The stream created in ksqlDB shows NULL value在 ksqlDB 中创建的流显示 NULL 值
【发布时间】：2021-08-11 18:19:52
【问题描述】：

我正在尝试在 ksqlDB 中创建一个流，以从 kafka 主题中获取数据并对其执行查询。

CREATE STREAM test_location (
  id VARCHAR,
  name VARCHAR,
  location VARCHAR
  )

 WITH (KAFKA_TOPIC='public.location',
       VALUE_FORMAT='JSON',
       PARTITIONS=10);

topic public.location 中的数据是 JSON 格式。

已更新主题消息。

print 'public.location' from beginning limit 1;
Key format: ¯\_(ツ)_/¯ - no data processed
Value format: JSON or KAFKA_STRING
rowtime: 2021/05/23 11:27:39.429 Z, key: <null>, value: {"sourceTable":{"id":"1","name":Sam,"location":Manchester,"ConnectorVersion":null,"connectorId":null,"ConnectorName":null,"DbName":null,"DbSchema":null,"TableName":null,"payload":null,"schema":null},"ConnectorVersion":null,"connectorId":null,"ConnectorName":null,"DbName":null,"DbSchema":null,"TableName":null,"payload":null,"schema":null}, partition: 3

创建流后，在创建的流上执行 SELECT 我在输出中得到 NULL。虽然主题有数据。

select * from test_location
>EMIT CHANGES limit 5;
+-----------------------------------------------------------------+-----------------------------------------------------------------+-----------------------------------------------------------------+
|ID                                                               |NAME                                                            |LOCATION                                                          |
+-----------------------------------------------------------------+-----------------------------------------------------------------+-----------------------------------------------------------------+
|null                                                             |null                                                             |null                                                             |
|null                                                             |null                                                             |null                                                             |
|null                                                             |null                                                             |null                                                             |
|null                                                             |null                                                             |null                                                             |
|null                                                             |null                                                             |null                                                             |
Limit Reached
Query terminated

这是来自 docker 文件的详细信息

version: '2'

services:

  ksqldb-server:
    image: confluentinc/ksqldb-server:0.18.0
    hostname: ksqldb-server
    container_name: ksqldb-server
    depends_on:
      - schema-registry
    ports:
      - "8088:8088"
    environment:
      KSQL_LISTENERS: "http://0.0.0.0:8088"
      KSQL_BOOTSTRAP_SERVERS: "broker:29092"
      KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
      # Configuration to embed Kafka Connect support.
      KSQL_CONNECT_GROUP_ID: "ksql-connect-01"
      KSQL_CONNECT_BOOTSTRAP_SERVERS: "broker:29092"
      KSQL_CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      KSQL_CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      KSQL_CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      KSQL_CONNECT_CONFIG_STORAGE_TOPIC: "_ksql-connect-01-configs"
      KSQL_CONNECT_OFFSET_STORAGE_TOPIC: "_ksql-connect-01-offsets"
      KSQL_CONNECT_STATUS_STORAGE_TOPIC: "_ksql-connect-01-statuses"
      KSQL_CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      KSQL_CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      KSQL_CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      KSQL_CONNECT_PLUGIN_PATH: "/usr/share/kafka/plugins"

更新：这是我在 Kafka 中看到的主题中的一条消息

{
   "sourceTable": {
      "id": "1",
      "name": Sam,
      "location": Manchester,
      "ConnectorVersion": null,
      "connectorId": null,
      "ConnectorName": null,
      "DbName": null,
      "DbSchema": null,
      "TableName": null,
      "payload": null,
      "schema": null
   },
   "ConnectorVersion": null,
   "connectorId": null,
   "ConnectorName": null,
   "DbName": null,
   "DbSchema": null,
   "TableName": null,
   "payload": null,
   "schema": null
}

我缺少哪个步骤或配置？

【问题讨论】：

请出示您的实际主题数据
@OneCricketeer 您好，我已经更新了 ksqlDB 主题中的主题，还添加了它在 Kafka 主题中的外观...数据中的 null 实际上是 null 字段...
这个数据来自debezium？您的架构+有效负载完全为空，因此没有要获取的 id、name、location 字段。此外，为了解析这些数据，您需要在 ksql 中定义一个 payload 字段或告诉 Debezium 从事件中提取有效负载
@OneCricketeer 此数据来自 SCDF 处理器进入 Kafka。它基本上是一个 ETL，在作为源连接器的开头也包含 debezium，但具体来说，这个主题是通过将数据处理成 SCDF 来创建的。
好吧，好吧，我的评论仍然有效。正如您所展示的，您的消息根本没有id,name,location，因此您应该期望在KSQL 中只看到空值。如果您希望能够解析该数据，则需要将 KSQL 设置为至少有一个 payload STRUCT 字段

标签： apache-kafka confluent-platform ksqldb

【解决方案1】：

鉴于您的有效负载，您需要声明架构嵌套，因为 id、name 和 location 不是 Json 中的“顶级”字段，但它们嵌套在 sourceTable 中。

CREATE STREAM est_location (
  sourceTable STRUCT<id VARCHAR, name VARCHAR, location VARCHAR>
)

定义架构时无法“解包”数据，但架构必须与主题中的内容相匹配。除了sourceTable，您还可以将ConnectorVersion 等添加到架构中，因为它们也是JSON 中的“顶级”字段。底线是，ksqlDB 中的该列只能在顶级字段上声明。其他所有内容都是嵌套数据，您可以使用 STRUCT 类型访问它们。

当然以后，当您查询est_location 时，您可以通过sourceTable->id 等引用各个字段。

如果您想取消嵌套架构，也可以声明派生的 STREAM：

CREATE STREAM unnested_est_location AS
  SELECT sourceTable->id AS id,
         sourceTable->name AS name,
         sourceTable->location AS location
  FROM est_location;

当然，这会将数据写入新主题。

【讨论】：