elasticsearch - 无法使用 Kafka 连接将记录从 Kafka MSK 发送到 Elasticsearch

Question

这是我上一个问题的后续问题。因为问题变得很长所以我想创建一个新的这是我的旧问题 Kafka connect setup to send record from Aurora using AWS MSK

所以我的连接器运行时出现了一些错误，这就是我的记录没有进入 Elastic Search 的原因

这是我的属性文件

quickstart-elasticsearch.properties

name=elasticsearch-sink
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=fspauditlambda
key.ignore=true
connection.url=https://drtrrterterterterterst-1.es.amazonaws.com
type.name=kafka-connect

这是我的 connect-standalone.properties 详细信息

bootstrap.servers=b-3.rtyrtyty.amazonaws.com:9092,b-6.rtyrtyty.amazonaws.com:9092,b-1.rtyrtyty.us-east-1.amazonaws.com: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

offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
plugin.path=/usr/local/confluent/share/java

然后我启动我的连接器，但是当我这样做时，我得到类似

的错误

   org.apache.kafka.connect.errors.ConnectException: Cannot create mapping 
{"kafka-connect":{"properties":{"ID":{"type":"text","fields":{"keyword": -- {"root_cause":[{"type":"illegal_argument_exception","reason":"Types cannot be provided in put mapping requests, unless the include_type_name parameter is set to true."}],"type":"illegal_argument_exception","reason":"Types cannot be provided in put mapping requests, 
unless the include_type_name parameter is set to true."}

当我将属性更改为以下时

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000

它在 Elasticsearch 中创建了 Indises，但数据没有运行并低于错误

[2020-01-03 12:27:12,906] ERROR WorkerSinkTask{id=elasticsearch-sink-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:177)
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
        at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:560)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.connect.errors.DataException: Cannot infer mapping without schema.
        at io.confluent.connect.elasticsearch.Mapping.inferMapping(Mapping.java:84)
        at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.createMapping(JestElasticsearchClient.java:292)
        at io.confluent.connect.elasticsearch.Mapping.createMapping(Mapping.java:66)
        at io.confluent.connect.elasticsearch.ElasticsearchWriter.write(ElasticsearchWriter.java:260)

这个配置我也试过

topic.schema.ignore=true

但是同样的错误

更新我的 MYSQL 表定义

CREATE TABLE FSP_AUDIT (
ID    NVARCHAR(255) NOT NULL,
VERSION    numeric(10,0) ,
ACTION_TYPE    NVARCHAR(255) ,
EVENT_TYPE    NVARCHAR(255) ,
CLIENT_ID    NVARCHAR(25) ,
DETAILS    TEXT(40000) ,
OBJECT_TYPE    NVARCHAR(255) ,
UTC_DATE_TIME    TIMESTAMP(6) NOT NULL,
POINT_IN_TIME_PRECISION    NVARCHAR(255) ,
TIME_ZONE    NVARCHAR(255) ,
TIMELINE_PRECISION    NVARCHAR(255) ,
GROUP_ID    NVARCHAR(255) ,
OBJECT_DISPLAY_NAME    NVARCHAR(200) ,
OBJECT_ID    NVARCHAR(255) ,
USR_DISPLAY_NAME    NVARCHAR(1500) ,
USR_ID    NVARCHAR(255) ,
PARENT_EVENT_ID    NVARCHAR(255) ,
NOTES    NVARCHAR(1000) ,
SUMMARY    NVARCHAR(4000) ,
ADTEVT_TO_UTC_DT    TIMESTAMP(6) ,
ADTEVT_TO_DATE_PITP    NVARCHAR(255) ,
ADTEVT_TO_DATE_TZ    NVARCHAR(255) ,
ADTEVT_TO_DATE_TP    NVARCHAR(255) ,
 PRIMARY KEY(ID)
);

要预先创建 Elastic 搜索的预期架构这样对吗 ？请提出更改建议

{ 
   "schema":{ 
      "type":"struct",
      "fields":[ 
         { 
            "type":"string",
            "optional":false,
            "field":"ID"
         },
         { 
            "type":"integer",
            "optional":true,
            "field":"VERSION"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"ACTION_TYPE"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"EVENT_TYPE"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"CLIENT_ID"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"DETAILS"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"OBJECT_TYPE"
         },
         { 
            "type":"int64",
            "optional":false,
            "name":"org.apache.kafka.connect.data.Timestamp",
            "version":1,
            "field":"UTC_DATE_TIME"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"POINT_IN_TIME_PRECISION"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"TIME_ZONE"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"TIMELINE_PRECISION"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"GROUP_ID"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"OBJECT_DISPLAY_NAME"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"OBJECT_ID"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"USR_DISPLAY_NAME"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"USR_ID"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"PARENT_EVENT_ID"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"NOTES"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"SUMMARY"
         },
         { 
            "type":"int64",
            "optional":true,
            "name":"org.apache.kafka.connect.data.Timestamp",
            "version":1,
            "field":"ADTEVT_TO_UTC_DT"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"ADTEVT_TO_DATE_PITP"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"ADTEVT_TO_DATE_TZ"
         },
         { 
            "type":"string",
            "optional":true,
            "field":"ADTEVT_TO_DATE_TP"
         }
      ],
      "optional":false,
      "name":"FSP_AUDIT"
   }
}

Answer 1

第二个错误是说您需要一个架构，因此您对属性文件的第一次编辑没问题。

第一个错误是因为 Elasticsearch 6+ 中的行为更改需要您自己定义索引映射

Types cannot be provided in put mapping requests, unless the include_type_name parameter is set to true