将DataSet数据发送到elasticsearch

Question

我正在尝试使用新的elasticsearch连接器将一些数据从DataSet发送到elasticsearch，但除了数据流结构之外我找不到任何资源：

https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html

我的数据集是行的数据集（来自sql查询），这是内容：

199947,6
199958,3
199964,2
199985,2

我创建了一个实现ElasticsearchSinkFunction的静态嵌套类：

public static class NumberOfTransactionsByBlocks implements ElasticsearchSinkFunction<Row> {

    public void process(Row element, RuntimeContext ctx, RequestIndexer indexer) {
        indexer.add(createIndexRequest(element));

    }

    public IndexRequest createIndexRequest(Row element) {
        Map<String, String> json = new HashMap<>();
        json.put("block_number", element.getField(0).toString());
        json.put("numberOfTransactions", element.getField(1).toString());

        return Requests.indexRequest()
                .index("nbOfTransactionsByBlocks")
                .type("count-transactions")
                .source(json);
    }
}

然后我的问题是我不知道如何发送我的内部类的实例...

DataSet<Row> data = tableEnv.toDataSet(sqlResult, Row.class);
List<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"));

Map<String, String> config = new HashMap<>();
config.put("bulk.flush.max.actions", "1");   // flush inserts after every event
config.put("cluster.name", "elasticsearch"); // default cluster name


data.output(new ElasticsearchSink<>(config, httpHosts, new NumberOfTransactionsByBlocks()));

我在实例化ElasticsearchSink时遇到错误，它说：

不能推断论点

但是，当我指定类型（行）时，它说：

ElasticsearchSink（java.util.Map，java.util.List，org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction，org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler，org.apache.flink.streaming。 connectors.elasticsearch6.RestClientFactory）'在'org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink'中拥有私有访问权限

难道我做错了什么 ？

Answer 1

目前有（1.6.0）four different连接器由Flink for ElasticSearch提供。

• v1.x：flink-connector-elasticsearch_2.11
• v2.x：flink-connector-elasticsearch2_2.11
• v5.x：flink-connector-elasticsearch5_2.11
• v6.x：flink-connector-elasticsearch6_2.11

确保为项目包含正确的maven依赖项。

...在org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink有私人通道

现在，从您共享的跟踪中猜测，看起来您正在使用v6.x的依赖项。看看source，它表明他们已经将构造函数移动到private并添加了Builder [commit]

所以，要添加一个ElasticsearchSink，你需要这样的东西：

data.output(
  new ElasticsearchSink.Builder<>(httpHosts, new NumberOfTransactionsByBlocks())
    .setBulkFlushMaxActions(1)
    .build());

此外，导入将是

import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;