1
私はストリーミングでSQLクエリを起動する次のコードを持っています。私の問題は、結果の1つにArrayIndexOutOfBoundsExceptionが現れた後です。なぜこれが起こるのですか?Spark SQL over Streaming - ArrayIndexOutOfBoundsException
import org.apache.spark._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.Duration
import org.apache.spark.sql.functions.udf
object StreamingSQL {
case class Persons(name: String, age: Int)
def main(args: Array[String]) {
val sparkConf = new SparkConf().setMaster("local").setAppName("HdfsWordCount")
val sc = new SparkContext(sparkConf)
// Create the context
val ssc = new StreamingContext(sc, Seconds(2))
val lines = ssc.textFileStream("/home/cloudera/Smartcare/stream/")
lines.foreachRDD(rdd=>rdd.foreach(println))
val sqc = new SQLContext(sc);
//import sqc.createSchemaRDD
import sqc.implicits._
// Create the FileInputDStream on the directory and use the
// stream to count words in new files created
lines.foreachRDD{rdd=>
val persons = rdd.map(_.split(",")).map(p => Persons(p(0), p(1).trim.toInt)).toDF()
persons.registerTempTable("data")
val teenagers = sqc.sql("SELECT name FROM data WHERE age >= 13 AND age <= 19")
teenagers.foreach(println)
}
ssc.start()
ssc.awaitTermination()
}
}
これは私が得る出力です。正しい結果の後、私はエラーをスキップ:
16/03/23 16:58:56 INFO GenerateUnsafeProjection: Code generated in 131.828141 ms
[Edgar]
16/03/23 16:58:56 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 1)
java.lang.ArrayIndexOutOfBoundsException: 1
私のTXTは次のとおりです。
Ana,31
Edgar,16
Luis,22
Noelia,26
Isabel50
Pablo,34
Laura,18
Paco,17
RDD内の実際のデータに2つのフィールドがあるかどうかを確認することから始めます。 – eliasah
RDDに2つのフィールドがあることを確認しました。 – nest
私は今テストすることはできませんが、入力データのサンプルを少なくとも提供していただきありがとうございます。 RDD全体を読み取って変換を実行すると、同じエラーが発生するかどうかを確認できますか? – eliasah