spark union 特別注意

今天遇到一個很詭異的問題。

表A

表B

表C

然后對表A進(jìn)行處理操作

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

然后查看t1, t1.show()

數(shù)據(jù)還在，然后 B union A 然后 join C(通過userid), 理論上應(yīng)該是有結(jié)果的，感覺就像1+1=2 這么肯定，但是還真沒有數(shù)據(jù)，非常詫異。

剛開始以為是自己程序哪里有問題，苦苦尋找，發(fā)現(xiàn)一切正常， 最后回到 union這個方法上。

為了看清楚前因后果， 我把B union A的數(shù)據(jù)打印了出來，發(fā)現(xiàn)了一個奇怪的事情

當(dāng)時一下子就明白為什么join 沒有數(shù)據(jù)了， A的schema已經(jīng)與B不一致了。
原來 union函數(shù)并不是按照列名合并，而是按照位置合并。
但是在JavaRDD<HistoryModelExt> rdd=removeDuplicateData(t1); 這步之前還是一致的，為什么轉(zhuǎn)成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前乖乖的執(zhí)行

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

這樣順序就又恢復(fù)了，大數(shù)據(jù)排查問題特別麻煩，感覺是一個很大的坑，希望能幫到后來人。