flink 写入Es 报错

[wasys1314]

` public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 开启checkpoint
// 设置Checkpoint的时间间隔为1000ms做一次Checkpoint/其实就是每隔1000ms发一次Barrier!
env.enableCheckpointing(1000);
// 设置State状态存储介质
env.setStateBackend(new FsStateBackend("file:///D:\data\ckp"));
//===========类型2:建议参数===========
//设置两个Checkpoint 之间最少等待时间,如设置Checkpoint之间最少是要等 500ms(为了避免每隔1000ms做一次Checkpoint的时候,前一次太慢和后一次重叠到一起去了)
//如:高速公路上,每隔1s关口放行一辆车,但是规定了两车之前的最小车距为500m
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);//默认是0
//设置如果在做Checkpoint过程中出现错误，是否让整体任务失败：true是 false不是
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);//默认值为0，表示不容忍任何检查点失败

    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

    String sourceSql = "CREATE TABLE airline (\n" +
            "     id INT PRIMARY KEY,\n" +
            "     AirlineNo INT,\n" +
            "     CompanyNo STRING,\n" +
            "     StartCity STRING,\n" +
            "     EndCity STRING,\n" +
            "     LeaveTime DATE\n" +
            "     ) WITH (\n" +
            "     'connector' = 'mysql-cdc',\n" +
            "     'hostname' = 'localhost',\n" +
            "     'port' = '3306',\n" +
            "     'username' = 'root',\n" +
            "     'password' = 'root',\n" +
            "     'scan.startup.mode' = 'initial',\n" +
            "     'server-time-zone' = 'UTC',\n" +
            "     'database-name' = 'testuser',\n" +
            "     'table-name' = 'airline')";

    String sinkSql = "CREATE TABLE es_airline (\n" +
            "     id INT ,\n" +
            "     AirlineNo INT,\n" +
            "     CompanyNo STRING,\n" +
            "     StartCity STRING,\n" +
            "     EndCity STRING,\n" +
            "     LeaveTime DATE\n" +
            "  ,PRIMARY KEY (id) NOT ENFORCED\n" +
            ") WITH (\n" +
            "  'connector' = 'elasticsearch-7',\n" +
            "  'hosts' = 'http://127.0.0.1:9200',\n" +
            "  'username' = 'root',\n" +
            "  'password' = 'root',\n" +
            "  'index' = 'airline'\n" +
            ")";


    tableEnv.executeSql(sourceSql);
    tableEnv.executeSql(sinkSql);
    System.out.println("sink success========");

    String querySql = "select * from airline";
    // 从source 中查数据
    Table table = tableEnv.sqlQuery(querySql);

    DataStream<Tuple2<Boolean, Row>> result1 = tableEnv.toRetractStream(table, Row.class);
    result1.printToErr("result1");

    // 插入语sink中
    String insertSql = "insert into es_airline select * from " + table;
    System.out.println("insertSql ======" + insertSql);

    table.executeInsert("es_airline");


    env.execute("FlinkCdcSql");

}`  

代码如上，执行的时候报错

13:48:06.478 [flink-akka.actor.default-dispatcher-6] DEBUG org.apache.flink.runtime.jobmaster.JobMaster - Archive local failure causing attempt 8acdd346ede920e569a02a1e88ee1028 to fail: java.lang.NoSuchMethodError: org.elasticsearch.client.Request.addParameters(Ljava/util/Map;)V at org.elasticsearch.client.RequestConverters.bulk(RequestConverters.java:256) at org.elasticsearch.client.RestHighLevelClient.internalPerformRequestAsync(RestHighLevelClient.java:1763) at org.elasticsearch.client.RestHighLevelClient.performRequestAsync(RestHighLevelClient.java:1732) at org.elasticsearch.client.RestHighLevelClient.performRequestAsyncAndParseEntity(RestHighLevelClient.java:1698) at org.elasticsearch.client.RestHighLevelClient.bulkAsync(RestHighLevelClient.java:549) at org.apache.flink.streaming.connectors.elasticsearch7.Elasticsearch7ApiCallBridge.lambda$createBulkProcessorBuilder$0(Elasticsearch7ApiCallBridge.java:82) at org.elasticsearch.action.bulk.Retry$RetryHandler.execute(Retry.java:204) at org.elasticsearch.action.bulk.Retry.withBackoff(Retry.java:59) at org.elasticsearch.action.bulk.BulkRequestHandler.execute(BulkRequestHandler.java:62) at org.elasticsearch.action.bulk.BulkProcessor.execute(BulkProcessor.java:454) at org.elasticsearch.action.bulk.BulkProcessor.execute(BulkProcessor.java:463) at org.elasticsearch.action.bulk.BulkProcessor.flush(BulkProcessor.java:485) at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.snapshotState(ElasticsearchSinkBase.java:344) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:118) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:99) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:89) at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:218) at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:169) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:371) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator(SubtaskCheckpointCoordinatorImpl.java:706) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures(SubtaskCheckpointCoordinatorImpl.java:627) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:590) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:312) at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$8(StreamTask.java:1089) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1073) at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointAsyncInMailbox(StreamTask.java:991) at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$triggerCheckpointAsync$7(StreamTask.java:955) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:344) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:330) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:202) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:681) at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:636) at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:620) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) at java.lang.Thread.run(Thread.java:748)

引用的依赖
`

org.apache.flink
flink-connector-elasticsearch7_2.11
1.13.2

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-json</artifactId>
        <version>1.13.2</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>7.10.2</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>7.10.2</version>
    </dependency>

`

[wasys1314]

Es的版本号不一致导致

        <!-- 排除两个依赖 这两个依赖是6版本的-->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.10.2</version>
            <exclusions>
                <exclusion>
                    <groupId>org.elasticsearch</groupId>
                    <artifactId>elasticsearch</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>elasticsearch-rest-client</artifactId>
                </exclusion>
            </exclusions>
        </dependency>


<!-- 上面排除的两个依赖 重新指定7.10.2版本的依赖 -->
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>7.10.2</version>
        </dependency>

        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>7.10.2</version>
        </dependency>