スパークストリーミングドライバとアプリケーションワークファイルのクリーンアップ

Question

私はspark 2.0.2を実行しており、スパークスタンドアロンクラスタではcluster deploy-modeのストリーミングジョブを展開しています。ストリーミングジョブは正常に動作しますが、アプリケーションとドライバの標準エラーのファイルが作業ディレクトリに作成されています（SPARK_HOME）。ストリーミングは常に実行されているので、これらのファイルはサイズが大きくなり、制御方法がわかりません。

1. Apache Spark does not delete temporary directories
2. How to log using log4j to local file system inside a Spark application that runs on YARN?

ことができます：私は、彼らが正確に手で問題に関連していないにもかかわらず、次の解決方法を試してみましたが、私はまだ試したと動作しませんでした

誰も私が作成されているこれらのファイルのサイズを制限する方法を助けてください？

P.S：conf/spark-env.shに以下の行を追加し、クラスタを再起動するという解決策を試みましたが、ストリーミングアプリケーションを実行している場合は機能しませんでした。

export SPARK_WORKER_OPTS="-Dspark.worker.cleanup.enabled=true -Dspark.worker.cleanup.interval=60 -Dspark.worker.cleanup.appDataTtl=60" 

EDIT：

@YuvalItzchakov私はあなたの提案を試してみましたが、それはうまくいきませんでした。

Launch Command: "/usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java" "-cp" "/mnt/spark2.0.2/conf/:/mnt/spark2.0.2/jars/*" "-Xmx2048M" "-Dspark.eventLog.enabled=true" "-Dspark.eventLog.dir=/mnt/spark2.0.2/JobsLogs" "-Dspark.executor.memory=2g" "-Dspark.deploy.defaultCores=2" "-Dspark.io.compression.codec=snappy" "-Dspark.submit.deployMode=cluster" "-Dspark.shuffle.consolidateFiles=true" "-Dspark.shuffle.compress=true" "-Dspark.app.name=Streamingjob" "-Dspark.kryoserializer.buffer.max=128M" "-Dspark.master=spark://172.16.0.27:7077" "-Dspark.shuffle.spill.compress=true" "-Dspark.serializer=org.apache.spark.serializer.KryoSerializer" "-Dspark.cassandra.input.fetch.size_in_rows=20000" "-Dspark.executor.extraJavaOptions=-Dlog4j.configuration=file:///mnt/spark2.0.2/sparkjars/log4j.xml" "-Dspark.jars=file:/mnt/spark2.0.2/sparkjars/StreamingJob-assembly-0.1.0.jar" "-Dspark.executor.instances=10" "-Dspark.driver.extraJavaOptions=-Dlog4j.configuration=file:///mnt/spark2.0.2/sparkjars/log4j.xml" "-Dspark.driver.memory=2g" "-Dspark.rpc.askTimeout=10" "-Dspark.eventLog.compress=true" "-Dspark.executor.cores=1" "-Dspark.driver.supervise=true" "-Dspark.history.fs.logDirectory=/mnt/spark2.0.2/JobsLogs" "-Dlog4j.configuration=file:///mnt/spark2.0.2/sparkjars/log4j.xml" "org.apache.spark.deploy.worker.DriverWrapper" "spark://Worker@172.16.0.29:34475" "/mnt/spark2.0.2/work/driver-20170210124424-0001/StreamingJob-assembly-0.1.0.jar" "Streamingjob" 
======================================== 

log4j:WARN No appenders could be found for logger (org.apache.hadoop.metrics2.lib.MutableMetricsFactory). 
log4j:WARN Please initialize the log4j system properly. 
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info. 
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties 
17/02/10 12:44:26 INFO SecurityManager: Changing view acls to: cassuser 
17/02/10 12:44:26 INFO SecurityManager: Changing modify acls to: cassuser 
17/02/10 12:44:26 INFO SecurityManager: Changing view acls groups to: 
17/02/10 12:44:26 INFO SecurityManager: Changing modify acls groups to: 

そして、私のlog4j.xmlファイルは、次のようになります：ドライバーのstderrログは以下の通りである：それはいくつかのエラーを与えるとして、私はその答えであなたのxmlからこのルートタグを削除した

<?xml version="1.0" encoding="UTF-8"?> 
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd" > 
<log4j:configuration> 
    <appender name="stdout" class="org.apache.log4j.RollingFileAppender"> 
     <param name="threshold" value="TRACE"/> 
     <param name="File" value="stdout"/> 
     <param name="maxFileSize" value="1MB"/> 
     <param name="maxBackupIndex" value="10"/> 
     <layout class="org.apache.log4j.PatternLayout"> 
      <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n"/> 
     </layout> 
     <filter class="org.apache.log4j.varia.LevelRangeFilter"> 
      <param name="levelMin" value="ALL" /> 
      <param name="levelMax" value="OFF" /> 
     </filter> 
    </appender> 

    <appender name="stderr" class="org.apache.log4j.RollingFileAppender"> 
     <param name="threshold" value="WARN"/> 
     <param name="File" value="stderr"/> 
     <param name="maxFileSize" value="1MB"/> 
     <param name="maxBackupIndex" value="10"/> 
     <layout class="org.apache.log4j.PatternLayout"> 
      <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n"/> 
     </layout> 
    </appender> 
</log4j:configuration> 

注意

<root> 
    <appender-ref ref="console"/> 
</root> 

Answer 1

カスタムlog4j xmlファイルを使用できます。

まず、あなたのXMLファイルに宣言：

spark-submit \ 
--conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=file:///path/to/log4j.xml \ 
--conf "spark.driver.extraJavaOptions=-Dlog4j.configuration=file:///path/to/log4j.xml 

注：あなたはストリーミングジョブを実行するとき、あなたはextraJavaOptionsを経由して、マスターと労働者をスパークするlog4j.xmlファイルを渡す必要があり、その後

<?xml version="1.0" encoding="UTF-8"?> 
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd" > 
<log4j:configuration> 
    <appender name="stdout" class="org.apache.log4j.RollingFileAppender"> 
     <param name="threshold" value="TRACE"/> 
     <param name="File" value="stdout"/> 
     <param name="maxFileSize" value="50MB"/> 
     <param name="maxBackupIndex" value="100"/> 
     <layout class="org.apache.log4j.PatternLayout"> 
      <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n"/> 
     </layout> 
     <filter class="org.apache.log4j.varia.LevelRangeFilter"> 
      <param name="levelMin" value="ALL" /> 
      <param name="levelMax" value="OFF" /> 
     </filter> 
    </appender> 

    <appender name="stderr" class="org.apache.log4j.RollingFileAppender"> 
     <param name="threshold" value="WARN"/> 
     <param name="File" value="stderr"/> 
     <param name="maxFileSize" value="50MB"/> 
     <param name="maxBackupIndex" value="100"/> 
     <layout class="org.apache.log4j.PatternLayout"> 
      <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n"/> 
     </layout> 
    </appender> 
    <root> 
     <appender-ref ref="console"/> 
    </root> 
</log4j:configuration> 

をJARとファイルをSparkにどのように展開するかによって、マスタとワーカーノードのパスが異なる可能性があります。あなたはクラスタモードを使用していると言いましたので、手動でJARと追加のファイルをディスパッチしていると仮定しますが、クライアントモードでこれを実行している人は--filesフラグでxmlファイルも追加する必要があります。