将 DataSet 数据发送到 elasticsearch

Question

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

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

我的 Dataset 是行的数据集（来自 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

目前 Flink 为 ElasticSearch 提供了 (1.6.0) four different 连接器。

• 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;