SF KAFKA CONNECTOR 详细信息：表没有兼容的架构 - 雪花卡夫卡连接器答案

【问题标题】：SF KAFKA CONNECTOR Detail: Table doesn't have a compatible schema - snowflake kafka connectorSF KAFKA CONNECTOR 详细信息：表没有兼容的架构 - 雪花卡夫卡连接器
【发布时间】：2020-10-23 13:04:51
【问题描述】：

我已经设置了雪花 - kafka 连接器。我在雪花中设置了一个示例表（kafka_connector_test），其中两个字段都是 VARCHAR 类型。字段为 CUSTOMER_ID 和 PURCHASE_ID。

这是我为连接器创建的配置

curl -X POST \
  -H "Content-Type: application/json" \
  --data '{
"name":"kafka_connector_test",
"config":{
"connector.class":"com.snowflake.kafka.connector.SnowflakeSinkConnector",
"tasks.max":"2",
"topics":"kafka-connector-test",
"snowflake.topic2table.map": "kafka-connector-test:kafka_connector_test",
"buffer.count.records":"10000",
"buffer.flush.time":"60",
"buffer.size.bytes":"5000000",
"snowflake.url.name":"XXXXXXXX.snowflakecomputing.com:443",
"snowflake.user.name":"XXXXXXXX",
"snowflake.private.key":"XXXXXXXX",
"snowflake.database.name":"XXXXXXXX",
"snowflake.schema.name":"XXXXXXXX",
"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"value.converter":"com.snowflake.kafka.connector.records.SnowflakeJsonConverter"}}'\

我将数据发送到我在连接器配置中配置的主题。

{"CUSTOMER_ID" : "test_id", "PURCHASE_ID" : "purchase_id_test"}

然后当我检查 kafka-connect 服务器时，我收到以下错误：

[SF KAFKA CONNECTOR] Detail: Table doesn't have a compatible schema

我需要在 kafka connect 或 snowflake 中设置什么来说明 json 的哪些部分进入表格的哪些列？不知道如何指定它如何解析 json。

我也设置了一个不同的主题，并没有在雪湖中创建表格。我能够填充此表，但连接器创建了一个包含 2 列 RECORD_METADATA 和 RECORD_CONTENT 的表。但我不想编写计划作业来解析这个我想直接插入到可查询的表中。

【问题讨论】：

标签： snowflake-cloud-data-platform apache-kafka-connect

【解决方案1】：

Snowflake Kafka 连接器按设计将数据写入 json。默认列 RECORD_METADATA 和 RECORD_CONTENT 是变体。如果您想查询它们，您可以在表格顶部创建一个视图来实现您的目标，并且您不需要计划的工作

所以，您的由连接器创建的表将类似于

RECORD_METADATA, RECORD_CONTENT

{metadata fields in json}, {"CUSTOMER_ID" : "test_id", "PURCHASE_ID" : "purchase_id_test"}

您可以创建一个视图来显示您的数据

create view v1 as 
select RECORD_CONTENT:CUSTOMER_ID::text CUSTOMER_ID,
RECORD_CONTENT:PURCHASE_ID::text PURCHASE_ID

您的查询将是

select CUSTOMER_ID , PURCHASE_ID from v1

附言。如果您想创建自己的表，则需要使用 variant 作为数据类型，而不是 varchar

【讨论】：

【解决方案2】：

参考this github issue，目前似乎不支持它

【讨论】：