Debezium MongoDB源JSON接收器到Cassandra (LENSES.IO)

Question

我需要将一个JSON对象从MongoDB插入到Cassandra中的一个列中。我用的是ExtractNewDocumentState，AvroConverter。但看起来我错了。以及在源或宿中使用的AvroConverter？如果我在源代码中使用它，那么我在接收器中也使用它吗？

{
"name": "mongodb_source_connector",
"config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
    "tasks.max": "1",
    "mongodb.hosts": "rs0/mongo:27017",
    "mongodb.name": "dbserver1",
    "mongodb.user": "scorpion",
    "mongodb.password": "123123123",
    "database.whitelist": "ladiform",
    "database.history.kafka.bootstrap.servers": "kafka_3:9093",
    "transforms": "route,unwrap",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
    "transforms.route.replacement": "$3",
    "transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "drop",
    "transforms.unwrap.operation.header": "true",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
}

}

{
"name": "cassandra_sink_connector",
"config": {
    "connector.class": "com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector",
    "tasks.max": "1",
    "topics": "test34",
    "connect.cassandra.port": "9042",
    "connect.cassandra.key.space": "test",
    "connect.cassandra.contact.points": "cassandra",
    "connect.cassandra.username": "cassandra",
    "connect.cassandra.password": "123123123",
    "connect.cassandra.kcql": "INSERT INTO test33 SELECT id, data FROM test34"
}

Answer 1

源端和接收端的转换器必须相同，因为序列化是在源端完成的，而反序列化是在接收端完成的。