scala.MatchError: in Dataframes

Question

У меня есть одно приложение Spark (version 1.3.1). В котором я пытаюсь преобразовать один в Dataframe, он имеет много полей с различными типами данных (Integer, String, List, Map, Double).

Но когда, я выполняю свой код.

messages.foreachRDD(new Function2<JavaRDD<Message>,Time,Void>(){ 
      @Override 
      public Void call(JavaRDD<Message> arg0, Time arg1) throws Exception { 
       SQLContext sqlContext = SparkConnection.getSqlContext(); 
       DataFrame df = sqlContext.createDataFrame(arg0, Message.class); 
       df.registerTempTable("messages"); 

Я получил эту ошибку

/06/12 17:27:40 INFO JobScheduler: Starting job streaming job 1434110260000 ms.0 from job set of time 1434110260000 ms 
15/06/12 17:27:40 ERROR JobScheduler: Error running job streaming job 1434110260000 ms.1 
scala.MatchError: interface java.util.List (of class java.lang.Class) 
    at org.apache.spark.sql.SQLContext$$anonfun$getSchema$1.apply(SQLContext.scala:1193) 
    at org.apache.spark.sql.SQLContext$$anonfun$getSchema$1.apply(SQLContext.scala:1192) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) 
    at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) 
    at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108) 
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:244) 
    at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108) 
    at org.apache.spark.sql.SQLContext.getSchema(SQLContext.scala:1192) 
    at org.apache.spark.sql.SQLContext.createDataFrame(SQLContext.scala:437) 
    at org.apache.spark.sql.SQLContext.createDataFrame(SQLContext.scala:465) 

Answer 1

Если Message имеет много различных полей, как List и точки сообщения об ошибке на ошибки в List матча, чем является это вопрос. Также, если вы посмотрите на the source code, вы можете видеть, что List не в матче.

Но помимо копаться в исходном коде это также очень четко указано в документации here under the Java tab:

Currently, Spark SQL does not support JavaBeans that contain nested or contain complex types such as Lists or Arrays.

Вы можете захотеть, чтобы переключиться в Scala, как это, кажется, поддерживается там:

Case classes can also be nested or contain complex types such as Sequences or Arrays. This RDD can be implicitly converted to a DataFrame and then be registered as a table.

Таким образом, решение заключается в использовании Scala или удалении List от вас JavaBean.

В качестве последнего средства вы можете посмотреть SQLUserDefinedType, чтобы определить, как это должно быть сохранено List, возможно, его можно взломать.

Answer 2

Я решил эту проблему, обновив версию программы Spark от 1.3.1 до 1.4.0. Теперь он работает с файлом.