Connecting to AWS Keyspaces from AWS EMR using pyspark

Question

This question was closed Sep 06, 2021 at 02:46 AM by Erick Ramirez for the following reason: Problem is not reproducible or outdated

question

digui asked Jul 31, '20 Erick Ramirez edited May 10, '22

Connecting to AWS Keyspaces from AWS EMR using pyspark

Hello everyone, I'm still taking baby steps, both on using spark and cassandra. I'd really love if someone could shed some light here.

I'm trying to connect to a table in AWS Keyspace through a jupyter notebook running on top of my AWS EMR Cluster:

Cluster configs:

Release label:emr-5.30.1

Hadoop distribution:Amazon 2.8.5

Applications:Hive 2.3.6, Pig 0.17.0, Hue 4.6.0, Livy 0.7.0, Spark 2.4.5

%%configure -f 
{
  "jars": ["s3://test-bucket/lib/spark-cassandra-connector_2.11-2.5.1.jar"]
}

from pyspark.sql import *
from pyspark.sql.types import *

spark = SparkSession.builder \
  .appName('SparkCassandraApp')
  .config('spark.jars.packages', 'com.datastax.spark:spark-cassandra-connector_2.11-2.5.1')\
  .config('spark.cassandra.connection.host', 'cassandra.sa-east-1.amazonaws.com') \
  .config('spark.cassandra.connection.port', '9142') \
  .config('spark.cassandra.connection.ssl.enabled','true') \
  .config('spark.cassandra.connection.ssl.trustStore.password','cassandratest') \
  .config("spark.cassandra.auth.username","myusername")\
  .config("spark.cassandra.auth.password","mypassword") \
  .getOrCreate()

keyspace = 'test'
table = 'test_table'
df = (
    sqlContext.
    read.format("org.apache.spark.sql.cassandra").
    options(table=table, keyspace=keyspace).
    load()
)

As you can see from above, I'm using spark-cassandra-connector_2.11-2.5.1.jar

I'm getting the following error:

An error occurred while calling o111.load.
: java.lang.NoClassDefFoundError: com/datastax/spark/connector/util/Logging
    at java.lang.ClassLoader.defineClass1(Native Method)
    at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
    at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
    at java.net.URLClassLoader.defineClass(URLClassLoader.java:468)
    at java.net.URLClassLoader.access$100(URLClassLoader.java:74)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:369)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:363)
    at java.security.AccessController.doPrivileged(Native Method)
    at java.net.URLClassLoader.findClass(URLClassLoader.java:362)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:405)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20$$anonfun$apply$12.apply(DataSource.scala:634)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20$$anonfun$apply$12.apply(DataSource.scala:634)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20.apply(DataSource.scala:634)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20.apply(DataSource.scala:634)
    at scala.util.Try.orElse(Try.scala:84)
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:634)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:194)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:167)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    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.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassNotFoundException: com.datastax.spark.connector.util.Logging
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
    ... 32 more

Traceback (most recent call last):
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py", line 172, in load
    return self._df(self._jreader.load())
  File "/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 63, in deco
    return f(*a, **kw)
  File "/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value
    format(target_id, ".", name), value)
Py4JJavaError: An error occurred while calling o111.load.
: java.lang.NoClassDefFoundError: com/datastax/spark/connector/util/Logging
    at java.lang.ClassLoader.defineClass1(Native Method)
    at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
    at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
    at java.net.URLClassLoader.defineClass(URLClassLoader.java:468)
    at java.net.URLClassLoader.access$100(URLClassLoader.java:74)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:369)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:363)
    at java.security.AccessController.doPrivileged(Native Method)
    at java.net.URLClassLoader.findClass(URLClassLoader.java:362)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:405)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20$$anonfun$apply$12.apply(DataSource.scala:634)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20$$anonfun$apply$12.apply(DataSource.scala:634)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20.apply(DataSource.scala:634)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20.apply(DataSource.scala:634)
    at scala.util.Try.orElse(Try.scala:84)
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:634)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:194)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:167)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    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.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassNotFoundException: com.datastax.spark.connector.util.Logging
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
    ... 32 more

Am I missing something here?

Any help is appreciated! Thanks!

spark-cassandra-connector unsupported aws keyspaces

10 |1000

Attachments: Up to 8 attachments (including images) can be used with a maximum of 1.0 MiB each and 10.0 MiB total.

DataStax Enterprise is powered by the best distribution of Apache Cassandra ™

Apache, Apache Cassandra, Cassandra, Apache Tomcat, Tomcat, Apache Lucene, Lucene, Apache Solr, Apache Hadoop, Hadoop, Apache Spark, Spark, Apache TinkerPop, TinkerPop, Apache Kafka and Kafka are either registered trademarks or trademarks of the Apache Software Foundation or its subsidiaries in Canada, the United States and/or other countries.

Privacy Policy Terms of Use

Answer 1 · 2020-07-31T02:04:23Z

Erick Ramirez answered Jul 31, '20 Erick Ramirez commented Jul 31, '20

The stack trace you posted indicates an environmental issue possibly with the classpath not configured correctly. I don't think the Spark connector is the problem here.

Consequently, I would recommend getting help from AWS since you are using their stack. AWS Keyspaces is a black box and there's limited public information on what it really is but my suspicion is that it's not really Cassandra under the hood but Dynamo DB and there's a skin/engine sitting on top of Dynamo that provides a CQL API so it can pretend to be a Cassandra instance to clients.

I'll reach out to the Analytics team here at DataStax to see if they can provide clues but I think since you're using AWS, they should be your first port of call. Cheers!

2 Share

10 |1000

Attachments: Up to 8 attachments (including images) can be used with a maximum of 1.0 MiB each and 10.0 MiB total.

digui commented · Jul 31, 2020 at 01:37 PM

Thanks @jaroslaw.grabowski_50515 and @Erick Ramirez for the reply. I actually changed the version of the package, and started getting this:

'Unsupported partitioner: com.amazonaws.cassandra.DefaultPartitioner

Honestly, I wanted to get a quickstart with Keyspaces before trying to setup a cluster in our Kubernetes environment, but given that it's not really being a "quick" start, I'm gonna start looking into creating the cluster. Thanks a lot for the help though!

0 ·

Erick Ramirez ♦♦ digui commented · Jul 31, 2020 at 02:06 PM

That's because there are only 2 Cassandra partitioners supported:

Murmur3Partitioner
RandomPartitioner

The connector doesn't know anything about AWS partitioners. Cheers!

0 ·

Answer 2 · 2020-07-31T11:54:13Z

jaroslaw.grabowski_50515 answered Jul 31, '20

Hi! It seams that you place only `spark-cassandra-connector` jar on your classpath with `jars option. You need to also have the connector's dependencies on your classpath. I could either:

- use `spark.jar.packages` before driver is started (pass it with --conf ...) argument. Passing it to SparkSession builder may not work.

- or use `spark-cassandra-connector-assembly` in `jars`, it is a fat jar with all the needed dependencies

Cheers!

Share

10 |1000

Attachments: Up to 8 attachments (including images) can be used with a maximum of 1.0 MiB each and 10.0 MiB total.

question

Connecting to AWS Keyspaces from AWS EMR using pyspark

2 Answers

question details

OTHER RESOURCES

question

Connecting to AWS Keyspaces from AWS EMR using pyspark

2 Answers

question details

Related Questions

OTHER RESOURCES