使用 kafka connect cassandra sink 在 Cassandra 中自动生成表

Question

我是using confluent.connect.cassandra.CassandraSinkConnector，为kafka连接cassandra sink。

我想知道是否可以使用 io.confluent.connect.cassandra.CassandraSinkConnector 作为连接器从 kafka 主题自动生成 cassandra 表。

如果可能，您能否建议设置什么配置来启用此功能。我已经尝试了文档中提到的所有配置，但我没有成功创建表。

这是我正在使用的配置：

{
  "name": "cassandra-test4",

  "config": {

    "connector.class": "io.confluent.connect.cassandra.CassandraSinkConnector",

    "tasks.max": "3",

    "topics": "orders-topic2",

    "cassandra.contact.points": "my_ip",

    "cassandra.keyspace": "test_cas",

    "cassandra.write.mode": "Insert",

    "cassandra.table.manage.enabled": "true",

    "cassandra.sink.route": "test_cas.orders",

    "key.converter.schema.registry.url": "http://localhost:8081",

    "value.converter.schema.registry.url": "http://localhost:8081",

    "value.converter": "io.confluent.connect.avro.AvroConverter",

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",

    "flush.size": "1",

    "cassandra.keyspace.create.enabled": "true",

    "name": "cassandra-test4"

  },

  "tasks": [

    {

      "connector": "cassandra-test4",

      "task": 0

    },

    {

      "connector": "cassandra-test4",

      "task": 1

    },

    {

      "connector": "cassandra-test4",

      "task": 2

    }

  ],

  "type": null

}

Answer 1

这应该通过将cassandra.keyspace.create.enabled 和cassandra.table.manage.enabled 属性设置为true 来完成。见documentation。

但请务必小心 - 在您的集群中很容易出现架构分歧，然后您需要执行额外的步骤才能从中恢复。最好在启动连接器之前预先创建表...