debezium mongodb源json接收器到cassandra(lenses.io)

zfycwa2u  于 2021-06-13  发布在  Cassandra
关注(0)|答案(1)|浏览(570)

我需要将一个json对象从mongodb接收到cassandra中的一列。我用的是extractnewdocumentstate,avroconverter。但看来我错了。avroconverter用于源还是汇?如果我把它用在源里,那么我也用在Flume里?

  1. {
  2. "name": "mongodb_source_connector",
  3. "config": {
  4. "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
  5. "tasks.max": "1",
  6. "mongodb.hosts": "rs0/mongo:27017",
  7. "mongodb.name": "dbserver1",
  8. "mongodb.user": "scorpion",
  9. "mongodb.password": "123123123",
  10. "database.whitelist": "ladiform",
  11. "database.history.kafka.bootstrap.servers": "kafka_3:9093",
  12. "transforms": "route,unwrap",
  13. "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  14. "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
  15. "transforms.route.replacement": "$3",
  16. "transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState",
  17. "transforms.unwrap.drop.tombstones": "false",
  18. "transforms.unwrap.delete.handling.mode": "drop",
  19. "transforms.unwrap.operation.header": "true",
  20. "key.converter": "io.confluent.connect.avro.AvroConverter",
  21. "value.converter": "io.confluent.connect.avro.AvroConverter",
  22. "key.converter.schema.registry.url": "http://schema-registry:8081",
  23. "value.converter.schema.registry.url": "http://schema-registry:8081"
  24. }

}

  1. {
  2. "name": "cassandra_sink_connector",
  3. "config": {
  4. "connector.class": "com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector",
  5. "tasks.max": "1",
  6. "topics": "test34",
  7. "connect.cassandra.port": "9042",
  8. "connect.cassandra.key.space": "test",
  9. "connect.cassandra.contact.points": "cassandra",
  10. "connect.cassandra.username": "cassandra",
  11. "connect.cassandra.password": "123123123",
  12. "connect.cassandra.kcql": "INSERT INTO test33 SELECT id, data FROM test34"
  13. }
s71maibg

s71maibg1#

源端和汇端的转换器必须相同,因为序列化在源端完成,反序列化在汇端完成。

相关问题