对大量的数据进行一系列的数据处理后DataFrame
此时有2W个分区(170W条数据,因此每个分区数量只有几百条),此时使用parquet
命令,将会往一个hdfs文件中同时写入了大量的碎文件。
提示(省略无用信息):
WARN TaskSetManager: Lost task: org.apache.spark.SparkException: Task failed while writing rows.
WARN TaskSetManager: Lost task:org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException):
No lease on /user/xx/sample_2016/_temporary/0/_temporary/attempt_201604141035_0058_m_019029_0/part-r-19029-1b93e1fa-9284-4f2c-821a-c83795ad27c1.gz.parquet:
File does not exist. Holder DFSClient_NONMAPREDUCE_1239207978_115 does not have any open files.
原因
提示为文件操作超租期,由于多个task操作写一个文件,其中某个task完成任务后删除了临时文件引起。
该参数和dfs.datanode.max.xcievers
有关,默认为256。
dfs.datanode.max.xcievers
表示每个datanode任一时刻可以打开的文件数量上限。
解决方法
有两个解决方法,一种是修改spark代码,一种是修改hdfs参数配置。
避免太高的并发度同时写一个文件。
所以在调用write.parquet
前,先使用repartition
合并碎片分区。
因为减少了分区数,下次再读取这份数据进行处理时,减少了启动task的开销。提高同时写的上限。
在hdfs-site.xml
中修改dfs.datanode.max.xcievers
,将其设置为4096<property> <name>dfs.datanode.max.xcievers</name> <value>4096</value> </property>
需要重启dataNode生效。