从MSSQL vis kafka connector到具有嵌套类型的elasticsearch的大容量数据更新失败

Question

{
   "schema":{
      "type":"struct",
      "fields":[
         {
            "type":"int32",
            "optional":false,
            "field":"prop"
         },
         {
            "type":"string",
            "optional":true,
            "field":"roles"
         }
 ],
      "optional":false
   },
   "payload":{ "prop":100, "roles":"[{"role":"actor"},{"role":"director"}]"}

Error was [{"type":"mapper_parsing_exception","reason":"object mapping for [roles] tried to parse field [roles] as object, but found a concrete value"}

name=prop-test
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
connection.url=<elasticseach url>

tasks.max=1

topics=test_prop
type.name=prop

#transforms=InsertKey, ExtractId

transforms.InsertKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.InsertKey.fields=prop

transforms.ExtractId.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.ExtractId.field=prop

name=test_prop_source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:sqlserver://*.*.*.*:1433;instance=databaseName=test;
connection.user=*****
connection.password=*****
query=EXEC <store proc>
mode=bulk
batch.max.rows=2000000
topic.prefix=test_prop
transforms=createKey,extractInt
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.createKey.fields=prop

transforms.extractInt.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractInt.field=prop

    bootstrap.servers=localhost:9092

    key.converter=org.apache.kafka.connect.json.JsonConverter
    value.converter=org.apache.kafka.connect.json.JsonConverter

    key.converter.schemas.enable=true
    value.converter.schemas.enable=true

Answer 1

事实上，jdbc源连接器将始终将columnname视为fieldname，将columnvalue视为Value。这种从字符串到数组的转换在现有的jdbc源连接器支持下是不可能的，它必须是自定义转换或自定义插件才能启用。

从MSSQL中获取数据并将其插入弹性搜索的最佳选择是使用logstash。它有丰富的过滤器插件，可以使来自MSSQL的数据以任何所需的格式流到任何所需的JSON输出环境（logstash/kafka主题）

流：MSSQL-->logstash-->Kakfa主题-->Kafka弹性接收器连接器-->弹性搜索