当前位置: 首页 > 工具软件 > scala0.json > 使用案例 >

scala.util.parsing.json.JSON.parseFull导致的spark任务无法完成的问题

张英范
2023-12-01

在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请求等稳定可靠。

 类似资料: