flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Martin Neumann <mneum...@sics.se>
Subject Re: unclear exception when writing to elasticsearch
Date Tue, 28 Feb 2017 17:21:43 GMT
Hej,

thanks for the fast reply.

I'm currently running things from inside my IDE so it should not be a
packaging problem. That said I added the plugin from the link provided but
I'm not sure what elastic search library is needed.

Where do I override the elastic search version? The only thing I'm
currently using is the flink-connector do I have to modify its code?

<dependency>
> <groupId>org.apache.flink</groupId>
> <artifactId>flink-connector-elasticsearch2_2.10</artifactId>
> <version>1.1.3</version>
> </dependency>


One thing I forgot to mention, I can only modify things locally packing it
into a jar. I'm stuck with stock Flink 1.1.3 for the execution since I'm
running things on top of Hopsworks.

cheers Martin

On Tue, Feb 28, 2017 at 5:42 PM, Tzu-Li (Gordon) Tai <tzulitai@apache.org>
wrote:

> Hi!
>
> This could be a Elasticsearch server / client version conflict, or that
> the uber jar of your code wasn’t built properly.
>
> For the first possible issue, we’re currently using Elasticsearch 2.3.5 to
> build the Flink Elasticsearch Connector. Could you try overriding this
> version to 2.4.1 when building your code and see if the problem remains?
>
> For the second issue, please check out https://ci.apache.org/
> projects/flink/flink-docs-release-1.3/dev/linking.html#
> packaging-dependencies-with-your-usercode-with-maven.
>
> Let me know if the problem remains after trying out the above :-)
>
> Cheers,
> Gordon
>
> On March 1, 2017 at 12:24:08 AM, Martin Neumann (mneumann@sics.se) wrote:
>
> Hej,
>
> I'm trying to write to elastic search from a streaming application and I
> get a weird error message I that I can't decipher. Hopefully, someone here
> can help me. I'm trying to run the java example
> <https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/connectors/elasticsearch2.html>
from
> the website.I doublechecked that I can reach the elastic search from the
> development machine by putting some data in with curl. Has anyone an idea
> what the problem is?
>
> *Technical info:*
> Flink 1.1.3
>
> Elasticsearch 2.4.1
>
> http://bbc2.sics.se:19208/
>> {
>> "name" : "hopsworks",
>> "cluster_name" : "hops",
>> "cluster_uuid" : "XIVrGHeaTc2nICQC85chpw",
>> "version" : {
>> "number" : "2.4.1",
>> "build_hash" : "c67dc32e24162035d18d6fe1e952c4cbcbe79d16",
>> "build_timestamp" : "2016-09-27T18:57:55Z",
>> "build_snapshot" : false,
>> "lucene_version" : "5.5.2"
>> },
>> "tagline" : "You Know, for Search"
>> }
>
>
> Changes in the code:
>
> Map<String, String> config = new HashMap<>();
>>
> // This instructs the sink to emit after every element, otherwise they
>> would be buffered
>
> config.put("bulk.flush.max.actions", "1");
>>
> config.put("cluster.name", "hops");
>>
>
>> ArrayList<InetSocketAddress> transports = new ArrayList<>();
>>
> transports.add(new InetSocketAddress(InetAddress.getByName("bbc2.sics.se")
>> , 19208));
>
>
>
> Exception:
>
> Caused by: java.lang.NoClassDefFoundError: Could not initialize class
>> org.elasticsearch.threadpool.ThreadPool
>> at org.elasticsearch.client.transport.TransportClient$
>> Builder.build(TransportClient.java:133)
>> at org.apache.flink.streaming.connectors.elasticsearch2.
>> ElasticsearchSink.open(ElasticsearchSink.java:164)
>> at org.apache.flink.api.common.functions.util.FunctionUtils.
>> openFunction(FunctionUtils.java:38)
>> at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.
>> open(AbstractUdfStreamOperator.java:91)
>> at org.apache.flink.streaming.runtime.tasks.StreamTask.
>> openAllOperators(StreamTask.java:376)
>> at org.apache.flink.streaming.runtime.tasks.StreamTask.
>> invoke(StreamTask.java:256)
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:585)
>> at java.lang.Thread.run(Thread.java:745)
>
>

Mime
View raw message