将数据集数据发送到弹性搜索

Question

我正在尝试使用新的弹性搜索连接器将一些数据从数据集发送到 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( Flink 为 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。查看源代码，这表明他们已经将构造函数移动到private并添加了^{Builder[提交]}

因此，要添加ElasticsearchSink，您需要类似以下内容：

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

此外，导入将是

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