DataprocのチェックポイントからSpark Streamingジョブを再起動するには？

これで、スパークの再起動をしても「dataproc.agentを殺す」トリックは、この例外がスローされると、ジョブのストリーミング（正確にいつからかわからない）過去数週間Spark streaming on dataproc throws FileNotFoundException DataprocのチェックポイントからSpark Streamingジョブを再起動するには？

にフォローアップ：

17/05/16 17:39:02 INFO org.apache.hadoop.yarn.client.RMProxy: Connecting to ResourceManager at stream-event-processor-m/10.138.0.3:8032 
17/05/16 17:39:03 INFO org.apache.hadoop.yarn.client.api.impl.YarnClientImpl: Submitted application application_1494955637459_0006 
17/05/16 17:39:04 ERROR org.apache.spark.SparkContext: Error initializing SparkContext. 
org.apache.spark.SparkException: Yarn application has already ended! It might have been killed or unable to launch application master. 
    at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.waitForApplication(YarnClientSchedulerBackend.scala:85) 
    at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.start(YarnClientSchedulerBackend.scala:62) 
    at org.apache.spark.scheduler.TaskSchedulerImpl.start(TaskSchedulerImpl.scala:149) 
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:497) 
    at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2258) 
    at org.apache.spark.streaming.StreamingContext.<init>(StreamingContext.scala:140) 
    at org.apache.spark.streaming.StreamingContext$$anonfun$getOrCreate$1.apply(StreamingContext.scala:826) 
    at org.apache.spark.streaming.StreamingContext$$anonfun$getOrCreate$1.apply(StreamingContext.scala:826) 
    at scala.Option.map(Option.scala:146) 
    at org.apache.spark.streaming.StreamingContext$.getOrCreate(StreamingContext.scala:826) 
    at com.thumbtack.common.model.SparkStream$class.main(SparkStream.scala:73) 
    at com.thumbtack.skyfall.StreamEventProcessor$.main(StreamEventProcessor.scala:19) 
    at com.thumbtack.skyfall.StreamEventProcessor.main(StreamEventProcessor.scala) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:498) 
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:736) 
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:185) 
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210) 
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124) 
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) 
17/05/16 17:39:04 INFO org.spark_project.jetty.server.ServerConnector: Stopped [email protected]{HTTP/1.1}{0.0.0.0:4479} 
17/05/16 17:39:04 WARN org.apache.spark.scheduler.cluster.YarnSchedulerBackend$YarnSchedulerEndpoint: Attempted to request executors before the AM has registered! 
17/05/16 17:39:04 ERROR org.apache.spark.util.Utils: Uncaught exception in thread main 
java.lang.NullPointerException 
    at org.apache.spark.network.shuffle.ExternalShuffleClient.close(ExternalShuffleClient.java:152) 
    at org.apache.spark.storage.BlockManager.stop(BlockManager.scala:1360) 
    at org.apache.spark.SparkEnv.stop(SparkEnv.scala:87) 
    at org.apache.spark.SparkContext$$anonfun$stop$11.apply$mcV$sp(SparkContext.scala:1797) 
    at org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1290) 
    at org.apache.spark.SparkContext.stop(SparkContext.scala:1796) 
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:565) 
    at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2258) 
    at org.apache.spark.streaming.StreamingContext.<init>(StreamingContext.scala:140) 
    at org.apache.spark.streaming.StreamingContext$$anonfun$getOrCreate$1.apply(StreamingContext.scala:826) 
    at org.apache.spark.streaming.StreamingContext$$anonfun$getOrCreate$1.apply(StreamingContext.scala:826) 
    at scala.Option.map(Option.scala:146) 
    at org.apache.spark.streaming.StreamingContext$.getOrCreate(StreamingContext.scala:826) 
    at com.thumbtack.common.model.SparkStream$class.main(SparkStream.scala:73) 
    at com.thumbtack.skyfall.StreamEventProcessor$.main(StreamEventProcessor.scala:19) 
    at com.thumbtack.skyfall.StreamEventProcessor.main(StreamEventProcessor.scala) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:498) 
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:736) 
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:185) 
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210) 
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124) 
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) 
Exception in thread "main" org.apache.spark.SparkException: Yarn application has already ended! It might have been killed or unable to launch application master. 
    at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.waitForApplication(YarnClientSchedulerBackend.scala:85) 
    at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.start(YarnClientSchedulerBackend.scala:62) 
    at org.apache.spark.scheduler.TaskSchedulerImpl.start(TaskSchedulerImpl.scala:149) 
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:497) 
    at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2258) 
    at org.apache.spark.streaming.StreamingContext.<init>(StreamingContext.scala:140) 
    at org.apache.spark.streaming.StreamingContext$$anonfun$getOrCreate$1.apply(StreamingContext.scala:826) 
    at org.apache.spark.streaming.StreamingContext$$anonfun$getOrCreate$1.apply(StreamingContext.scala:826) 
    at scala.Option.map(Option.scala:146) 
    at org.apache.spark.streaming.StreamingContext$.getOrCreate(StreamingContext.scala:826) 
    at com.thumbtack.common.model.SparkStream$class.main(SparkStream.scala:73) 
    at com.thumbtack.skyfall.StreamEventProcessor$.main(StreamEventProcessor.scala:19) 
    at com.thumbtack.skyfall.StreamEventProcessor.main(StreamEventProcessor.scala) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:498) 
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:736) 
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:185) 
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210) 
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124) 
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) 
Job output is complete

Dataprocクラスタのチェックポイントからスパークストリーミングジョブを再開するにはどうすればいいですか？

出典

2017-05-16 Venkatesh Iyer

私たちは最近dataprocジョブ（gcloud betaトラックとv1 APIで利用可能）に自動再起動機能を追加しました。

自動再起動を利用するには、ジョブを修復/クリーンアップできる必要があります。ジョブは修正されません。ただし、はのチェックポイントファイルを使用したで動作します。。

restart-dataproc-agentトリックはもう必要ありません。自動再起動は、ジョブクラッシュ、Dataprocエージェント障害、およびVMの再起動時の移行イベントに対して回復します。

例： gcloud beta dataproc jobs submit spark ... --max-failures-per-hour 1

参照： https://cloud.google.com/dataproc/docs/concepts/restartable-jobs

あなたは回復をテストしたい場合は、マスターVMを再起動することで、VMの移行をシミュレートすることができます[1]。その後、ジョブ[2]を記述して、statusHistoryのATTEMPT_FAILUREエントリを参照できるようになります。

[1] gcloud compute instances reset <cluster-name>-m

を[2] gcloud dataproc jobs describe

出典

2017-05-16 18:12:11 tix

はご回答いただきありがとうございます。これを使って強制的に再起動するにはどうすればよいですか？ストリーミングジョブが出力操作で停止していて、まだクラッシュしていないとしますが、強制的に再起動する必要がありますか？また、この機能がベータ版からプロダクション版に昇格する予定のタイムラインは何ですか？ –

強制的に再起動するには、 'gcloud dataproc jobs kill'を使ってジョブを強制終了し、再度再実行してください。元のチェックポイントファイルを使用すると、すべてが正常に機能します（Job.yarnApplicationsは報告されません）。 – tix

ありがとうございました！ Btw、私はCLIオプション "--max-failures-per-hour 3"を使用しなければなりませんでした。 "--max_retries_per_hour"ではなく、docと書いてあります。おそらくそこの修正が必要でしょう。また、これがいつベータ版から本番版に移行するかについての考えはありますか？ –

DataprocのチェックポイントからSpark Streamingジョブを再起動するには？

答えて

関連する問題