StructuredStreaming处理时参数checkpointLocation配置

问题记录：

在spark实时计算时，通过Kafka写下阅读数据mysql在中间，由于本地测试环境提交任务可以正常执行，但在生产环境中总是报告以下错误，提示文件找不到，但检查不是具体的配置文件或jar包类文件，一时摸不着头脑，后来通过错误的日志 trace找到原因。原生产环境强制要求在写流时添加参数checkpointLocation，最决最后的问题

错误信息：

Showing 4096 bytes. Click here for full log  amenode.dfs.test.net" port: 9000 file: "/home/spark/cache/sys/.sparkStaging/test_1639467581591_25244/__spark_conf__.zip" } size: 338563 timestamp: 1639532451695 type: ARCHIVE visibility: PRIVATE  =============================================================================== 21/12/15 09:41:18 INFO ApplicationMaster: Started progress reporter thread with (heartbeat : 1000, initial allocation : 200) intervals 21/12/15 09:41:25 WARN Utils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf. 21/12/15 09:41:41 WARN StreamingQueryManager: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled. 21/12/15 09:41:41 ERROR ApplicationMaster: User class threw exception: java.io.FileNotFoundException: /data12/yarn/nm-local-dir/usercache/hdp-bdview-sys/appcache/application_1639467581591_25244/container_e209_1639467581591_25244_01_000001/tmp/temporary-80a66b1b-629a-4f98-bc1e-99af39e009fd java.io.FileNotFoundException: /data12/yarn/nm-local-dir/usercache/hdp-bdview-sys/appcache/application_1639467581591_25244/container_e209_1639467581591_25244_01_000001/tmp/temporary-80a66b1b-629a-4f98-bc1e-99af39e009fd  at org.apache.hadoop.fs.viewfs.InodeTree.resolve(InodeTree.java:452)  at org.apache.hadoop.fs.viewfs.ViewFileSystem.mkdirs(ViewFileSystem.java:446)  at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:1913)  at org.apache.spark.sql.execution.streaming.StreamExecution.(StreamExecution.scala:90)  at org.apache.spark.sql.execution.streaming.MicroBatchExecution.(MicroBatchExecution.scala:48)  at org.apache.spark.sql.streaming.StreamingQueryManager.createQuery(StreamingQueryManager.scala:281)  at org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:322)  at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:297)  at com.test.demo1.KafkaToMySQL$.main(KafkaToMySQL.scala:91)  at com.test.demo1.KafkaToMySQL.main(KafkaToMySQL.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:497)  at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:685) 21/12/15 09:41:41 INFO ApplicationMaster: Final app status: FAILED, exitCode: 15, (reason: User class threw exception: java.io.FileNotFoundException: /data12/yarn/nm-local-dir/usercache/hdp-bdview-sys/appcache/application_1639467581591_25244/container_e209_1639467581591_25244_01_000001/tmp/temporary-80a66b1b-629a-4f98-bc1e-99af39e009fd  at org.apache.hadoop.fs.viewfs.InodeTree.resolve(InodeTree.java:452)  at org.apache.hadoop.fs.viewfs.ViewFileSystem.mkdirs(ViewFileSystem.java:446)  at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:1913)  at org.apache.spark.sql.execution.streaming.StreamExecution.(StreamExecution.scala:90)  at org.apache.spark.sql.execution.streaming.MicroBatchExecution.(MicroBatchExecution.scala:48)  at org.apache.spark.sql.streaming.StreamingQueryManager.createQuery(StreamingQueryManager.scala:281)  at org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:322)  at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:297)  at com.test.demo1.KafkaToMySQL$.main(KafkaToMySQL.scala:91)  at com.test.demo1.KafkaToMySQL.main(KafkaToMySQL.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:497)  at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:685) )

纠正后代码：

///需要添加的参数： //option("checkpointLocation", "./chk1")     df.as(Encoders.product[Test])       .writeStream       .option("checkpointLocation", "./chk1")       .outputMode(OutputMode.Append)       .trigger(trigger)       .foreachBatch(MySQLWriter.foreachBatchFunction(xxxxxxx))       .start()