例子:
{"id":"1","firstName":"abc","lastName":"xyz","dob":"12/09/1995","age":"23"}
此消息结构位于Kafka主题中,但我希望在ElasticSearch中对其进行索引,如下所示
{"id":"1","name"{"firstName":"abc","lastName":"xyz"},"dob":"12/09/1995","age":"23"}
我怎样才能做到这一点?
两种选择:
针对Kafka主题中的数据进行流处理。使用Kafka流,您可以根据需要对数据模型进行争论。ksql可以反过来解决这个问题,但不支持 创建 尚未构造。其他流处理选项包括Flink、Spark流等。
在通过Kafka Connect时,使用单个消息转换修改数据。没有预构建的转换可以做到这一点,但是您可以使用 the API .
免责声明:我为Confluent工作,是开源ksql项目背后的公司,Kafka Streams的构造,Kafka Connect等。
