[flink-sql-connector-elasticsearch]相关问题

[Li,Qian(DXM,PB)]

Hi,all:

我在使用Flink SQL CLI向ES6写数据的时候，任务一直执行失败，
Log日志显示没有ElasticsearchSink类，请问是什么原因造成的呢？
我是用的jar包是这个flink-sql-connector-elasticsearch6_2.11-1.11.0.jar，ES版本是6.5。
谢谢～

2020-08-25 17:19:38,245 WARN  org.apache.flink.runtime.taskmanager.Task  [] - Source: TableSourceScan(table=[[default_catalog, default_database, order_info]], fields=[id, user_id, create_time, operate_time, province_id, or
der_status, total_amount]) -> Calc(select=[user_id, province_id]) ->
Sink: Sink(table=[default_catalog.default_database.user_log_sink_6], fields=[user_id, province_id]) (1/1)
(c2d1cb4035d826036f93f6e7749d7119) switched from RUNNING to FAILED.
org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load user class: org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
ClassLoader info: URL ClassLoader:
    file: '/tmp/blobStore-b9a083f0-b8a8-46c0-a1fa-5eeff9ab3399/job_5f7e950500c23b8909dbda1ad41ef6c9/blob_p-949ecf57aaab045c3136da62fe2e2ab39f502c30-a4a8c21072eb5ee42e299ed5f2bba98e' (valid JAR)
Class not resolvable through given classloader.
        at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:288) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:471) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:393) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:459) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:393) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:155) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:453) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721) [flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546) [flink-dist_2.11-1.11.1.jar:1.11.1]
        at java.lang.Thread.run(Thread.java:745) [?:1.8.0_51]
Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
        at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[?:1.8.0_51]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[?:1.8.0_51]
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:61) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[?:1.8.0_51]
        at java.lang.Class.forName0(Native Method) ~[?:1.8.0_51]
        at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_51]
        at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:78) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
        at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1774) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) ~[?:1.8.0_51]
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) ~[?:1.8.0_51]

[taochanglian]

pom里面把这个依赖设置为provided,然后将这个jar包copy到服务器上flink的lib中，然后打包项目，在运行，试一下。

在 2020/8/25 17:38, Li,Qian(DXM,PB) 写道:

> Hi,all:
>
> 我在使用Flink SQL CLI向ES6写数据的时候，任务一直执行失败，
> Log日志显示没有ElasticsearchSink类，请问是什么原因造成的呢？
> 我是用的jar包是这个flink-sql-connector-elasticsearch6_2.11-1.11.0.jar，ES版本是6.5。
> 谢谢～
>
> 2020-08-25 17:19:38,245 WARN  org.apache.flink.runtime.taskmanager.Task  [] - Source: TableSourceScan(table=[[default_catalog, default_database, order_info]], fields=[id, user_id, create_time, operate_time, province_id, or
> der_status, total_amount]) -> Calc(select=[user_id, province_id]) ->
> Sink: Sink(table=[default_catalog.default_database.user_log_sink_6], fields=[user_id, province_id]) (1/1)
> (c2d1cb4035d826036f93f6e7749d7119) switched from RUNNING to FAILED.
> org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load user class: org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
> ClassLoader info: URL ClassLoader:
>      file: '/tmp/blobStore-b9a083f0-b8a8-46c0-a1fa-5eeff9ab3399/job_5f7e950500c23b8909dbda1ad41ef6c9/blob_p-949ecf57aaab045c3136da62fe2e2ab39f502c30-a4a8c21072eb5ee42e299ed5f2bba98e' (valid JAR)
> Class not resolvable through given classloader.
>          at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:288) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:471) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:393) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:459) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:393) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:155) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:453) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721) [flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546) [flink-dist_2.11-1.11.1.jar:1.11.1]
>          at java.lang.Thread.run(Thread.java:745) [?:1.8.0_51]
> Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
>          at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[?:1.8.0_51]
>          at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[?:1.8.0_51]
>          at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:61) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[?:1.8.0_51]
>          at java.lang.Class.forName0(Native Method) ~[?:1.8.0_51]
>          at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_51]
>          at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:78) ~[flink-dist_2.11-1.11.1.jar:1.11.1]
>          at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1774) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) ~[?:1.8.0_51]
>          at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) ~[?:1.8.0_51]
>
>
>

[Xintong Song]

Hi all,

I'd like to kindly remind you that [hidden email] is an international
mailing list. It is important to have the discussions in English, so that
every subscriber can understand. Discussions in Chinese can be posted in
the [hidden email] mailing list. Please refer to the community
information page[1] for the available mailing lists and how to subscribe.

请注意，[hidden email] 是面向全球的邮件列表。为了便于所有订阅者理解邮件的内容，请使用英文进行讨论。中文讨论可以到
[hidden email] 进行。有关邮件列表的详细信息及订阅方式，请参考社区信息页面[1]。

Thank you~

Xintong Song


[1] https://flink.apache.org/community.html#mailing-lists

On Tue, Aug 25, 2020 at 5:57 PM taochanglian <[hidden email]> wrote:

> pom里面把这个依赖设置为provided,然后将这个jar包copy到服务器上flink的lib中，然后打包项目，在运行，试一下。
>
> 在 2020/8/25 17:38, Li,Qian(DXM,PB) 写道:
> > Hi,all:
> >
> > 我在使用Flink SQL CLI向ES6写数据的时候，任务一直执行失败，
> > Log日志显示没有ElasticsearchSink类，请问是什么原因造成的呢？
> > 我是用的jar包是这个flink-sql-connector-elasticsearch6_2.11-1.11.0.jar，ES版本是6.5。
> > 谢谢～
> >
> > 2020-08-25 17:19:38,245 WARN  org.apache.flink.runtime.taskmanager.Task
> [] - Source: TableSourceScan(table=[[default_catalog, default_database,
> order_info]], fields=[id, user_id, create_time, operate_time, province_id,
> or
> > der_status, total_amount]) -> Calc(select=[user_id, province_id]) ->
> > Sink: Sink(table=[default_catalog.default_database.user_log_sink_6],
> fields=[user_id, province_id]) (1/1)
> > (c2d1cb4035d826036f93f6e7749d7119) switched from RUNNING to FAILED.
> > org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot
> load user class:
> org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
> > ClassLoader info: URL ClassLoader:
> >      file:
> '/tmp/blobStore-b9a083f0-b8a8-46c0-a1fa-5eeff9ab3399/job_5f7e950500c23b8909dbda1ad41ef6c9/blob_p-949ecf57aaab045c3136da62fe2e2ab39f502c30-a4a8c21072eb5ee42e299ed5f2bba98e'
> (valid JAR)
> > Class not resolvable through given classloader.
> >          at
> org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:288)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:471)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:393)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:459)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:393)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:155)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:453)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
> [flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
> [flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at java.lang.Thread.run(Thread.java:745) [?:1.8.0_51]
> > Caused by: java.lang.ClassNotFoundException:
> org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
> >          at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> ~[?:1.8.0_51]
> >          at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> ~[?:1.8.0_51]
> >          at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:61)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> ~[?:1.8.0_51]
> >          at java.lang.Class.forName0(Native Method) ~[?:1.8.0_51]
> >          at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_51]
> >          at
> org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:78)
> ~[flink-dist_2.11-1.11.1.jar:1.11.1]
> >          at
> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1774)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000)
> ~[?:1.8.0_51]
> >          at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)
> ~[?:1.8.0_51]
> >
> >
> >
>