Как зарегистрировать foreachRDD в потоковом приложении?

Question

Мы работаем над приложением, использующим сообщения kafka, используя искровые потоки. Охватывая искровой блок hadoop/jarn. У меня есть свойства log4j, которые хранятся как у водителя, так и у рабочего ... но я до сих пор не вижу журнал сообщений внутри в .еЙ foreachRDD видит «старт для каждого РДА» и «конца для каждого РДА»

val broadcaseLme=sc.broadcast(lme) 
logInfo("start for each rdd: ") 
val lines: DStream[MetricTypes.InputStreamType] = myConsumer.createDefaultStream() 
      lines.foreachRDD(rdd => { 
      if ((rdd != null) && (rdd.count() > 0) && (!rdd.isEmpty())) { 
       **logInfo("filteredLines: " + rdd.count())** 
       **logInfo("start loop")** 
       rdd.foreach{x => 
       val lme = broadcastLme.value  
       lme.aParser(x).get 
        } 
       logInfo("end loop") 
      } }) 

     logInfo("end of for each rdd ") 

       lines.print(10) 

Я бег приложения на кластере с помощью этого

spark-submit --verbose --class DevMain --master yarn-cluster --deploy-mode cluster --conf "spark.driver.extraJavaOptions=-Dlog4j.configuration=log4j.p‌​roperties" --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=log4j‌​.properties" --files "hdfs://hdfs-name-node:8020/user/hadoopuser/log4j.properties‌​" hdfs://hdfs-name-node:8020/user/hadoopuser/streaming_2.10-1.‌​0.0-SNAPSHOT.jar hdfs://hdfs-name-node:8020/user/hadoopuser/enriched.properti‌​es 

Я новичок в искры, может кто-нибудь, пожалуйста, помогите, почему я не вижу сообщения журнала внутри foreachrdd, это log4j.properties

log4j.rootLogger=WARN, rolling 

log4j.appender.rolling=org.apache.log4j.RollingFileAppender 
log4j.appender.rolling.layout=org.apache.log4j.PatternLayout 
log4j.appender.rolling.layout.conversionPattern=[%p] %d %c %M - %m%n 
log4j.appender.rolling.maxFileSize=100MB 
log4j.appender.rolling.maxBackupIndex=10 
log4j.appender.rolling.file=${spark.yarn.app.container.log.dir}/titanium-spark-enriched.log 
#log4j.appender.rolling.encoding=URF-8 

log4j.logger.org.apache,spark=WARN 
log4j.logger.org.eclipse.jetty=WARN 

log4j.logger.com.x1.projectname=INFO 

#log4j.appender.console=org.apache.log4j.ConsoleAppender 
#log4j.appender.console.target=System.err 
#log4j.appender.console.layout=org.apache.log4j.PatternLayout 
#log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n 

# Settings to quiet third party logs that are too verbose 
#log4j.logger.org.spark-project.jetty=WARN 
#log4j.logger.org.spark-project.jetty.util.component.AbstractLifeCycle=ERROR 
#log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=INFO 
#log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=INFO 

# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support 
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL 
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR 

#log4j.appender.RollingAppender=org.apache.log4j.DailyRollingFileAppender 
#log4j.appender.RollingAppender.File=./logs/spark/enriched.log 
#log4j.appender.RollingAppender.DatePattern='.'yyyy-MM-dd 
#log4j.appender.RollingAppender.layout=org.apache.log4j.PatternLayout 
#log4j.appender.RollingAppender.layout.ConversionPattern=[%p] %d %c %M - %m%n 


#log4j.rootLogger=INFO, RollingAppender, console 

Answer 1

Проблема с Спарк потокового приложения, кажется, с отсутствующим start после того как вы построили свой потоковый вычисления, т.е.

ssc.start() 

Цитируя scaladoc of StreamingContext:

После создания и преобразование DStreams, потоковое вычисление может быть начато и остановлено с использованием context.start() и context.stop(), соответственно.

context.awaitTermination() позволяет текущему потоку ждать окончания контекста на stop() или путем исключения.