Unable to load elasticsearch class

[gvsrini]

I am trying to run this code:

            var es_reader = spark.Read()
                            .Format("org.elasticsearch.spark.sql")
                            .Option("inferSchema", "true")
                            .Option("es.read.field.as.array.include", "tags")
                            .Option("es.nodes", "myelasticsearch:9200")
                            .Option("es.net.http.auth.user", "elastic")
                            .Option("es.net.http.auth.pass", "mypass");

                           var security_df = es_reader.Load("my-index/");

[JvmBridge] java.lang.ClassNotFoundException: Failed to find data source: org.elasticsearch.spark.sql. Please find packages at http://spark.apache.org/third-party-projects.html

This same code is working in Jupyter. I have no clue what's going on.... Can someone help me pls

[imback82]

You need to pass in the jar that contains org.elasticsearch.spark.sql when you run spark-submit.

[gvsrini]

I added this jar: elasticsearch-spark-20_2.11-7.3.0.jar

and ran this command:
spark-submit --class org.apache.spark.deploy.dotnet.DotnetRunner --jars C:\Users\xxxxxx\Downloads\elasticsearch-spark-20_2.11-7.3.0.jar --master spark://MySparkServerIP:7077 bin\Debug\netcoreapp3.0\microsoft-spark-2.4.x-0.4.0.jar debug

Got this exception - Is it coming from JNI layer? :

[2019-08-20T14:08:00.6891319Z] [LAPTOP] [Error] [JvmBridge] JVM method execution failed: Nonstatic method load failed for class 8 when called with 1 arguments ([Index=1, Type=String[], Value=System.String[]], )
[2019-08-20T14:08:00.6894096Z] [LAPTOP] [Error] [JvmBridge] org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: resource must be specified for Elasticsearch resources.
        at org.elasticsearch.spark.sql.DefaultSource$$anonfun$params$1$$anonfun$apply$1.apply(DefaultSource.scala:188)
        at org.elasticsearch.spark.sql.DefaultSource$$anonfun$params$1$$anonfun$apply$1.apply(DefaultSource.scala:188)
        at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
        at scala.collection.AbstractMap.getOrElse(Map.scala:59)
        at org.elasticsearch.spark.sql.DefaultSource$$anonfun$params$1.apply(DefaultSource.scala:188)
        at org.elasticsearch.spark.sql.DefaultSource$$anonfun$params$1.apply(DefaultSource.scala:188)
        at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
        at scala.collection.AbstractMap.getOrElse(Map.scala:59)
        at org.elasticsearch.spark.sql.DefaultSource.params(DefaultSource.scala:187)
        at org.elasticsearch.spark.sql.DefaultSource.createRelation(DefaultSource.scala:97)
        at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:318)
        at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:188)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
        at java.lang.reflect.Method.invoke(Unknown Source)
        at org.apache.spark.api.dotnet.DotnetBackendHandler.handleMethodCall(DotnetBackendHandler.scala:162)
        at org.apache.spark.api.dotnet.DotnetBackendHandler.handleBackendRequest(DotnetBackendHandler.scala:102)
        at org.apache.spark.api.dotnet.DotnetBackendHandler.channelRead0(DotnetBackendHandler.scala:29)
        at org.apache.spark.api.dotnet.DotnetBackendHandler.channelRead0(DotnetBackendHandler.scala:24)
        at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
        at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
        at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:138)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
        at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
        at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
        at java.lang.Thread.run(Unknown Source)

[2019-08-20T14:08:00.8455316Z] [LAPTOP] [Exception] [JvmBridge] JVM method execution failed: Nonstatic method load failed for class 8 when called with 1 arguments ([Index=1, Type=String[], Value=System.String[]], )
   at Microsoft.Spark.Interop.Ipc.JvmBridge.CallJavaMethod(Boolean isStatic, Object classNameOrJvmObjectReference, String methodName, Object[] args) in D:\Applications\Projects\spark\src\csharp\Microsoft.Spark\Interop\Ipc\JvmBridge.cs:line 195

[GoEddie]

resource must be specified for Elasticsearch resources

https://www.elastic.co/guide/en/elasticsearch/hadoop/master/configuration.html#_required_settings

You must add the option es.resource ??

[imback82]

@gvsrini did @GoEddie's suggestion help?

[gvsrini]

Thanks for asking... I'm getting some error or the other!

Added resource option as suggested by @GoEddie (though it was not required in my Jupyter notebook) as below

            var es_reader = spark.Read()
                            .Format("org.elasticsearch.spark.sql")
                            .Option("inferSchema", "true")
                            .Option("es.read.field.as.array.include", "tags")
                            .Option("es.nodes", "myelasticsearch:9200")
                            .Option("es.net.http.auth.user", "elastic")
                            .Option("es.resource", "mytable")
                            .Option("es.net.http.auth.pass", "mypassword");

And I am now getting this exception:

[2019-08-22T06:19:00.2412900Z] [LAPTOP] [Error] [JvmBridge] java.lang.NoSuchMethodError: org.elasticsearch.spark.cfg.SparkSettingsManager.load(Lorg/apache/spark/SparkConf;)Lorg/elasticsearch/spark/cfg/SparkSettings;

I have even added the ES Hadoop jar which contains this SparkSettings class but still no luck.

spark-submit --class org.apache.spark.deploy.dotnet.DotnetRunner --jars C:\Users\xxxxx\Downloads\elasticsearch-spark-20_2.11-7.3.0.jar,C:\Users\xxxxx\Downloads\elasticsearch-hadoop-7.3.0.jar --master spark://MySparkServer:7077 bin\Debug\netcoreapp3.0\microsoft-spark-2.4.x-0.4.0.jar debug

1. My question is why do I need to even run a local Spark and spark-submit. Can't .Net just connect to my already running Spark on the server. It's only a matter of firing commands on my existing Spark server the way Jupyter notebook does?
2. If I don't run in Debug mode, can I avoid running the local spark-submit and thereby achieve what I want to do.

Am I really missing something here? This is a very critical project for me so any help will really save my day!

[mohanwd]

Hi, Folks I am also facing the similar issue. But one small difference is if i add the dependency in pyspark --packages org.elasticsearch:elasticsearch-hadoop:7.7.1
I am able to establish a connection / its trying to connect to the es cluster. Whereas with spark-submit i could not solve the class not found issue tried with hadoop / spark jars both