入坑
前一阵子，强哥遇到一个需求，通过livy执行pyspark编写的sql语句，将sql的查询结果转成csv文件存入s3上。大致的代码如下：
from pyspark.sql.functions import *spark.sql("SELECT id FROM USER LIMIT 10").coalesce(1).write.option("header", "true").option("escape", "\"").csv("s3://tmp/business/10554210609/")
在开发完提测后，在测试环境遇到了类似如下报错：
org.apache.hadoop.fs.FileAlreadyExistsException: File already exists:s3://tmp/business/10554210609/part-00000-33282eac.csvat com.amazon.ws.emr.hadoop.fs.s3.upload.plan.RegularUploadPlanner.checkExistenceIfNotOverwriting(RegularUploadPlanner.java:36)at com.amazon.ws.emr.hadoop.fs.s3.upload.plan.RegularUploadPlanner.plan(RegularUploadPlanner.java:30)at com.amazon.ws.emr.hadoop.fs.s3.upload.plan.UploadPlannerChain.plan(UploadPlannerChain.java:37)at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.create(S3NativeFileSystem.java:703)at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:932)at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:913)at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:810)at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.create(EmrFileSystem.java:217)at org.apache.spark.sql.execution.datasources.CodecStreams$.createOutputStream(CodecStreams.scala:81)at org.apache.spark.sql.execution.datasources.CodecStreams$.createOutputStreamWriter(CodecStreams.scala:92)at org.apache.spark.sql.execution.datasources.csv.CsvOutputWriter.<init>(CSVFileFormat.scala:177)at org.apache.spark.sql.execution.datasources.csv.CSVFileFormat$$anon$1.newInstance(CSVFileFormat.scala:85)at org.apache.spark.sql.execution.datasources.SingleDirectoryDataWriter.newOutputWriter(FileFormatDataWriter.scala:120)at org.apache.spark.sql.execution.datasources.SingleDirectoryDataWriter.<init>(FileFormatDataWriter.scala:108)at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:236)
从报错中很容易看出，是因为FileAlreadyExistsException导致的，从错误名我们可以想到是因为文件已经存在s3上改了，所以执行该语句会因为文件存在而报错。可问题是，上面代码里，保存的文件目录：/10554210609/是根据业务逻辑动态生成的任务id，每个新任务都是新的，怎么会有重复的可能呢？
对此强哥又带着疑惑重新review了自己的业务代码，保证不会是因为并发及重复执行等业务逻辑出错导致执行了多次上面的pyspark语句使得文件多次创建而触发FileAlreadyExistsException报错。
通过Debug等方式再三确认后，强哥排除了业务代码问题，每次在执行上面语句之前s3上都是没有对应的目录和文件的，可是，一旦调用livy执行该spark语句就会报错。
业务代码没问题，那会不会就是上面那句pyspark语句有问题？由于报错原因是文件已存在无法写入，那么基于上面的语句再加上覆盖文件相关的方法会不会就没这个问题了。于是将代码修改如下：
from pyspark.sql.functions import *spark.sql("SELECT id FROM USER LIMIT 10").coalesce(1).write.mode("overwrite").option("header", "true").option("escape", "\"").csv("s3://tmp/business/10554210609/")
加入了.write.mode("overwrite")即文件覆盖模式，可是代码运行后，还是报了FileAlreadyExistsException的错误，这……
山穷水复
难不成覆盖语句这么写有问题？可是照理来说，应该没错才对，因为之前也有经常这么写过保存覆盖csv文件。而且，非常奇怪的是将相同的pyspark语句在jupyter上执行居然是能执行成功的。文件成功保存也没有报错。
那问题就来了，业务代码逻辑没错，pyspark语句在jupyter也能正常执行，那么就是说，问题出在livy上？试了几次，确实测试环境在livy上执行类似的语句就是会报同样的错误。那么，是不是就可以说，问题就出在livy上呢？而且头疼的是如果真是livy内部bug，那就更麻烦了，因为当前的其他业务都是基于livy来执行的？
由于一下子没有了头绪，只能把自己能想到的办法都去试试，于是乎，修改livy配置，创建session执行pyspark语句，失败；再修改语句再创建session执行pyspark语句，失败。来来回回搞了一天，感觉自己就像无头苍蝇一样的陷入了一个怪圈之中，这个人也烦躁了起来。
难不成就这么跪了，没办法，只能将bug挂着，第二天再搞了。毕竟有非常经典的经验之谈：“如果一个问题用一种方式花了半天解决不了，那么肯定是自己的解决方式出了问题。试着把自己抽出来，转换下别的思路没准就很快解决了”。
柳暗花明
第二天一早，放弃昨天的无脑尝试，改改思路：之前的报错信息都是通过livy执行返回的，虽然明确指明了是FileAlreadyExistsException错误，可是根据昨天将近一天的尝试，可以判断应该不是报错信息字面义：文件已存在导致无法写入的问题。那么，到底错在哪里？有没有更明确的错误堆栈信息呢？
不多犹豫，找运维同学要了当时执行语句失败的hadoop节点对应的完整日志，终于终于，强哥在一大串的日志中，找到了执行的错误完整信息：
Caused by: org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire 228 bytes of memory, got 0 at org.apache.spark.memory.MemoryConsumer.throwOom(MemoryConsumer.java:157) at org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:119) at org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.acquireNewPageIfNecessary(UnsafeExternalSorter.java:383) at org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.insertRecord(UnsafeExternalSorter.java:407) at org.apache.spark.sql.execution.UnsafeExternalRowSorter.insertRow(UnsafeExternalRowSorter.java:135) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage4.sort_addToSorter_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage4.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636) at org.apache.spark.sql.execution.aggregate.SortAggregateExec$$anonfun$doExecute$1$$anonfun$3.apply(SortAggregateExec.scala:80) at org.apache.spark.sql.execution.aggregate.SortAggregateExec$$anonfun$doExecute$1$$anonfun$3.apply(SortAggregateExec.scala:77) at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndexInternal$1$$anonfun$12.apply(RDD.scala:823) …… org.apache.hadoop.fs.FileAlreadyExistsException: File already exists:s3://tmp/business/10554210609/part-00000-33282eac.csv at com.amazon.ws.emr.hadoop.fs.s3.upload.plan.RegularUploadPlanner.checkExistenceIfNotOverwriting(RegularUploadPlanner.java:36) at com.amazon.ws.emr.hadoop.fs.s3.upload.plan.RegularUploadPlanner.plan(RegularUploadPlanner.java:30) at com.amazon.ws.emr.hadoop.fs.s3.upload.plan.UploadPlannerChain.plan(UploadPlannerChain.java:37) at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.create(S3NativeFileSystem.java:703) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:932) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:913)
可以看到，在FileAlreadyExistsException的错误信息之前，有一个Caused by报错，也就是SparkOutOfMemoryError导致的问题。
额，错误信息如此明显为什么之前没有看到呢？因为之前在业务逻辑代码中，使用livy执行语句报错后，livy把该错误信息给丢弃了，只返回了最后的FileAlreadyExistsException，导致了我一直被这个错误牵着鼻子搞了一天，让人心痛。
出坑
好吧，既然找到了真正的bug所在，后面的解决相对就简单许多。复制SparkOutOfMemoryError对应的错误信息去网上搜索了一番，找到类似的问题并得到了解决方案。
原来问题还是出在pyspark语句上，关键点就在coalesce(1)：
However, if you’re doing a drastic coalesce, e.g. to numPartitions = 1, this may result in your computation taking place on fewer nodes than you like(e.g. one node in the case of numPartitions = 1). To avoid this, you can call repartition(1) instead. This will add a shuffle step, but means the current upstream partitions will be executed in parallel(per whatever the current partitioning is).
也就是说当coalesce参数过小时（例如1），并行度不够只会在很少的节点上运行，很可能导致资源不足，报SparkOutOfMemoryError错误。而repartition(1)虽然会加入shuffle步骤，但是上游的分区会并行执行。
于是，便最终将语句改成如下：
from pyspark.sql.functions import *spark.sql("SELECT id FROM USER LIMIT 10").repartition(1).write.mode("overwrite").option("header", "true").option("escape", "\"").csv("s3://tmp/business/10554210609/")
在测试环境livy上再次执行终于顺利保存文件不报错了。问题顺利解决。
知识参考
不过，是不是coalesce就一定不如repartition，当然也不一定，具体两者的区别大家有兴趣的可以参考如下链接详细了解一下：https://www.cnblogs.com/jiangxiaoxian/p/9539760.html
最后，总结一下：解决这个问题的过程相对比较痛苦，很大原因也是强哥对大数据还不是非常熟。另外，解决问题，切不可盲目不断的去重复试错，否则很容易使自己越来越烦躁而失去解决问题的动力。遇到问题，最重要的还是能找到具体详细的日志详细，找到病根对症下药才是最有效的方式！！
![[翻译]Spark on MR3——运行 Apache Spark 的新方式](https://ata2-img.oss-cn-zhangjiakou.aliyuncs.com/neweditor/ec111cc6-9193-4936-816d-840eb458f7ed.png?x-oss-process=image/resize,h_160,m_lfit)
