Apache Flink如何读取Kafka的Confluent Avro数据写入Cassandra

方案一：使用Flink SQL（优先推荐，操作最简便） #

Flink SQL原生兼容带Confluent Schema Registry的Avro格式，无需自行开发序列化/反序列化逻辑，仅需通过建表配置即可完成整个链路搭建：

• 第一步：创建Kafka源表，指定格式为avro-confluent并配置Schema Registry地址

CREATE TABLE kafka_avro_source (
  -- 按实际Avro Schema定义对应字段名与类型即可
  id INT,
  name STRING,
  event_time TIMESTAMP
) WITH (
  'connector' = 'kafka',
  'topic' = '你的Kafka主题名称',
  'properties.bootstrap.servers' = '你的Kafka Broker地址',
  'properties.group.id' = '自定义消费组ID',
  'scan.startup.mode' = 'latest-offset', -- 可根据需求调整为earliest-offset等其他消费模式
  'format' = 'avro-confluent',
  'avro-confluent.schema-registry.url' = '你的Confluent Schema Registry服务地址'
);

• 第二步：创建Cassandra结果表

CREATE TABLE cassandra_sink (
  id INT PRIMARY KEY,
  name STRING,
  event_time TIMESTAMP
) WITH (
  'connector' = 'cassandra',
  'host' = '你的Cassandra服务地址',
  'port' = '9042',
  'keyspace' = '目标KeySpace名称',
  'table-name' = '目标表名称',
  'username' = 'Cassandra登录账号（无鉴权可省略）',
  'password' = 'Cassandra登录密码（无鉴权可省略）'
);

• 第三步：执行写入语句，即可完成整链路数据流转

INSERT INTO cassandra_sink SELECT * FROM kafka_avro_source;

方案二：DataStream API自定义反序列化逻辑 #

如果你需要沿用已有的DataStream API代码栈，可以自行实现DeserializationSchema<Row>适配Confluent Avro格式：

1. 首先引入Confluent Schema Registry客户端依赖
2. 自定义反序列化Schema，核心逻辑是调用Confluent官方的KafkaAvroDeserializer解析二进制消息为Avro GenericRecord，再按字段映射转换为Flink的Row类型，核心代码示例如下：

public class ConfluentAvroToRowDeserializationSchema implements DeserializationSchema<Row> {
    private transient KafkaAvroDeserializer avroDeserializer;
    private final String schemaRegistryUrl;
    private final RowType rowType;

    // 构造方法传入Schema Registry地址和对应Row的类型定义
    public ConfluentAvroToRowDeserializationSchema(String schemaRegistryUrl, RowType rowType) {
        this.schemaRegistryUrl = schemaRegistryUrl;
        this.rowType = rowType;
    }

    @Override
    public void open(InitializationContext context) {
        Map<String, Object> config = new HashMap<>();
        config.put("schema.registry.url", schemaRegistryUrl);
        config.put("specific.avro.reader", false);
        this.avroDeserializer = new KafkaAvroDeserializer(null, config);
    }

    @Override
    public Row deserialize(byte[] bytes) throws IOException {
        GenericRecord record = (GenericRecord) avroDeserializer.deserialize("", bytes);
        Row row = Row.withPositions(rowType.getFieldCount());
        for (int i = 0; i < rowType.getFieldCount(); i++) {
            String fieldName = rowType.getFieldNames().get(i);
            row.setField(i, record.get(fieldName));
        }
        return row;
    }

    @Override
    public boolean isEndOfStream(Row row) {
        return false;
    }

    @Override
    public TypeInformation<Row> getProducedType() {
        return TypeInformation.of(rowType);
    }
}

将上述自定义反序列化Schema传入你的Kafka Consumer，得到的数据流就是Row类型，直接对接你已实现的CassandraSink即可使用。

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