スパークストリーミングHiveContext NullPointerException

Question

私はCDH 5.8.3クラスタでSpark 1.6.0を使用してSpark Streamingアプリケーションを作成しています。アプリケーションは非常にシンプルです：Kafkaから読み込み、DStream/RDDの一部の変換を行い、それをHiveテーブルに出力します。私はまた、sqlContextを使用していくつかの愚かなサンプルコードを入れようとしましたが、エラーはまだそこにあります。

私の問題は、DStreamのforeachRDDステートメントでHiveContextを使用できないことです。

私のコードは次のようになります。

val sc = new SparkContext() 
val sqlContext = new HiveContext(sc) 
val ssc = new StreamingContext(sc, Minutes(sparkBatchInterval)) 
ssc.checkpoint(CHECKPOINT_DIR) 
ssc.sparkContext.setLogLevel("WARN") 
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokersList, "auto.offset.reset" -> "smallest") 
KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set(kafkaTopic)) 
val validatedAndPersisted = dstream.transform(rdd => {...}).persist(StorageLevel.MEMORY_AND_DISK_SER) 
    val recordsToBeIngested = ... 
    recordsToBeIngested.foreachRDD(rdd=> { 
    rdd.persist(StorageLevel.MEMORY_AND_DISK) 

    val ingestCount = rdd.count 
    if(ingestCount>0) { 
    sqlContext.tables("sc4").show() //here actually I shoud have a insertInto 
    } 
} 

そして、私が取得エラーは、このいずれかになります。

Exception in thread "main" java.lang.NullPointerException 
    at org.apache.spark.sql.hive.client.ClientWrapper.conf(ClientWrapper.scala:205) 
    at org.apache.spark.sql.hive.HiveContext.hiveconf$lzycompute(HiveContext.scala:554) 
    at org.apache.spark.sql.hive.HiveContext.hiveconf(HiveContext.scala:553) 
    at org.apache.spark.sql.hive.HiveContext$$anonfun$configure$1.apply(HiveContext.scala:540) 
    at org.apache.spark.sql.hive.HiveContext$$anonfun$configure$1.apply(HiveContext.scala:539) 
    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.immutable.List.foreach(List.scala:318) 
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:244) 
    at scala.collection.AbstractTraversable.map(Traversable.scala:105) 
    at org.apache.spark.sql.hive.HiveContext.configure(HiveContext.scala:539) 
    at org.apache.spark.sql.hive.HiveContext.metadataHive$lzycompute(HiveContext.scala:252) 
    at org.apache.spark.sql.hive.HiveContext.metadataHive(HiveContext.scala:239) 
    at org.apache.spark.sql.hive.HiveContext$$anon$2.<init>(HiveContext.scala:459) 
    at org.apache.spark.sql.hive.HiveContext.catalog$lzycompute(HiveContext.scala:459) 
    at org.apache.spark.sql.hive.HiveContext.catalog(HiveContext.scala:458) 
    at org.apache.spark.sql.hive.HiveContext$$anon$3.<init>(HiveContext.scala:475) 
    at org.apache.spark.sql.hive.HiveContext.analyzer$lzycompute(HiveContext.scala:475) 
    at org.apache.spark.sql.hive.HiveContext.analyzer(HiveContext.scala:474) 
    at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:34) 
    at org.apache.spark.sql.DataFrame.<init>(DataFrame.scala:133) 
    at org.apache.spark.sql.DataFrame$.apply(DataFrame.scala:52) 
    at org.apache.spark.sql.SQLContext.tables(SQLContext.scala:855) 
    at myPackage.Ingestion$$anonfun$createStreamingContext$1.apply(Ingestion.scala:173) 
    at myPackage.Ingestion$$anonfun$createStreamingContext$1.apply(Ingestion.scala:166) 
    at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:661) 
    at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:661) 
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:50) 
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:50) 
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:50) 
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:426) 
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:49) 
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:49) 
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:49) 
    at scala.util.Try$.apply(Try.scala:161) 
    at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39) 
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:224) 
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:224) 
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:224) 
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57) 
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:223) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 
    at java.lang.Thread.run(Thread.java:745) 

は、あなたがこのエラーの理由であるか、どのようにできたことかについてどんな考えを持っていますか私が直します？

は、私が自分で答えを見つけ、 マルコ

Answer 1

、ありがとうございました。この問題は、StreamingContextの前にHiveContextを作成したためです。 StreamingContextの作成後に作成を移動することで問題は解決しました。