开发者社区> 问答> 正文

将DataSet数据发送到elasticsearch

flink小助手 2018-12-10 13:47:32 527

正在尝试使用新的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 {

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 data = tableEnv.toDataSet(sqlResult, Row.class);
List httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"));

Map 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'中拥有私有访问权限
到底出了什么问题?

SQL JSON 流计算 数据格式
分享到
取消 提交回答
全部回答(1)
  • flink小助手
    2019-07-17 23:19:14

    目前有(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并添加了[commit]Builder

    所以,要添加一个ElasticsearchSink,你需要这样的东西:

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

    .setBulkFlushMaxActions(1)
    .build());

    此外,导入将是

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

    0 0
+ 订阅

大数据计算实践乐园,近距离学习前沿技术

推荐文章
相似问题
推荐课程