Spark踩坑vlog——避免使用两个字段进行外连接时产生笛卡尔积的一种办法

        日常开发中遇到了一种需要使用两个字段对两张表进行外连接的场景，但是在这种情况下，会产生笛卡尔积，造成数据量疯狂增长，于是本文中使用union all替代这种情况下的join，来解决这个问题。

        1.文章表中有“纯文字”、“图片”、“长视频”、“短视频”四类，其中，需要修正指标“阅读量”，此指标为新添字段，为了与各种类下的原有阅读量区分，取名“新阅读量”；
        2.需要添加的“新阅读量”字段只在“长视频”类别下添加，其余类别下的“新阅读量”字段值记为0。
        3.以作者为纬度进行统计，最终结果为此作者在各个类型下的文章情况；
        4.一个作者纬度下，此四种文章类型不保证全都有；
        5.若一个作者纬度下，在添加新字段前，不存在“长视频”类，则对此作者新添加一个“长视频”类别，其余指标全部置为0，新增的“新阅读量”指标值记为本次新值；

        当前有两张表，一张是原始数据表，另一张是存放“新阅读量”的表，Schema如下：

• 1.原表(author_old)
  • author_id //作者唯一编码
  • type //文章类型，值包括“纯文字”/“图片”/“长视频”/“短视频”四类
  • view_count_old //原阅读量
  • like_count //点赞量
• 2.新表(author_new)
  • author_id //作者唯一编码
  • type //文章类型，只包括“长视频”
  • view_count_new //新阅读量
• 3.结果表(author_result)
  • author_id //作者唯一编码
  • type //文章类型，值包括“纯文字”/“图片”/“长视频”/“短视频”四类
  • view_count_old //原阅读量
  • like_count //点赞量
  • view_count_new //新阅读量

其中，author_old数据量约为500万，author_new数据量约为400万。

val result_df = spark.sql(s"""
		SELECt COALESECE(T1.author_id, T2.author_id) AS author_id,
			COALESECE(T1.type, T2.type) AS type,
			COALESECE(view_count_old, 0) AS view_count_old,
			COALESECE(like_count, 0) as like_count,
			COALESECE(view_count_new, 0) AS view_count_new
		FROM author_old T1
		FULL OUTER JOIN author_new T2
		ON T1.author_id = T2.author_id 
			AND T1.type = T2.type
	""")

但是在执行时发现，由于采用了FULL OUTER JOIN并且使用两个字断进行连接，造成了笛卡尔积，使得数据量暴增，提交到YARN上运行时，总是会运行失败，查看DAG图发现，因为程序中设置了参数spark.sql.shuffle.partition=200，但是在程序运行完前199个shuffle-partition后，在最后一个shuffle-partition的执行中，耗时相当长，其shuffle-read更是达到了50G还是没有读取完毕，因此判断产生了笛卡尔积，造成数据极具膨胀，从而导致资源不够使用，频频出现磁盘溢写、节点连接错误等情况；

        阿里开发手册里规定：两表连接，不推荐使用两个字断进行连接，踩了坑可算是知道为啥了；
        此次join主要产生了笛卡尔积问题，还有join本身自带的shuffle的情况，所以我们从这两个方面入手考虑解决问题。因为此次数据量并不大，所以shuffle可以不用管，只消除笛卡尔积即可。可以考虑不使用join实现此需求，即使用union + groupByKey的方式实现此需求，实现方式见下；

val result_df = spark.sql(s"""
	SELECt author_id,
			type,
			view_count_old,
			like_count,
			view_count_new
	FROM (
		SELECt author_id,
			type,
			view_count_old,
			like_count,
			0 AS view_count_new
		FROM author_old
		UNIOn ALL
		SELECt author_id,
			type,
			0 AS view_count_old,
			0 AS like_count,
			view_count_new
		FROM author_new
		)
	""")
	.rdd
	.map(row => {
		val author_id = if( row.get(0)!=null ) row.get(0).toString else ""
		val type = if( row.get(1)!=null ) row.get(1).toString else ""
		val view_count_old = if( row.get(2)!=null ) row.get(2).toString else ""
		val like_count = if( row.get(3)!=null ) row.get(3).toString else ""
		val view_count_new = if( row.get(4)!=null ) row.get(4).toString else "" 
		( (author_id, type), (view_count_old, like_count, view_count_new) )
	})
	.groupByKey()
	.map(row => {
		val (author_id, type) = row._1
		val view_count_old = if ( row._2.exist(r=>r._1>0) ) row._2.filter(r=>r._1>0).head._1
		val like_count = if ( row._2.exist(r=>r._2>0) ) row._2.filter(r=>r._2>0).head._2
		val view_count_new = if ( row._2.exist(r=>r._3>0) ) row._2.filter(r=>r._3>0).head._3
		(author_id, type, view_count_old, like_count, view_count_new)
	}).toDF("author_id", "type", "view_count_old", "like_count", "view_count_new")

        使用这种办法，虽然在代码书写上复杂了一点，但是在运行中，成功避免了由两个字段join造成的笛卡尔积。
        完结，撒花～