有没有可能把debezium产生的卡夫卡信息沉到雪花上

Question

我使用debezium-ui回购来测试debezium mysql cdc特性，消息通常可以流。

在kafka中，创建mysql连接的请求主体如下：

    {
      "name": "inventory-connector",  
      "config": {  
          "connector.class": "io.debezium.connector.mysql.MySqlConnector",
          "tasks.max": "1",  
          "database.hostname": "dbzui-db-mysql",  
          "database.port": "3306",
          "database.user": "mysqluser",
          "database.password": "mysql",
          "database.server.id": "184054",  
          "database.server.name": "inventory-connector-mysql",  
          "database.include.list": "inventory",  
          "database.history.kafka.bootstrap.servers": "dbzui-kafka:9092",  
          "database.history.kafka.topic": "dbhistory.inventory"  
      }
    }

然后我需要把卡夫卡的信息转化成雪花，我的团队使用的数据仓库。我创建了一个雪花接收器连接器来下沉它，请求体如下所示：

{
    "name": "kafka2-04",
    "config": {
        "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
        "tasks.max": 1,
        "topics": "inventory-connector-mysql.inventory.orders",
        "snowflake.topic2table.map": "inventory-connector-mysql.inventory.orders:tbl_orders",
        "snowflake.url.name": "**.snowflakecomputing.com",
        "snowflake.user.name": "kafka_connector_user_1",
        "snowflake.private.key": "*******",
        "snowflake.private.key.passphrase": "",
        "snowflake.database.name": "kafka_db",
        "snowflake.schema.name": "kafka_schema",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "com.snowflake.kafka.connector.records.SnowflakeJsonConverter",
        "header.converter": "org.apache.kafka.connect.storage.SimpleHeaderConverter",
        "value.converter.schemas.enable":"true"
    }
}

但是在运行之后，数据沉入我的雪花中就像：雪花数据，雪花表中的模式与mysql表不同。是我的接收器连接器配置不正确，还是不可能用SnowflakeSinkConnector接收debezium生成的kafka数据。

Answer 1

这是雪花中的默认行为，并被记录为这里

由Kafka连接器加载的每个雪花表都有一个由两个不同列组成的模式：

RECORD_CONTENT. This contains the Kafka message.

RECORD_METADATA. This contains metadata about the message, for example, the topic from which the message was read.

如果雪花创建了表，那么该表只包含这两列。如果用户为Kafka连接器创建要向其添加行的表，则表可以包含这两列以上(任何额外的列都必须允许空值，因为连接器中的数据不包括这些列的值)。