Getting issue in using spark connector #5653

xs2tarunkukreja · 2023-07-21T06:34:02Z

xs2tarunkukreja
Jul 21, 2023

I am using Running Spark version 3.3.1 and nebula-spark-connector-3.0.0.jar. I have placed the jar file in SPARK_CLASSPATH.
But while calling simple read call, I am getting following error.

Code -
df = self.ctxt.spark_session.read.format(
"com.vesoft.nebula.connector.NebulaDataSource").option(
"type", "vertex").option(
"spaceName", self.ctxt.infra_config.space).option(
"label", "cases").option(
"returnCols", "*").option(
"metaAddress", self.ctxt.infra_config.meta_svc).option(
"partitionNumber", 1).option(
"user", self.ctxt.infra_config.username).option(
"passwd", self.ctxt.infra_config.password).load()

Error -
File "/opt/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py", line 184, in load
File "/opt/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1321, in call
File "/opt/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 190, in deco
File "/opt/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o78.load.
: java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/ReadSupport
at java.base/java.lang.ClassLoader.defineClass1(Native Method)
at java.base/java.lang.ClassLoader.defineClass(Unknown Source)
at java.base/java.security.SecureClassLoader.defineClass(Unknown Source)
at java.base/jdk.internal.loader.BuiltinClassLoader.defineClass(Unknown Source)
at java.base/jdk.internal.loader.BuiltinClassLoader.findClassOnClassPathOrNull(Unknown Source)
at java.base/jdk.internal.loader.BuiltinClassLoader.loadClassOrNull(Unknown Source)
at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source)
at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source)
at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
at org.apache.spark.sql.execution.datasources.DataSource$.$anonfun$lookupDataSource$3(DataSource.scala:661)
at scala.util.Try$.apply(Try.scala:213)
at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:661)
at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSourceV2(DataSource.scala:725)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:207)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:171)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.base/java.lang.reflect.Method.invoke(Unknown Source)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.sources.v2.ReadSupport
at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source)
at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source)
at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
... 29 more

Answered by wey-gu

Jul 27, 2023

Could you please add overwrite write mode for ableProvider implementation com.vesoft.nebula.connector.NebulaDataSource cannot be written with ErrorIfExists mode, please use Append or Overwrite modes instead.?

https://sparkbyexamples.com/spark/spark-write-modes-explained

View full answer

xs2tarunkukreja · 2023-07-21T06:35:24Z

xs2tarunkukreja
Jul 21, 2023
Author

@wey-gu - Would you please help me?

0 replies

wey-gu · 2023-07-21T06:38:04Z

wey-gu
Jul 21, 2023
Collaborator

The released version of spark connector doesn't support spark 3 yet, please use this one instead:

spark-connector3.0-3.0-snapshot

@Nicole00 shall we release a version with spark3 support ASAP?

9 replies

xs2tarunkukreja Jul 21, 2023
Author

@wey-gu - I have used it .. now I am getting following error.

Py4JJavaError: An error occurred while calling o59.load.
: java.util.NoSuchElementException: key not found: operateType
at scala.collection.MapLike.default(MapLike.scala:236)
at scala.collection.MapLike.default$(MapLike.scala:235)
at org.apache.spark.sql.catalyst.util.CaseInsensitiveMap.default(CaseInsensitiveMap.scala:30)
at scala.collection.MapLike.apply(MapLike.scala:144)
at scala.collection.MapLike.apply$(MapLike.scala:143)
at org.apache.spark.sql.catalyst.util.CaseInsensitiveMap.apply(CaseInsensitiveMap.scala:30)
at com.vesoft.nebula.connector.NebulaOptions.(NebulaOptions.scala:38)
at com.vesoft.nebula.connector.NebulaDataSource.getNebulaOptions(NebulaDataSource.scala:82)
at com.vesoft.nebula.connector.NebulaDataSource.inferSchema(NebulaDataSource.scala:51)
at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.getTableFromProvider(DataSourceV2Utils.scala:90)
at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.loadV2Source(DataSourceV2Utils.scala:140)
at org.apache.spark.sql.DataFrameReader.$anonfun$load$1(DataFrameReader.scala:209)
at scala.Option.flatMap(Option.scala:271)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:207)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:171)
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 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
at java.lang.Thread.run(Unknown Source)

xs2tarunkukreja Jul 21, 2023
Author

It is not compatibility issue. But error doesn't state the reason for failure.

wey-gu Jul 21, 2023
Collaborator

Could you ensure the jar package and the class was properly loaded in pyspark?

xs2tarunkukreja Jul 21, 2023
Author

If you check the call stack then it showing that in-between it is calling to Nebula Connecter -
at com.vesoft.nebula.connector.NebulaOptions.(NebulaOptions.scala:38)
at com.vesoft.nebula.connector.NebulaDataSource.getNebulaOptions(NebulaDataSource.scala:82)
at com.vesoft.nebula.connector.NebulaDataSource.inferSchema(NebulaDataSource.scala:51)

xs2tarunkukreja Jul 21, 2023
Author

FYI - I tried the same at 2 systems -

On my laptop where Nebula is running in docker client.
On K8S Cluster.

Getting same error.

wey-gu · 2023-07-21T14:21:12Z

wey-gu
Jul 21, 2023
Collaborator

It says

: java.util.NoSuchElementException: key not found: operateType

df = self.ctxt.spark_session.read.format(
      "com.vesoft.nebula.connector.NebulaDataSource").option(
      "type", "vertex").option(
      "spaceName", self.ctxt.infra_config.space).option(
      "label", "cases").option(
      "returnCols", "*").option(
      "metaAddress", self.ctxt.infra_config.meta_svc).option(
      "partitionNumber", 1).option(
      "user", self.ctxt.infra_config.username).option(
      "passwd", self.ctxt.infra_config.password).option(
      "operateType", "read").load()

could you try this with operateType being added?

I didn't get time to try out pyspark with spark-3, where there are slightly different options needed.

Also, for reader from metad/storaged, user/password is not needed as those are for graphd authentication.

0 replies

wey-gu · 2023-07-21T14:25:21Z

wey-gu
Jul 21, 2023
Collaborator

issue created: vesoft-inc/nebula-spark-connector#114

6 replies

wey-gu Jul 21, 2023
Collaborator

"read", see the last line of the code block I shared above :-P

xs2tarunkukreja Jul 24, 2023
Author

ohh my bad.. Thanks

wey-gu Jul 24, 2023
Collaborator

Thanks, my bad actually, I meant "it says : java.util.NoSuchElementException: key not found: operateType" but I only paste the code, thus being quite confusing, (edited with the error now)

xs2tarunkukreja Jul 24, 2023
Author

:). I have a question. We also mentioning partitionNumber. Do we really need to specify the same?

wey-gu Jul 25, 2023
Collaborator

It's the partition of spark itself(not for NebulaGraph), you could specify based on your situation, don't have to be 1 as the example :-D

xs2tarunkukreja · 2023-07-24T12:32:26Z

xs2tarunkukreja
Jul 24, 2023
Author

Hi @wey-gu,

I tried your suggestion, now getting another error. Just for your information, we are running PySpark on one Cluster and Nebula is running in another cluster.

df = self.ctxt.spark_session.read.format(

File "/opt/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py", line 184, in load
File "/opt/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1321, in call
File "/opt/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 190, in deco
File "/opt/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o77.load.
: com.facebook.thrift.transport.TTransportException: java.net.UnknownHostException: nebula-metad-1.nebula-metad-headless.nebula.svc.cluster.local
at com.facebook.thrift.transport.TSocket.open(TSocket.java:206)
at com.facebook.thrift.transport.TFramedTransport.open(TFramedTransport.java:70)
at com.vesoft.nebula.client.meta.MetaClient.getClient(MetaClient.java:151)
at com.vesoft.nebula.client.meta.MetaClient.freshClient(MetaClient.java:179)
at com.vesoft.nebula.client.meta.MetaClient.getSpace(MetaClient.java:242)
at com.vesoft.nebula.client.meta.MetaClient.getTag(MetaClient.java:307)
at com.vesoft.nebula.connector.nebula.MetaProvider.getTag(MetaProvider.scala:79)
at com.vesoft.nebula.connector.NebulaUtils$.getSchema(NebulaUtils.scala:171)
at com.vesoft.nebula.connector.NebulaDataSource.inferSchema(NebulaDataSource.scala:53)
at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.getTableFromProvider(DataSourceV2Utils.scala:90)
at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.loadV2Source(DataSourceV2Utils.scala:140)
at org.apache.spark.sql.DataFrameReader.$anonfun$load$1(DataFrameReader.scala:209)
at scala.Option.flatMap(Option.scala:271)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:207)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:171)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.base/java.lang.reflect.Method.invoke(Unknown Source)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.net.UnknownHostException: nebula-metad-1.nebula-metad-headless.nebula.svc.cluster.local
at java.base/java.net.AbstractPlainSocketImpl.connect(Unknown Source)
at java.base/java.net.SocksSocketImpl.connect(Unknown Source)
at java.base/java.net.Socket.connect(Unknown Source)
at com.facebook.thrift.transport.TSocket.open(TSocket.java:201)
... 26 more

8 replies

wey-gu Jul 25, 2023
Collaborator

For read:

There is, indeed, but the read ngql is limited to query like this only: match ()-[e:friend]-(v) return e", but this to query via graphd is not recommended in large scale of data.

https://github.com/vesoft-inc/nebula-spark-connector/pull/84/files#diff-ba7679094cb874d533374d6fa1156e1ce8d9cecf3284596b79062f6f88350a61

While for write:

The write path goes to graphd, thus pyspark outside of the namespace should be fine(without issue), there will be no limitation here.

xs2tarunkukreja Jul 25, 2023
Author

ok write we can try.

xs2tarunkukreja Jul 26, 2023
Author

@wey-gu - I am getting error - "requirement failed: Option 'metaAddress' is required" for Write operation.

df.write.format("com.vesoft.nebula.connector.NebulaDataSource").option(
"type", "vertex").option(
"spaceName", "re").option(
"label", "cases").option(
"vidPolicy", "").option(
"vertexField", "_vertexId").option(
"batch", 1).option(
"graphAddress", "localhost:50430").option(
"passwd", "magic").option(
"user", "magic").option(
"operateType", "write").save()

I have not specified the metaAddress. As you mentioned that we can do write based on graphd only.

wey-gu Jul 26, 2023
Collaborator

Sorry I meant expose an accessible meta address will be enough( not like the reader case, where it need to access storages host lists and all meta from metad, which requires being in same namespace)

I think expose the metad address would do the job.

wey-gu Jul 26, 2023
Collaborator

Also, for writing, we don't have to use spark connector( it's the only utils that requires storages layer access) we could do with normal Java client.

And if we are doing the data pipeline you could look into Nebula-exchange(it's spark based but we could run dedicated ad-hoc spark workload in same k8s ns in an event-driven fashion), too.

xs2tarunkukreja · 2023-07-26T06:24:04Z

xs2tarunkukreja
Jul 26, 2023
Author

@wey-gu - I tried write operation with graphd and metad but getting following error.

Traceback (most recent call last):
File "c:\spark\spark33\python\pyspark\sql\utils.py", line 196, in deco
raise converted from None
pyspark.sql.utils.AnalysisException: TableProvider implementation com.vesoft.nebula.connector.NebulaDataSource cannot be written with ErrorIfExists mode, please use Append or Overwrite modes instead.

df.write.format("com.vesoft.nebula.connector.NebulaDataSource").option(
    "type", "vertex").option(
    "spaceName", "sample").option(
    "label", "person").option(
    "vidPolicy", "").option(
    "vertexField", "id").option(
    "batch", 1).option(
    "graphAddress", "localhost:50430").option(
    "metaAddress", "localhost:58844").option(
    "passwd", "***").option(
    "user", "***").option(
    "operateType", "write").option(
    "writeMode", "update").save()

9 replies

xs2tarunkukreja Jul 26, 2023
Author

@wey-gu - Would you please guide me related to above 2 queries?

Why is it throwing exception for metad address in Write Call?
Write is failing with an exception.

wey-gu Jul 27, 2023
Collaborator

Could you please add overwrite write mode for ableProvider implementation com.vesoft.nebula.connector.NebulaDataSource cannot be written with ErrorIfExists mode, please use Append or Overwrite modes instead.?

https://sparkbyexamples.com/spark/spark-write-modes-explained

Answer selected by xs2tarunkukreja

xs2tarunkukreja Jul 27, 2023
Author

Thanks @wey-gu. I will try this.

In meanwhile, would you please help me to understand why write operation needs metad address if we are using graphd service?

wey-gu Jul 27, 2023
Collaborator

I think this is an implementation choice, as meta-client is the easiest way to get the schema of the graph.

Sorry I didn't have the time to try Pyspark for spark 3.x to update the docs to enable your smooth initial attempts(like add the overwrite option)

xs2tarunkukreja Jul 27, 2023
Author

Thanks @wey-gu for your support.

xs2tarunkukreja Jul 27, 2023
Author

@wey-gu - Write operation work for me.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Getting issue in using spark connector #5653

{{title}}

Replies: 6 comments 32 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

Getting issue in using spark connector #5653

xs2tarunkukreja Jul 21, 2023

Replies: 6 comments · 32 replies

xs2tarunkukreja Jul 21, 2023 Author

wey-gu Jul 21, 2023 Collaborator

xs2tarunkukreja Jul 21, 2023 Author

xs2tarunkukreja Jul 21, 2023 Author

wey-gu Jul 21, 2023 Collaborator

xs2tarunkukreja Jul 21, 2023 Author

xs2tarunkukreja Jul 21, 2023 Author

wey-gu Jul 21, 2023 Collaborator

wey-gu Jul 21, 2023 Collaborator

wey-gu Jul 21, 2023 Collaborator

xs2tarunkukreja Jul 24, 2023 Author

wey-gu Jul 24, 2023 Collaborator

xs2tarunkukreja Jul 24, 2023 Author

wey-gu Jul 25, 2023 Collaborator

xs2tarunkukreja Jul 24, 2023 Author

wey-gu Jul 25, 2023 Collaborator

xs2tarunkukreja Jul 25, 2023 Author

xs2tarunkukreja Jul 26, 2023 Author

wey-gu Jul 26, 2023 Collaborator

wey-gu Jul 26, 2023 Collaborator

xs2tarunkukreja Jul 26, 2023 Author

xs2tarunkukreja Jul 26, 2023 Author

wey-gu Jul 27, 2023 Collaborator

xs2tarunkukreja Jul 27, 2023 Author

wey-gu Jul 27, 2023 Collaborator

xs2tarunkukreja Jul 27, 2023 Author

xs2tarunkukreja Jul 27, 2023 Author

xs2tarunkukreja
Jul 21, 2023

Replies: 6 comments 32 replies

xs2tarunkukreja
Jul 21, 2023
Author

wey-gu
Jul 21, 2023
Collaborator

xs2tarunkukreja Jul 21, 2023
Author

xs2tarunkukreja Jul 21, 2023
Author

wey-gu Jul 21, 2023
Collaborator

xs2tarunkukreja Jul 21, 2023
Author

xs2tarunkukreja Jul 21, 2023
Author

wey-gu
Jul 21, 2023
Collaborator

wey-gu
Jul 21, 2023
Collaborator

wey-gu Jul 21, 2023
Collaborator

xs2tarunkukreja Jul 24, 2023
Author

wey-gu Jul 24, 2023
Collaborator

xs2tarunkukreja Jul 24, 2023
Author

wey-gu Jul 25, 2023
Collaborator

xs2tarunkukreja
Jul 24, 2023
Author

wey-gu Jul 25, 2023
Collaborator

xs2tarunkukreja Jul 25, 2023
Author

xs2tarunkukreja Jul 26, 2023
Author

wey-gu Jul 26, 2023
Collaborator

wey-gu Jul 26, 2023
Collaborator

xs2tarunkukreja
Jul 26, 2023
Author

xs2tarunkukreja Jul 26, 2023
Author

wey-gu Jul 27, 2023
Collaborator

xs2tarunkukreja Jul 27, 2023
Author

wey-gu Jul 27, 2023
Collaborator

xs2tarunkukreja Jul 27, 2023
Author

xs2tarunkukreja Jul 27, 2023
Author