spark union 特别注意

今天遇到一个很诡异的问题。

表A

userid	housecode	res	ctime
u1	code1	1	1301

表B

userid	housecode	res	ctime
u2	code2	0	1302

表C

userid	name	type	time
u1	大海	0	1303

而后对表A进行处理操做

表A.createOrReplaceTempView("t1");
JavaRDD<HistoryModelExt> rdd=removeDuplicateData(t1);
t1= s.createDataFrame(rdd, HistoryModelExt.class);

而后查看t1, t1.show()

u1	code1	1	1301
..	..	..	..

数据还在，而后 B union A 而后 join C(经过userid), 理论上应该是有结果的，感受就像1+1=2 这么确定，可是还真没有数据，很是诧异。

刚开始觉得是本身程序哪里有问题，苦苦寻找，发现一切正常， 最后回到 union这个方法上。

为了看清楚来龙去脉， 我把B union A的数据打印了出来，发现了一个奇怪的事情

userid	housecode	res	ctime
u2	code2	0	1302
1301	code1	1	u1

当时一会儿就明白为何join 没有数据了， A的schema已经与B不一致了。
原来 union函数并非按照列名合并，而是按照位置合并。
可是在JavaRDD<HistoryModelExt> rdd=removeDuplicateData(t1); 这步以前仍是一致的，为何转成java对象后，schema就变了呢

查看源代码

/**
   * Applies a schema to an RDD of Java Beans.
   *
   * WARNING: Since there is no guaranteed ordering for fields in a Java Bean,
   * SELECT * queries will return the columns in an undefined order.
   *
   * @since 2.0.0
   */

 def createDataFrame(rdd: RDD[_], beanClass: Class[_]): DataFrame = {
    val attributeSeq: Seq[AttributeReference] = getSchema(beanClass)
    val className = beanClass.getName
    val rowRdd = rdd.mapPartitions { iter =>
    // BeanInfo is not serializable so we must rediscover it remotely for each partition.
      SQLContext.beansToRows(iter, Utils.classForName(className), attributeSeq)
    }
    Dataset.ofRows(self, LogicalRDD(attributeSeq, rowRdd.setName(rdd.name))(self))
  }

看注释，fields的顺序是不保证的， 原来如此。

这样你在union前乖乖的执行

t1.select("userId","houseCode","res","ctime");

这样顺序就又恢复了，大数据排查问题特别麻烦，感受是一个很大的坑，但愿能帮到后来人。