Почему функция map sc.cassandraTable («test», «users»). Select («username») не может работать?

Question

После spark-cassandra-connector's demo и Installing the Cassandra/Spark OSS Stack, под свечой скорлупе, я попытался следующий фрагмент кода:

sc.stop 
val conf = new SparkConf(true) 
     .set("spark.cassandra.connection.host", "172.21.0.131") 
     .set("spark.cassandra.auth.username", "adminxx") 
     .set("spark.cassandra.auth.password", "adminxx") 
val sc = new SparkContext("172.21.0.131", "Cassandra Connector Test", conf) 
val rdd = sc.cassandraTable("test", "users").select("username") 

Многие операторы rdd могут работать нормально, такие как:

rdd.first 
rdd.count 

Но когда я использую map:

val result = rdd.map(x => 1) //just for simple 
result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[61] at map at <console>:32 

Затем я запускаю:

result.first 

я получил следующие ошибки:

15/12/11 15:09:00 WARN TaskSetManager: Lost task 0.0 in stage 31.0 (TID 104, 124.250.36.124): java.lang.ClassNotFoundException: 

$line346.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1 


Caused by: java.lang.ClassNotFoundException: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:366) 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:355) 
    at java.security.AccessController.doPrivileged(Native Method) 
    at java.net.URLClassLoader.findClass(URLClassLoader.java:354) 
    at java.lang.Class.forName0(Native Method) 
    at java.lang.Class.forName(Class.java:278) 
    at org.apache.spark.serializer.JavaDeserializationStream$$anon$1.resolveClass(JavaSerializer.scala:67) 
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612) 

Я не знаю, почему я получил такую ​​ошибку? Любые советы будут оценены!

ОБНОВЛЕНО: По ответ @RussSpitzer «с для CassandraRdd.map(row => row.getInt("id)) does not work , java.lang.ClassNotFoundException happened!, я решил эту ошибку с помощью следующих ошибок, вместо того, чтобы использовать sc.stop и создавая новый SparkContext, я начинаю spark-shell с опциями:

bin/spark-shell -conf spark.cassandra.connection.host=172.21.0.131 --conf spark.cassandra.auth.username=adminxx --conf spark.cassandra.auth.password=adminxx 

И тогда все шаги одинаковы и работают нормально.

Answer 1

Приложения Spark обычно отправляют свои скомпилированные коды в виде файлов jar исполнителям. Таким образом, у исполнителя есть map.

Ситуация более сложная в spark-shell. Он должен компилировать и транслировать код для каждой строки интерактивно. Нет даже класса, в котором вы работаете. Он создает эти поддельные классы $$iwC$$, чтобы решить эту проблему.

Обычно это хорошо работает, но, возможно, вы столкнулись с ошибкой spark-shell. Вы можете попытаться обойти его, помещая свой код внутри класса в spark-shell:

object Obj { val mapper = { x: String => 1 } } 
val result = rdd.map(Obj.mapper) 

Но это, вероятно, самый безопасный для реализации кода в качестве приложения, а не просто писать его в spark-shell.

Answer 2

Russell Spitzer's answer from the spark-connector-user list:

Я уверен, что главная проблема в том, что вы начинаете контекст с --jars, а затем убить этот контекст, а затем запустить еще один. Попробуйте упростить свой код, вместо того, чтобы устанавливать все эти параметры искрового conf и создавать новые контексты, запускайте свою оболочку. Также банкой, которую вы хотите на пути к классам, является сборка сборки соединителя, а не пользовательская сборка скрипта Scala, который вы хотите запустить.

./spark-shell --conf spark.casandra.connection.host=10.129.20.80 ...

Вы не должны изменять ack.wait.timeout или executor.extraClasspath.