KSQL Kafka 键格式化
KSQL Kafka formatting of keys
我的一个主题有一个字符串-json作为键-{"city":"X","id":22}。
在我的 ksql 语句中，我想将它提取到 2 个不同的字段而不是一个，这样我以后可以过滤和加入。在文档中，它似乎只允许我将整个字符串粘贴到密钥中，而不是允许将其格式化为 JSON（就像 FORMAT_VALUE），请参见下文......关于优雅解决方案的任何建议这个？
VALUE_FORMAT（必填）指定主题中消息值的序列化格式。支持的格式：JSON、DELIMITED 和 AVRO
KEY 将 Kafka 主题中的消息键与 KSQL 流中的列相关联。
您应该能够使用 ROWKEY 列上的 EXTRACTJSONFIELD UDF 来执行此操作。 ROWKEY 列是架构中的 "column"，它包含给定行的键。 EXTRACTJSONFIELD UDF 允许您从包含 json 字符串的列中提取字段。因此，对于您的示例，您可以 运行:
SELECT EXTRACTJSONFIELD(ROWKEY, '$.city') AS city, EXTRACTJSONFIELD(ROWKEY, '$.id') AS id FROM FOO;
我的一个主题有一个字符串-json作为键-{"city":"X","id":22}。 在我的 ksql 语句中，我想将它提取到 2 个不同的字段而不是一个，这样我以后可以过滤和加入。在文档中，它似乎只允许我将整个字符串粘贴到密钥中，而不是允许将其格式化为 JSON（就像 FORMAT_VALUE），请参见下文......关于优雅解决方案的任何建议这个？
VALUE_FORMAT（必填）指定主题中消息值的序列化格式。支持的格式：JSON、DELIMITED 和 AVRO
KEY 将 Kafka 主题中的消息键与 KSQL 流中的列相关联。
您应该能够使用 ROWKEY 列上的 EXTRACTJSONFIELD UDF 来执行此操作。 ROWKEY 列是架构中的 "column"，它包含给定行的键。 EXTRACTJSONFIELD UDF 允许您从包含 json 字符串的列中提取字段。因此，对于您的示例，您可以 运行:
SELECT EXTRACTJSONFIELD(ROWKEY, '$.city') AS city, EXTRACTJSONFIELD(ROWKEY, '$.id') AS id FROM FOO;