在6台机器的CDH上处理1.5亿条数据时 ,发现经常出现大部分task完成后,只有几个task一直执行的情况,开始以为是数据发生了倾斜,经过加盐、repartition等操作都无法解决。且系统日志没有报错
后来以为可能是hive写入过慢,尝试转成文件存储到hdfs再导入外部表,发现有时仍然存在超长时间执行,且文件根本没有写入的问题。
最后分析sparkUI 当前任务执行的SQL状态,发现查询执行在某个map中应该已经完成的JSON解析位置,该位置利用scala.util.parsing.json.JSON.parseFull对redis中读取的key进行转换,代码如下:
if (jedis.exists(deviceId)) {
val res = JSON.parseFull(jedis.hgetAll(deviceId).get(deviceflagIndex)) match {
case Some(map: Map[String, String]) => map
case _ => null
}
res.map(x => flagMap.put(x._1, x._2))
}
由此对当前状态下的redis连接情况进行了查询,但是发现连接数一直是0,也就是说这步操作应该早就结束了,但是为什么还会卡在这里就非常奇怪了。
为了排除是否是此解析的问题,将JSON的解析改成阿里的fastJson包,代码如下:
if (jedis.exists(deviceId)) {
val res = com.alibaba.fastjson.JSON.parseObject(jedis.hgetAll(deviceId).get(deviceflagIndex))
res.asScala.map(x => flagMap.put(x._1, x._2.toString))
}
再次运行spark-submit,程序不会再卡主,数据可以很快的写入到Hive中了。
分析:具体原因不详,但是在spark RDD的map中读取和解析数据可能会影响到map结束后的某些stage状态,导致进入下一个状态时shuffle出现异常,因此要注意在Map中数据处理时调用的方法如JSON解析、HTTP请求等稳定可靠。