Re: Flink Elastic Sink

[Leonard Xu]

Hi, aj

> I was confused before as I was thinking the sink builder is called only once 
> but it gets called for every batch request, correct me if my understanding is 
> wrong. 

You’re right that sink builder should be called only once rather than every 
batch requests, could you post some code piece of using the sink?

Best,
Leonard Xu



> On Fri, May 29, 2020 at 9:08 AM Leonard Xu  > wrote:
> Hi，aj
> 
> In the implementation of ElasticsearchSink, ElasticsearchSink  won't create 
> index and only start a Elastic client for sending requests to
> the Elastic cluster. You can simply extract the index(date value in your 
> case) from your timestamp field and then put it to an IndexRequest[2],  
> ElasticsearchSink will send the IndexRequests to the Elastic cluster, Elastic 
> cluster will create corresponding index and flush the records.
> 
> BTW, If you’re using Flink SQL you can use dynamic index in Elasticsearch sql 
> connector [2], you can simply config 'connector.index' = 
> ‘myindex_{ts_field|-MM-dd}’ to achieve your goals.
> 
> Best,
> Leoanrd Xu
> [1] 
> https://github.com/apache/flink/blob/master/flink-end-to-end-tests/flink-elasticsearch7-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch7SinkExample.java#L119
>  
> 
>  
> [2] 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#elasticsearch-connector
>  
> 
> 
> 
> 
> 
>> 在 2020年5月29日，02:43，aj mailto:ajainje...@gmail.com>> 
>> 写道：
>> 
>> Hello All,
>> 
>> I am getting many events in Kafka and I have written a link job that sinks 
>> that Avro records from Kafka to S3 in parquet format. 
>> 
>> Now, I want to sink these records into elastic search. but the only 
>> challenge is that I want to sink record on time indices. Basically, In 
>> Elastic, I want to create a per day index with the date as the suffix. 
>> So in Flink stream job if I create an es sink how will I change the sink to 
>> start writing  in a new index when the first event of the day arrives
>> 
>> Thanks,
>> Anuj. 
>> 
>> 
>>  
>> 
>> 
>>  
> 
> 
> -- 
> Thanks & Regards,
> Anuj Jain
> Mob. : +91- 8588817877 
> Skype : anuj.jain07
>  
> 
> 
>  

Re: Flink Elastic Sink

[aj]

Thanks, It worked.

I was confused before as I was thinking the sink builder is called only
once but it gets called for every batch request, correct me if my
understanding is wrong.

On Fri, May 29, 2020 at 9:08 AM Leonard Xu  wrote:

> Hi，aj
>
> In the implementation of ElasticsearchSink, ElasticsearchSink  won't
> create index and only start a Elastic client for sending requests to
> the Elastic cluster. You can simply extract the index(date value in your
> case) from your timestamp field and then put it to an IndexRequest[2],
>  ElasticsearchSink will send the IndexRequests to the Elastic cluster,
> Elastic cluster will create corresponding index and flush the records.
>
> BTW, If you’re using Flink SQL you can use dynamic index in Elasticsearch
> sql connector [2], you can simply config 'connector.index' =
> ‘myindex_{ts_field|-MM-dd}’ to achieve your goals.
>
> Best,
> Leoanrd Xu
> [1]
> https://github.com/apache/flink/blob/master/flink-end-to-end-tests/flink-elasticsearch7-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch7SinkExample.java#L119
>
> [2]
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#elasticsearch-connector
>
>
>
>
> 在 2020年5月29日，02:43，aj  写道：
>
> Hello All,
>
> I am getting many events in Kafka and I have written a link job that sinks
> that Avro records from Kafka to S3 in parquet format.
>
> Now, I want to sink these records into elastic search. but the only
> challenge is that I want to sink record on time indices. Basically, In
> Elastic, I want to create a per day index with the date as the suffix.
> So in Flink stream job if I create an es sink how will I change the sink
> to start writing  in a new index when the first event of the day arrives
>
> Thanks,
> Anuj.
>
>
> 
>
>
> 
>
>
>

-- 
Thanks & Regards,
Anuj Jain
Mob. : +91- 8588817877
Skype : anuj.jain07

Re: Flink Elastic Sink

[Leonard Xu]

Hi，aj

In the implementation of ElasticsearchSink, ElasticsearchSink  won't create 
index and only start a Elastic client for sending requests to
the Elastic cluster. You can simply extract the index(date value in your case) 
from your timestamp field and then put it to an IndexRequest[2],  
ElasticsearchSink will send the IndexRequests to the Elastic cluster, Elastic 
cluster will create corresponding index and flush the records.

BTW, If you’re using Flink SQL you can use dynamic index in Elasticsearch sql 
connector [2], you can simply config 'connector.index' = 
‘myindex_{ts_field|-MM-dd}’ to achieve your goals.

Best,
Leoanrd Xu
[1] 
https://github.com/apache/flink/blob/master/flink-end-to-end-tests/flink-elasticsearch7-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch7SinkExample.java#L119
 

 
[2] 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#elasticsearch-connector
 





> 在 2020年5月29日，02:43，aj  写道：
> 
> Hello All,
> 
> I am getting many events in Kafka and I have written a link job that sinks 
> that Avro records from Kafka to S3 in parquet format. 
> 
> Now, I want to sink these records into elastic search. but the only challenge 
> is that I want to sink record on time indices. Basically, In Elastic, I want 
> to create a per day index with the date as the suffix. 
> So in Flink stream job if I create an es sink how will I change the sink to 
> start writing  in a new index when the first event of the day arrives
> 
> Thanks,
> Anuj. 
> 
> 
>  
> 
> 
>  

Re: Flink Elastic Sink

[Yangze Guo]

Hi, Anuj.

>From my understanding, you could send IndexRequest to the indexer in
`ElasticsearchSink`. It will create a document under the given index
and type. So, it seems you only need to get the timestamp and concat
the `date` to your index. Am I understanding that correctly? Or do you
want to emit only 1 record per day?

Best,
Yangze Guo

On Fri, May 29, 2020 at 2:43 AM aj  wrote:
>
> Hello All,
>
> I am getting many events in Kafka and I have written a link job that sinks 
> that Avro records from Kafka to S3 in parquet format.
>
> Now, I want to sink these records into elastic search. but the only challenge 
> is that I want to sink record on time indices. Basically, In Elastic, I want 
> to create a per day index with the date as the suffix.
> So in Flink stream job if I create an es sink how will I change the sink to 
> start writing  in a new index when the first event of the day arrives
>
> Thanks,
> Anuj.
>
>
>
>
>

Flink Elastic Sink

[aj]

Hello All,

I am getting many events in Kafka and I have written a link job that sinks
that Avro records from Kafka to S3 in parquet format.

Now, I want to sink these records into elastic search. but the only
challenge is that I want to sink record on time indices. Basically, In
Elastic, I want to create a per day index with the date as the suffix.
So in Flink stream job if I create an es sink how will I change the sink to
start writing  in a new index when the first event of the day arrives

Thanks,
Anuj.

Re: Connection leak with flink elastic Sink

[Vijay Bhaskar]

Sure, let me try out with more debug logs and get back to you

Regards
Bhaskar

On Fri, Dec 14, 2018 at 4:41 PM Tzu-Li (Gordon) Tai 
wrote:

> Hi,
>
> (Removed dev@ from the mail thread)
>
> I took a look at the logs you provided, and it seems like the sink
> operators should have been properly tear-down, and therefore closing the
> RestHighLevelClient used internally.
>
> I’m at this point not really sure what else could have caused this besides
> a bug with the Elasticsearch client itself not cleaning up properly.
> Have you tried turning on debug level for logging to see if there is
> anything suspicious?
>
> Cheers,
> Gordon
>
>
> On 13 December 2018 at 7:35:33 PM, Vijay Bhaskar (bhaskar.eba...@gmail.com)
> wrote:
>
> Hi Gordon,
> We are using flink cluster 1.6.1, elastic search connector version:
> flink-connector-elasticsearch6_2.11
> Attached the stack trace.
>
> Following are the max open file descriptor limit of theTask manager
> process and open connections to the elastic
> search cluster
>
> Regards
> Bhaskar
> * #lsof -p 62041 | wc -l*
>
> *65583*
>
> *All the connections to elastic cluster reached to:*
>
> *netstat -aln | grep 9200 | wc -l*
>
> *2333*
>
>
>
>
> On Thu, Dec 13, 2018 at 4:12 PM Tzu-Li (Gordon) Tai 
> wrote:
>
>> Hi,
>>
>> Besides the information that Chesnay requested, could you also provide a
>> stack trace of the exception that caused the job to terminate in the first
>> place?
>>
>> The Elasticsearch sink does indeed close the internally used
>> Elasticsearch client, which should in turn properly release all resources
>> [1].
>> I would like to double check whether or not the case here is that that
>> part of the code was never reached.
>>
>> Cheers,
>> Gordon
>>
>> [1]
>> https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java#L334
>>
>> On 13 December 2018 at 5:59:34 PM, Chesnay Schepler (ches...@apache.org)
>> wrote:
>>
>> Specifically which connector are you using, and which Flink version?
>>
>> On 12.12.2018 13:31, Vijay Bhaskar wrote:
>> > Hi
>> > We are using flink elastic sink which streams at the rate of 1000
>> > events/sec, as described in
>> >
>> https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html
>> .
>> > We are observing connection leak of elastic connections. After few
>> > minutes all the open connections are exceeding the process limits of
>> > the max open descriptors and Job is getting terminated. But the http
>> > connections with the elastic search server remain open forever. Am i
>> > missing any specific configuration setting to close the open
>> > connection, after serving the request?
>> > But there is no such setting is described in the above documentation
>> > of elastic sink
>> >
>> > Regards
>> > Bhaskar
>>
>>
>>

Re: Connection leak with flink elastic Sink

[Tzu-Li (Gordon) Tai]

Hi,

(Removed dev@ from the mail thread)

I took a look at the logs you provided, and it seems like the sink operators 
should have been properly tear-down, and therefore closing the 
RestHighLevelClient used internally.

I’m at this point not really sure what else could have caused this besides a 
bug with the Elasticsearch client itself not cleaning up properly.
Have you tried turning on debug level for logging to see if there is anything 
suspicious?

Cheers,
Gordon


On 13 December 2018 at 7:35:33 PM, Vijay Bhaskar (bhaskar.eba...@gmail.com) 
wrote:

Hi Gordon,
We are using flink cluster 1.6.1, elastic search connector version: 
flink-connector-elasticsearch6_2.11
Attached the stack trace. 

Following are the max open file descriptor limit of theTask manager  process 
and open connections to the elastic
search cluster

Regards
Bhaskar
#lsof -p 62041 | wc -l
65583
All the connections to elastic cluster reached to:
netstat -aln | grep 9200 | wc -l
2333



On Thu, Dec 13, 2018 at 4:12 PM Tzu-Li (Gordon) Tai  wrote:
Hi,

Besides the information that Chesnay requested, could you also provide a stack 
trace of the exception that caused the job to terminate in the first place?

The Elasticsearch sink does indeed close the internally used Elasticsearch 
client, which should in turn properly release all resources [1].
I would like to double check whether or not the case here is that that part of 
the code was never reached.

Cheers,
Gordon

[1] 
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java#L334

On 13 December 2018 at 5:59:34 PM, Chesnay Schepler (ches...@apache.org) wrote:

Specifically which connector are you using, and which Flink version?

On 12.12.2018 13:31, Vijay Bhaskar wrote:
> Hi
> We are using flink elastic sink which streams at the rate of 1000
> events/sec, as described in
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html.
> We are observing connection leak of elastic connections. After few
> minutes all the open connections are exceeding the process limits of
> the max open descriptors and Job is getting terminated. But the http
> connections with the elastic search server remain open forever. Am i
> missing any specific configuration setting to close the open
> connection, after serving the request?
> But there is no such setting is described in the above documentation
> of elastic sink
>
> Regards
> Bhaskar

Re: Connection leak with flink elastic Sink

[Vijay Bhaskar]

Hi Gordon,
We are using flink cluster 1.6.1, elastic search connector version:
flink-connector-elasticsearch6_2.11
Attached the stack trace.

Following are the max open file descriptor limit of theTask manager
process and open connections to the elastic
search cluster

Regards
Bhaskar
*#lsof -p 62041 | wc -l*

*65583*

*All the connections to elastic cluster reached to:*

*netstat -aln | grep 9200 | wc -l*

*2333*




On Thu, Dec 13, 2018 at 4:12 PM Tzu-Li (Gordon) Tai 
wrote:

> Hi,
>
> Besides the information that Chesnay requested, could you also provide a
> stack trace of the exception that caused the job to terminate in the first
> place?
>
> The Elasticsearch sink does indeed close the internally used Elasticsearch
> client, which should in turn properly release all resources [1].
> I would like to double check whether or not the case here is that that
> part of the code was never reached.
>
> Cheers,
> Gordon
>
> [1]
> https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java#L334
>
> On 13 December 2018 at 5:59:34 PM, Chesnay Schepler (ches...@apache.org)
> wrote:
>
> Specifically which connector are you using, and which Flink version?
>
> On 12.12.2018 13:31, Vijay Bhaskar wrote:
> > Hi
> > We are using flink elastic sink which streams at the rate of 1000
> > events/sec, as described in
> >
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html.
>
> > We are observing connection leak of elastic connections. After few
> > minutes all the open connections are exceeding the process limits of
> > the max open descriptors and Job is getting terminated. But the http
> > connections with the elastic search server remain open forever. Am i
> > missing any specific configuration setting to close the open
> > connection, after serving the request?
> > But there is no such setting is described in the above documentation
> > of elastic sink
> >
> > Regards
> > Bhaskar
>
>
>
2018-12-13 06:14:42,120 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - 

2018-12-13 06:14:42,122 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  Starting 
StandaloneSessionClusterEntrypoint (Version: 1.6.1, Rev:23e2636, 
Date:14.09.2018 @ 19:56:46 UTC)
2018-12-13 06:14:42,122 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  OS current 
user: root
2018-12-13 06:14:42,122 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  Current 
Hadoop/Kerberos user: 
2018-12-13 06:14:42,122 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  JVM: OpenJDK 
64-Bit Server VM - Oracle Corporation - 1.8/25.181-b13
2018-12-13 06:14:42,122 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  Maximum heap 
size: 981 MiBytes
2018-12-13 06:14:42,123 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  JAVA_HOME: 
(not set)
2018-12-13 06:14:42,123 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  No Hadoop 
Dependency available
2018-12-13 06:14:42,123 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  JVM Options:
2018-12-13 06:14:42,123 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - -Xms1024m
2018-12-13 06:14:42,123 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - -Xmx1024m
2018-12-13 06:14:42,123 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - 
-Dlog.file=/root/flink-1.6.1/log/flink-root-standalonesession-0-contrail-eng-raisa-cloudsecure-eng-raisa-flink.log
2018-12-13 06:14:42,123 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - 
-Dlog4j.configuration=file:/root/flink-1.6.1/conf/log4j.properties
2018-12-13 06:14:42,124 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - 
-Dlogback.configurationFile=file:/root/flink-1.6.1/conf/logback.xml
2018-12-13 06:14:42,124 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  Program 
Arguments:
2018-12-13 06:14:42,124 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - --configDir
2018-12-13 06:14:42,124 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - 
/root/flink-1.6.1/conf
2018-12-13 06:14:42,124 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - 
--executionMode
2018-12-13 06:14:42,124 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - cluster
2018-12-13 06:14:42,124 INFO  
org.apache.flink.runtime.entrypoint.ClusterEntrypoint -  Classpath: 
/root/flink-1.6.1/lib/fl

Re: Connection leak with flink elastic Sink

[Tzu-Li (Gordon) Tai]

Hi,

Besides the information that Chesnay requested, could you also provide a stack 
trace of the exception that caused the job to terminate in the first place?

The Elasticsearch sink does indeed close the internally used Elasticsearch 
client, which should in turn properly release all resources [1].
I would like to double check whether or not the case here is that that part of 
the code was never reached.

Cheers,
Gordon

[1] 
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java#L334

On 13 December 2018 at 5:59:34 PM, Chesnay Schepler (ches...@apache.org) wrote:

Specifically which connector are you using, and which Flink version?  

On 12.12.2018 13:31, Vijay Bhaskar wrote:  
> Hi  
> We are using flink elastic sink which streams at the rate of 1000  
> events/sec, as described in  
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html.
>   
> We are observing connection leak of elastic connections. After few  
> minutes all the open connections are exceeding the process limits of  
> the max open descriptors and Job is getting terminated. But the http  
> connections with the elastic search server remain open forever. Am i  
> missing any specific configuration setting to close the open  
> connection, after serving the request?  
> But there is no such setting is described in the above documentation  
> of elastic sink  
>  
> Regards  
> Bhaskar  

Re: Connection leak with flink elastic Sink

[Chesnay Schepler]

Specifically which connector are you using, and which Flink version?

On 12.12.2018 13:31, Vijay Bhaskar wrote:

Hi
We are using flink elastic sink which streams at the rate of 1000 
events/sec, as described in 
https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html.
We are observing connection leak of elastic connections. After few 
minutes all the open connections are exceeding the process limits of 
the max open descriptors and Job is getting terminated. But the  http 
connections with the elastic search server remain open forever. Am i 
missing any specific configuration setting to close the open 
connection, after serving the request?
But there is no such setting is described in the above documentation 
of elastic sink


Regards
Bhaskar

Re: Connection leak with flink elastic Sink

[Andrey Zagrebin]

Hi Bhaskar,

I think Gordon might help you, I am pulling him into the discussion.

Best,
Andrey

> On 12 Dec 2018, at 13:31, Vijay Bhaskar  wrote:
> 
> Hi
> We are using flink elastic sink which streams at the rate of 1000 events/sec, 
> as described in 
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html
>  
> <https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html>.
> We are observing connection leak of elastic connections. After few minutes 
> all the open connections are exceeding the process limits of the max open 
> descriptors and Job is getting terminated. But the  http connections with the 
> elastic search server remain open forever. Am i missing any specific 
> configuration setting to close the open connection, after serving the request?
> But there is no such setting is described in the above documentation of 
> elastic sink
> 
> Regards
> Bhaskar

Connection leak with flink elastic Sink

[Vijay Bhaskar]

Hi
We are using flink elastic sink which streams at the rate of 1000
events/sec, as described in
https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html
.
We are observing connection leak of elastic connections. After few minutes
all the open connections are exceeding the process limits of the max open
descriptors and Job is getting terminated. But the  http connections with
the elastic search server remain open forever. Am i missing any specific
configuration setting to close the open connection, after serving the
request?
But there is no such setting is described in the above documentation of
elastic sink

Regards
Bhaskar

Re: Flink Elastic Sink AWS ES

[arpit srivastava]

It seems AWS ES setup is hiding the nodes ip.

Then I think you can try @vinay patil's solution.

Thanks,
Arpit



On Tue, Aug 29, 2017 at 3:56 AM, ant burton  wrote:

> Hey Arpit,
>
> _cat/nodes?v=ip,port
>
>
> returns the following which I have not added the x’s they were returned on
> the response
>
> ip port
>
> x.x.x.x 9300
>
> Thanks your for you help
>
> Anthony
>
>
> On 28 Aug 2017, at 10:34, arpit srivastava  wrote:
>
> Hi Ant,
>
> Can you try this.
>
> curl -XGET 'http:///_cat/nodes?v=ip,port'
>
> This should give you ip and port
>
> On Mon, Aug 28, 2017 at 3:42 AM, ant burton  wrote:
>
>> Hi Arpit,
>>
>> The response fromm _nodes doesn’t contain an ip address in my case. Is
>> this something that you experienced?
>>
>> curl -XGET 'http:///_nodes'
>>>
>>>
>> Thanks,
>>
>>
>> On 27 Aug 2017, at 14:32, ant burton  wrote:
>>
>> Thanks! I'll check later this evening.
>>
>> On Sun, 27 Aug 2017 at 07:44, arpit srivastava 
>> wrote:
>>
>>> We also had same setup where ES cluster was behind a proxy server for
>>> which port 80 was used which redirected it to ES cluster 9200 port.
>>>
>>> For using Flink we got the actual ip address of the ES nodes and put
>>> that in ips below.
>>>
>>> transportAddresses.add(new 
>>> InetSocketAddress(InetAddress.getByName("127.0.0.1"), 
>>> 9300))transportAddresses.add(new 
>>> InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300))
>>>
>>> But this worked only because 9300 port was open on ES nodes in our setup
>>> and so accessible from our Flink cluster.​
>>>
>>> Get your node list on your ES Cluster using
>>>
>>> curl -XGET 'http:///_nodes'
>>>
>>>
>>>
>>> ​and then check whether you can telnet on that  on port 9300
>>> from your flink cluster nodes
>>>
>>> $ *telnet  9300*
>>>
>>> If this works then you can use above solution.​
>>>
>>>
>>> On Sun, Aug 27, 2017 at 4:09 AM, ant burton 
>>> wrote:
>>>
 Hi Ted,

 Changing the port from 9300 to 9200 in the example you provides causes
 the error in the my original message

 my apologies for not providing context in the form of code in my
 original message, to confirm I am using the example you provided in my
 application and have it working using port 9300 in a docker environment
 locally.

 Thanks,

 On 26 Aug 2017, at 23:24, Ted Yu  wrote:

 If port 9300 in the following example is replaced by 9200, would that
 work ?

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

 Please use Flink 1.3.1+

 On Sat, Aug 26, 2017 at 3:00 PM, ant burton 
 wrote:

> Hello,
>
> Has anybody been able to use the Flink Elasticsearch connector to sink
> data to AWS ES.
>
> I don’t believe this is possible as AWS ES only allows access to port
> 9200 (via port 80) on the master node of the ES cluster, and not port 9300
> used by the the Flink Elasticsearch connector.
>
> The error message that occurs when attempting to connect to AWS ES via
> port 80 (9200) with the Flink Elasticsearch connector is:
>
> Elasticsearch client is not connected to any Elasticsearch nodes!
>
> Could anybody confirm the above? and if possible provide an
> alternative solution?
>
> Thanks you,




>>>
>>
>
>

Re: Flink Elastic Sink AWS ES

[ant burton]

Hey Arpit,

> _cat/nodes?v=ip,port


returns the following which I have not added the x’s they were returned on the 
response

ipport
x.x.x.x 9300
Thanks your for you help

Anthony


> On 28 Aug 2017, at 10:34, arpit srivastava  wrote:
> 
> Hi Ant,
> 
> Can you try this.
> 
> curl -XGET 'http:///_cat/nodes?v=ip,port'
> 
> This should give you ip and port
> 
> On Mon, Aug 28, 2017 at 3:42 AM, ant burton  > wrote:
> Hi Arpit,
> 
> The response fromm _nodes doesn’t contain an ip address in my case. Is this 
> something that you experienced?
> 
>> curl -XGET 'http:///_nodes'
> 
> Thanks,
> 
> 
>> On 27 Aug 2017, at 14:32, ant burton > > wrote:
>> 
>> Thanks! I'll check later this evening.
>> 
>> On Sun, 27 Aug 2017 at 07:44, arpit srivastava > > wrote:
>> We also had same setup where ES cluster was behind a proxy server for which 
>> port 80 was used which redirected it to ES cluster 9200 port.
>> 
>> For using Flink we got the actual ip address of the ES nodes and put that in 
>> ips below.
>> 
>> transportAddresses.add(new 
>> InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300))
>> transportAddresses.add(new 
>> InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300))
>> But this worked only because 9300 port was open on ES nodes in our setup and 
>> so accessible from our Flink cluster.​
>> 
>> Get your node list on your ES Cluster using
>> curl -XGET 'http:///_nodes'
>> 
>> ​and then check whether you can telnet on that  on port 9300 
>> from your flink cluster nodes
>> 
>> $ telnet  9300
>> 
>> If this works then you can use above solution.​
>> 
>> 
>> On Sun, Aug 27, 2017 at 4:09 AM, ant burton > > wrote:
>> Hi Ted,
>> 
>> Changing the port from 9300 to 9200 in the example you provides causes the 
>> error in the my original message
>> 
>> my apologies for not providing context in the form of code in my original 
>> message, to confirm I am using the example you provided in my application 
>> and have it working using port 9300 in a docker environment locally. 
>> 
>> Thanks,
>> 
>>> On 26 Aug 2017, at 23:24, Ted Yu >> > wrote:
>>> 
>>> If port 9300 in the following example is replaced by 9200, would that work ?
>>> 
>>> https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/elasticsearch.html
>>>  
>>> 
>>> 
>>> Please use Flink 1.3.1+
>>> 
>>> On Sat, Aug 26, 2017 at 3:00 PM, ant burton >> > wrote:
>>> Hello,
>>> 
>>> Has anybody been able to use the Flink Elasticsearch connector to sink data 
>>> to AWS ES.
>>> 
>>> I don’t believe this is possible as AWS ES only allows access to port 9200 
>>> (via port 80) on the master node of the ES cluster, and not port 9300 used 
>>> by the the Flink Elasticsearch connector.
>>> 
>>> The error message that occurs when attempting to connect to AWS ES via port 
>>> 80 (9200) with the Flink Elasticsearch connector is:
>>> 
>>> Elasticsearch client is not connected to any Elasticsearch nodes!
>>> 
>>> Could anybody confirm the above? and if possible provide an alternative 
>>> solution?
>>> 
>>> Thanks you,
>>> 
>> 
>> 
> 
> 

Re: Flink Elastic Sink AWS ES

[arpit srivastava]

Hi Ant,

Can you try this.

curl -XGET 'http:///_cat/nodes?v=ip,port'

This should give you ip and port

On Mon, Aug 28, 2017 at 3:42 AM, ant burton  wrote:

> Hi Arpit,
>
> The response fromm _nodes doesn’t contain an ip address in my case. Is
> this something that you experienced?
>
> curl -XGET 'http:///_nodes'
>>
>>
> Thanks,
>
>
> On 27 Aug 2017, at 14:32, ant burton  wrote:
>
> Thanks! I'll check later this evening.
>
> On Sun, 27 Aug 2017 at 07:44, arpit srivastava 
> wrote:
>
>> We also had same setup where ES cluster was behind a proxy server for
>> which port 80 was used which redirected it to ES cluster 9200 port.
>>
>> For using Flink we got the actual ip address of the ES nodes and put that
>> in ips below.
>>
>> transportAddresses.add(new 
>> InetSocketAddress(InetAddress.getByName("127.0.0.1"), 
>> 9300))transportAddresses.add(new 
>> InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300))
>>
>> But this worked only because 9300 port was open on ES nodes in our setup
>> and so accessible from our Flink cluster.​
>>
>> Get your node list on your ES Cluster using
>>
>> curl -XGET 'http:///_nodes'
>>
>>
>>
>> ​and then check whether you can telnet on that  on port 9300
>> from your flink cluster nodes
>>
>> $ *telnet  9300*
>>
>> If this works then you can use above solution.​
>>
>>
>> On Sun, Aug 27, 2017 at 4:09 AM, ant burton  wrote:
>>
>>> Hi Ted,
>>>
>>> Changing the port from 9300 to 9200 in the example you provides causes
>>> the error in the my original message
>>>
>>> my apologies for not providing context in the form of code in my
>>> original message, to confirm I am using the example you provided in my
>>> application and have it working using port 9300 in a docker environment
>>> locally.
>>>
>>> Thanks,
>>>
>>> On 26 Aug 2017, at 23:24, Ted Yu  wrote:
>>>
>>> If port 9300 in the following example is replaced by 9200, would that
>>> work ?
>>>
>>> https://ci.apache.org/projects/flink/flink-docs-
>>> release-1.3/dev/connectors/elasticsearch.html
>>>
>>> Please use Flink 1.3.1+
>>>
>>> On Sat, Aug 26, 2017 at 3:00 PM, ant burton 
>>> wrote:
>>>
 Hello,

 Has anybody been able to use the Flink Elasticsearch connector to sink
 data to AWS ES.

 I don’t believe this is possible as AWS ES only allows access to port
 9200 (via port 80) on the master node of the ES cluster, and not port 9300
 used by the the Flink Elasticsearch connector.

 The error message that occurs when attempting to connect to AWS ES via
 port 80 (9200) with the Flink Elasticsearch connector is:

 Elasticsearch client is not connected to any Elasticsearch nodes!

 Could anybody confirm the above? and if possible provide an alternative
 solution?

 Thanks you,
>>>
>>>
>>>
>>>
>>
>

Re: Flink Elastic Sink AWS ES

[ant burton]

Hi Arpit,

The response fromm _nodes doesn’t contain an ip address in my case. Is this 
something that you experienced?

> curl -XGET 'http:///_nodes'

Thanks,


> On 27 Aug 2017, at 14:32, ant burton  wrote:
> 
> Thanks! I'll check later this evening.
> 
> On Sun, 27 Aug 2017 at 07:44, arpit srivastava  > wrote:
> We also had same setup where ES cluster was behind a proxy server for which 
> port 80 was used which redirected it to ES cluster 9200 port.
> 
> For using Flink we got the actual ip address of the ES nodes and put that in 
> ips below.
> 
> transportAddresses.add(new 
> InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300))
> transportAddresses.add(new 
> InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300))
> But this worked only because 9300 port was open on ES nodes in our setup and 
> so accessible from our Flink cluster.​
> 
> Get your node list on your ES Cluster using
> curl -XGET 'http:///_nodes'
> 
> ​and then check whether you can telnet on that  on port 9300 from 
> your flink cluster nodes
> 
> $ telnet  9300
> 
> If this works then you can use above solution.​
> 
> 
> On Sun, Aug 27, 2017 at 4:09 AM, ant burton  > wrote:
> Hi Ted,
> 
> Changing the port from 9300 to 9200 in the example you provides causes the 
> error in the my original message
> 
> my apologies for not providing context in the form of code in my original 
> message, to confirm I am using the example you provided in my application and 
> have it working using port 9300 in a docker environment locally. 
> 
> Thanks,
> 
>> On 26 Aug 2017, at 23:24, Ted Yu > > wrote:
>> 
>> If port 9300 in the following example is replaced by 9200, would that work ?
>> 
>> https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/elasticsearch.html
>>  
>> 
>> 
>> Please use Flink 1.3.1+
>> 
>> On Sat, Aug 26, 2017 at 3:00 PM, ant burton > > wrote:
>> Hello,
>> 
>> Has anybody been able to use the Flink Elasticsearch connector to sink data 
>> to AWS ES.
>> 
>> I don’t believe this is possible as AWS ES only allows access to port 9200 
>> (via port 80) on the master node of the ES cluster, and not port 9300 used 
>> by the the Flink Elasticsearch connector.
>> 
>> The error message that occurs when attempting to connect to AWS ES via port 
>> 80 (9200) with the Flink Elasticsearch connector is:
>> 
>> Elasticsearch client is not connected to any Elasticsearch nodes!
>> 
>> Could anybody confirm the above? and if possible provide an alternative 
>> solution?
>> 
>> Thanks you,
>> 
> 
> 

Re: Flink Elastic Sink AWS ES

[vinay patil]

Hi,

We have recently moved to AWS ES service, I am using the following code:

https://github.com/awslabs/flink-stream-processing-refarch/blob/master/flink-taxi-stream-processor/src/main/java/com/amazonaws/flink/refarch/utils/ElasticsearchJestSink.java

(Note that this is not the inbuilt Flink ESSink)

You can read this blog post:
https://aws.amazon.com/blogs/big-data/build-a-real-time-stream-processing-pipeline-with-apache-flink-on-aws/


Regards,
Vinay Patil

On Sun, Aug 27, 2017 at 7:02 PM, ant burton [via Apache Flink User Mailing
List archive.] <ml+s2336050n15173...@n4.nabble.com> wrote:

> Thanks! I'll check later this evening.
>
> On Sun, 27 Aug 2017 at 07:44, arpit srivastava <[hidden email]
> <http:///user/SendEmail.jtp?type=node=15173=0>> wrote:
>
>> We also had same setup where ES cluster was behind a proxy server for
>> which port 80 was used which redirected it to ES cluster 9200 port.
>>
>> For using Flink we got the actual ip address of the ES nodes and put that
>> in ips below.
>>
>> transportAddresses.add(new 
>> InetSocketAddress(InetAddress.getByName("127.0.0.1"), 
>> 9300))transportAddresses.add(new 
>> InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300))
>>
>> But this worked only because 9300 port was open on ES nodes in our setup
>> and so accessible from our Flink cluster.​
>>
>> Get your node list on your ES Cluster using
>>
>> curl -XGET 'http:///_nodes'
>>
>>
>>
>> ​and then check whether you can telnet on that  on port 9300
>> from your flink cluster nodes
>>
>> $ *telnet  9300*
>>
>> If this works then you can use above solution.​
>>
>>
>> On Sun, Aug 27, 2017 at 4:09 AM, ant burton <[hidden email]
>> <http:///user/SendEmail.jtp?type=node=15173=1>> wrote:
>>
>>> Hi Ted,
>>>
>>> Changing the port from 9300 to 9200 in the example you provides causes
>>> the error in the my original message
>>>
>>> my apologies for not providing context in the form of code in my
>>> original message, to confirm I am using the example you provided in my
>>> application and have it working using port 9300 in a docker environment
>>> locally.
>>>
>>> Thanks,
>>>
>>> On 26 Aug 2017, at 23:24, Ted Yu <[hidden email]
>>> <http:///user/SendEmail.jtp?type=node=15173=2>> wrote:
>>>
>>> If port 9300 in the following example is replaced by 9200, would that
>>> work ?
>>>
>>> https://ci.apache.org/projects/flink/flink-docs-
>>> release-1.3/dev/connectors/elasticsearch.html
>>>
>>> Please use Flink 1.3.1+
>>>
>>> On Sat, Aug 26, 2017 at 3:00 PM, ant burton <[hidden email]
>>> <http:///user/SendEmail.jtp?type=node=15173=3>> wrote:
>>>
>>>> Hello,
>>>>
>>>> Has anybody been able to use the Flink Elasticsearch connector to sink
>>>> data to AWS ES.
>>>>
>>>> I don’t believe this is possible as AWS ES only allows access to port
>>>> 9200 (via port 80) on the master node of the ES cluster, and not port 9300
>>>> used by the the Flink Elasticsearch connector.
>>>>
>>>> The error message that occurs when attempting to connect to AWS ES via
>>>> port 80 (9200) with the Flink Elasticsearch connector is:
>>>>
>>>> Elasticsearch client is not connected to any Elasticsearch nodes!
>>>>
>>>> Could anybody confirm the above? and if possible provide an alternative
>>>> solution?
>>>>
>>>> Thanks you,
>>>
>>>
>>>
>>>
>>
>
> --
> If you reply to this email, your message will be added to the discussion
> below:
> http://apache-flink-user-mailing-list-archive.2336050.
> n4.nabble.com/Flink-Elastic-Sink-AWS-ES-tp15162p15173.html
> To start a new topic under Apache Flink User Mailing List archive., email
> ml+s2336050n1...@n4.nabble.com
> To unsubscribe from Apache Flink User Mailing List archive., click here
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code=1=dmluYXkxOC5wYXRpbEBnbWFpbC5jb218MXwxODExMDE2NjAx>
> .
> NAML
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/template/NamlServlet.jtp?macro=macro_viewer=instant_html%21nabble%3Aemail.naml=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>
>




--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Elastic-Sink-AWS-ES-tp15162p15174.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Flink Elastic Sink AWS ES

[arpit srivastava]

We also had same setup where ES cluster was behind a proxy server for which
port 80 was used which redirected it to ES cluster 9200 port.

For using Flink we got the actual ip address of the ES nodes and put that
in ips below.

transportAddresses.add(new
InetSocketAddress(InetAddress.getByName("127.0.0.1"),
9300))transportAddresses.add(new
InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300))

But this worked only because 9300 port was open on ES nodes in our setup
and so accessible from our Flink cluster.​

Get your node list on your ES Cluster using

curl -XGET 'http:///_nodes'



​and then check whether you can telnet on that  on port 9300
from your flink cluster nodes

$ *telnet  9300*

If this works then you can use above solution.​


On Sun, Aug 27, 2017 at 4:09 AM, ant burton  wrote:

> Hi Ted,
>
> Changing the port from 9300 to 9200 in the example you provides causes the
> error in the my original message
>
> my apologies for not providing context in the form of code in my original
> message, to confirm I am using the example you provided in my application
> and have it working using port 9300 in a docker environment locally.
>
> Thanks,
>
> On 26 Aug 2017, at 23:24, Ted Yu  wrote:
>
> If port 9300 in the following example is replaced by 9200, would that work
> ?
>
> https://ci.apache.org/projects/flink/flink-docs-
> release-1.3/dev/connectors/elasticsearch.html
>
> Please use Flink 1.3.1+
>
> On Sat, Aug 26, 2017 at 3:00 PM, ant burton  wrote:
>
>> Hello,
>>
>> Has anybody been able to use the Flink Elasticsearch connector to sink
>> data to AWS ES.
>>
>> I don’t believe this is possible as AWS ES only allows access to port
>> 9200 (via port 80) on the master node of the ES cluster, and not port 9300
>> used by the the Flink Elasticsearch connector.
>>
>> The error message that occurs when attempting to connect to AWS ES via
>> port 80 (9200) with the Flink Elasticsearch connector is:
>>
>> Elasticsearch client is not connected to any Elasticsearch nodes!
>>
>> Could anybody confirm the above? and if possible provide an alternative
>> solution?
>>
>> Thanks you,
>
>
>
>

Re: Flink Elastic Sink AWS ES

[ant burton]

Hi Ted,

Changing the port from 9300 to 9200 in the example you provides causes the 
error in the my original message

my apologies for not providing context in the form of code in my original 
message, to confirm I am using the example you provided in my application and 
have it working using port 9300 in a docker environment locally. 

Thanks,

> On 26 Aug 2017, at 23:24, Ted Yu  wrote:
> 
> If port 9300 in the following example is replaced by 9200, would that work ?
> 
> https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/elasticsearch.html
>  
> 
> 
> Please use Flink 1.3.1+
> 
> On Sat, Aug 26, 2017 at 3:00 PM, ant burton  > wrote:
> Hello,
> 
> Has anybody been able to use the Flink Elasticsearch connector to sink data 
> to AWS ES.
> 
> I don’t believe this is possible as AWS ES only allows access to port 9200 
> (via port 80) on the master node of the ES cluster, and not port 9300 used by 
> the the Flink Elasticsearch connector.
> 
> The error message that occurs when attempting to connect to AWS ES via port 
> 80 (9200) with the Flink Elasticsearch connector is:
> 
> Elasticsearch client is not connected to any Elasticsearch nodes!
> 
> Could anybody confirm the above? and if possible provide an alternative 
> solution?
> 
> Thanks you,
> 

Re: Flink Elastic Sink AWS ES

[Ted Yu]

If port 9300 in the following example is replaced by 9200, would that work ?

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

Please use Flink 1.3.1+

On Sat, Aug 26, 2017 at 3:00 PM, ant burton  wrote:

> Hello,
>
> Has anybody been able to use the Flink Elasticsearch connector to sink
> data to AWS ES.
>
> I don’t believe this is possible as AWS ES only allows access to port 9200
> (via port 80) on the master node of the ES cluster, and not port 9300 used
> by the the Flink Elasticsearch connector.
>
> The error message that occurs when attempting to connect to AWS ES via
> port 80 (9200) with the Flink Elasticsearch connector is:
>
> Elasticsearch client is not connected to any Elasticsearch nodes!
>
> Could anybody confirm the above? and if possible provide an alternative
> solution?
>
> Thanks you,

Flink Elastic Sink AWS ES

[ant burton]

Hello,

Has anybody been able to use the Flink Elasticsearch connector to sink data to 
AWS ES.

I don’t believe this is possible as AWS ES only allows access to port 9200 (via 
port 80) on the master node of the ES cluster, and not port 9300 used by the 
the Flink Elasticsearch connector.

The error message that occurs when attempting to connect to AWS ES via port 80 
(9200) with the Flink Elasticsearch connector is:

Elasticsearch client is not connected to any Elasticsearch nodes!

Could anybody confirm the above? and if possible provide an alternative 
solution?

Thanks you,