StructuredStreaming处理时参数checkpointLocation配置

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()