Re: Re: Re: ElasticsearchSink on DataSet

2017-05-11 Thread Flavio Pompermaier
Great! I was just thinking that, in principle, a streaming sink is an extension of a batch one. Am I wrong? This would avoid a lot of code duplication and would improve the overall maintainability.. On Thu, May 11, 2017 at 4:35 AM, wyphao.2007 wrote: > Hi Flavio, I made a PR for this : https://g

Re:Re: Re: ElasticsearchSink on DataSet

2017-05-10 Thread wyphao.2007
Hi Flavio, I made a PR for this : https://github.com/apache/flink/pull/3869 And it also support ActionRequestFailureHandler in DataSet's ElasticsearchSink Best 在2017年05月09 15时30分, "Flavio Pompermaier"写道: Just one note: I took a look at your connector and it doesn't provide any failure handl

Re: Re: ElasticsearchSink on DataSet

2017-05-09 Thread Flavio Pompermaier
Just one note: I took a look at your connector and it doesn't provide any failure handling mechanism that is very useful for us. Maybe it could worth to add ActionRequestFailureHandler as provided now by the current ES streaming connector and introduced by commit https://github.com/apache/flink/com

Re:Re: ElasticsearchSink on DataSet

2017-05-08 Thread Flavio Pompermaier
Thanks a lot for the support! On 9 May 2017 07:53, "Tzu-Li (Gordon) Tai" wrote: > Hi! > > Thanks for sharing that repo! I think that would be quite an useful > contribution to Flink for the users, if you’re up to preparing a PR for it > :) > > It also looks like you’ve adopted most of the curren

Re:Re: ElasticsearchSink on DataSet

2017-05-08 Thread Tzu-Li (Gordon) Tai
Hi! Thanks for sharing that repo! I think that would be quite an useful contribution to Flink for the users, if you’re up to preparing a PR for it :) It also looks like you’ve adopted most of the current ElasticsearchSink APIs (RequestIndexer, ElasticsearchSinkFunction, etc.) for the Elasticse

Re:Re: ElasticsearchSink on DataSet

2017-05-08 Thread wyphao.2007
Hi Flavio Maybe this is what you want: https://github.com/397090770/flink-elasticsearch2-connector, It can save Flink DataSet to elasticsearch. importscala.collection.JavaConversions._ valconfig=Map("bulk.flush.max.actions"->"1000", "cluster.name"->"elasticsearch")valhosts="www.iteblog.com"val

Re: ElasticsearchSink on DataSet

2017-05-08 Thread Tzu-Li (Gordon) Tai
Hi Flavio, I don’t think there is a bridge class for this. At the moment you’ll have to implement your own OutputFormat. The ElasticsearchSink is a SinkFunction which is part of the DataStream API, which generally speaking at the moment has no bridge or unification yet with the DataSet API. Ch

ElasticsearchSink on DataSet

2017-05-03 Thread Flavio Pompermaier
Hi to all, at the moment I have a Flink Job that generates a DataSet that I write to a File that is read by Logstash to index data on ES. I'd like to use the new ElasticsearchSink to index those JSON directly from Flink but ElasticsearchSink only works with streaming environment. Is there any brid