Я довольно новый, чтобы искриться, и я пытаюсь получить DStream, структурированный как json из темы kafka, и я хочу разобрать содержание каждого json. Я получаю JSON-то вроде этого:Parsing json in spark-streaming
{"type":"position","ident":"IBE32JZ","air_ground":"A","alt":"34000","clock":"1409733420","id":"IBE32JZ-1409715361-ed-0002:0","gs":"446","heading":"71","lat":"44.50987","lon":"2.98972","reg":"ECJRE","squawk":"1004","updateType":"A","altChange":" "}
Я пытаюсь извлечь поле только идентификационное, по крайней мере, на данный момент, и я использую библиотеку подъемной JSon для анализа де данных. Моя программа выглядит следующим образом:
, но он бросает мне исключение ниже:
java.lang.NoClassDefFoundError: scala/reflect/ClassManifest
at net.liftweb.json.JsonAST$JValue.extract(JsonAST.scala:300)
at aero.catec.stratio.ScalaExample$.parser(ScalaExample.scala:33)
at aero.catec.stratio.ScalaExample$$anonfun$2.apply(ScalaExample.scala:48)
at aero.catec.stratio.ScalaExample$$anonfun$2.apply(ScalaExample.scala:48)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:312)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
at scala.collection.AbstractIterator.to(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
at org.apache.spark.rdd.RDD$$anonfun$28.apply(RDD.scala:1003)
at org.apache.spark.rdd.RDD$$anonfun$28.apply(RDD.scala:1003)
at org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1083)
at org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1083)
at org.apache.spark.scheduler.DAGScheduler.runLocallyWithinThread(DAGScheduler.scala:575)
at org.apache.spark.scheduler.DAGScheduler$$anon$1.run(DAGScheduler.scala:560)
Caused by: java.lang.ClassNotFoundException: scala.reflect.ClassManifest
at java.net.URLClassLoader$1.run(URLClassLoader.java:372)
at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:360)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
Дело в том, что если запустить тот же без использования искры (чтение из файла) он отлично работает. Проблема начинается, когда я пытаюсь поместить ее в искровую программу. Кроме того, если я могу изменить функцию анализатора примерно так:
def parser(json: String): JValue = {
val parsedJson = parse(json)
return (parsedJson \\ "ident")
}
Он также работает. Но когда я пытаюсь извлечь фактическую строку, я получаю ту же ошибку.
Благодарим за помощь. Надеюсь, я хорошо объяснил это.
Это, вероятно, несоответствие в версии scala, которую вы используете. –
Могу ли я предположить, что «paso1.extract [PlaneInfo]» следует разобрать Json.extract [PlaneInfo]? – Gillespie