如何设置Kafka输出Topic数据格式以写入Cassandra对应表列

建议采用结构化JSON格式，是开发成本最低、适配性最好的选择，结构只需要匹配Cassandra表的两个字段即可，示例如下：
{"emp_id": 1, "emp_name": "张三"}

如果你的生产环境有schema管控需求、追求更高序列化性能，也可以选择Avro格式，需要搭配Schema Registry使用。JSON格式不需要额外依赖组件，Spring Kafka的Jackson序列化器可以直接完成序列化、反序列化操作，后续写入Cassandra时可以直接映射为实体类，几乎没有额外转换成本。

1. Kafka流处理模块实现（给消息加自增ID） #

你已经用了Spring Kafka生态，推荐直接用Kafka Streams实现流处理，自增ID的实现要注意分布式多实例场景下的唯一性，不要用本地全局变量，避免不同实例生成重复ID：

• 引入Kafka Streams的Spring Boot Starter依赖
• 配置Kafka Streams的状态存储（KeyValueStore），用于持久化当前最大的emp_id值
• 编写流处理逻辑：
  1. 消费原始输入Topic的仅包含姓名的消息
  2. 从状态存储中读取当前最大的emp_id，自增1后更新回状态存储
  3. 组装包含emp_id和emp_name的结构体，序列化后写入OutputTopic

注意：如果不需要严格连续的自增ID，也可以直接用雪花算法生成唯一ID，省略状态存储的配置，开发成本更低。请给Kafka Streams的状态存储配置持久化和副本数，避免服务重启或实例故障导致ID重复生成。

2. 消费OutputTopic写入Cassandra #

第一步：修正Cassandra建表语句（原语句多了多余逗号） #

CREATE TABLE emp(
   emp_id int PRIMARY KEY,
   emp_name text
);

第二步：编写Cassandra对应实体类 #

引入Spring Data Cassandra依赖后，编写和表结构映射的实体类：

import org.springframework.data.cassandra.core.mapping.PrimaryKey;
import org.springframework.data.cassandra.core.mapping.Table;

@Table("emp")
public class Emp {
    @PrimaryKey
    private Integer empId;
    private String empName;

    // 空构造、全参构造、getter、setter方法省略
}

第三步：编写Kafka消费者和存储逻辑 #

配置Kafka消费者的反序列化规则为JSON反序列化，监听OutputTopic，收到消息后直接反序列化为Emp对象，调用Spring Data Cassandra的Repository的save()方法即可完成数据写入，无需额外格式转换。

内容的提问来源于stack exchange，提问作者vigneshwar reddy