SparkのDataFrameを合体させる

# 動機
dataframeをunionするとき、カラムのスキーマが一致していないとできない。あとからテーブルにカラムが追加されてしまうと、新しいテーブルと古いテーブルをunionできなくなってしまう。

#  解決策

古いテーブルには強制的にnullのカラムを追加する。

```python
from pyspark.sql import SparkSession
import pyspark.sql.types as T
import pyspark.sql.functions as F
```


```python
data = [("Tom",21,175.3),("Mary",30,165.3)]
schema = T.StructType([
        T.StructField("name", T.StringType(), True),
        T.StructField("age", T.IntegerType(), True),
        T.StructField("hight", T.DoubleType(), True)
    ])
df = spark.createDataFrame(data,schema)
df.show()
```

    +----+---+-----+
    |name|age|hight|
    +----+---+-----+
    | Tom| 21|175.3|
    |Mary| 30|165.3|
    +----+---+-----+
    



```python
data2 = [("Bob",23,165.3,60.2),("Jack",30,165.3,82.4),("Judy",22,172.3,63.4)]
schema2 = T.StructType([
        T.StructField("name", T.StringType(), True),
        T.StructField("age", T.IntegerType(), True),
        T.StructField("hight", T.DoubleType(), True),
        T.StructField("weight", T.DoubleType(), True)
    ])
df2 = spark.createDataFrame(data2,schema2)
df2.show()
```

    +----+---+-----+------+
    |name|age|hight|weight|
    +----+---+-----+------+
    | Bob| 23|165.3|  60.2|
    |Jack| 30|165.3|  82.4|
    |Judy| 22|172.3|  63.4|
    +----+---+-----+------+
    



```python
df2.union(df)#エラー！！
```


    ---------------------------------------------------------------------------

    Py4JJavaError                             Traceback (most recent call last)

    ~/spark-2.3.0-bin-hadoop2.7/python/pyspark/sql/utils.py in deco(*a, **kw)
         62         try:
    ---> 63             return f(*a, **kw)
         64         except py4j.protocol.Py4JJavaError as e:


    ~/spark-2.3.0-bin-hadoop2.7/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
        319                     "An error occurred while calling {0}{1}{2}.\n".
    --> 320                     format(target_id, ".", name), value)
        321             else:


    Py4JJavaError: An error occurred while calling o117.union.
    : org.apache.spark.sql.AnalysisException: Union can only be performed on tables with the same number of columns, but the first table has 4 columns and the second table has 3 columns;;
    'Union
    :- AnalysisBarrier
    :     +- LogicalRDD [name#107, age#108, hight#109, weight#110], false
    +- AnalysisBarrier
          +- LogicalRDD [name#101, age#102, hight#103], false
    
    	at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.failAnalysis(CheckAnalysis.scala:41)
    	at org.apache.spark.sql.catalyst.analysis.Analyzer.failAnalysis(Analyzer.scala:91)
    	at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$11.apply(CheckAnalysis.scala:245)
    	at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$11.apply(CheckAnalysis.scala:242)
    	at scala.collection.immutable.List.foreach(List.scala:381)
    	at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:242)
    	at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:80)
    	at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:127)
    	at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.checkAnalysis(CheckAnalysis.scala:80)
    	at org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis(Analyzer.scala:91)
    	at org.apache.spark.sql.catalyst.analysis.Analyzer.executeAndCheck(Analyzer.scala:104)
    	at org.apache.spark.sql.execution.QueryExecution.analyzed$lzycompute(QueryExecution.scala:57)
    	at org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:55)
    	at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:47)
    	at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:74)
    	at org.apache.spark.sql.Dataset.withSetOperator(Dataset.scala:3307)
    	at org.apache.spark.sql.Dataset.union(Dataset.scala:1829)
    	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    	at java.lang.reflect.Method.invoke(Method.java:498)
    	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    	at py4j.Gateway.invoke(Gateway.java:282)
    	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    	at py4j.commands.CallCommand.execute(CallCommand.java:79)
    	at py4j.GatewayConnection.run(GatewayConnection.java:214)
    	at java.lang.Thread.run(Thread.java:748)


    
    During handling of the above exception, another exception occurred:


    AnalysisException                         Traceback (most recent call last)

    <ipython-input-50-eb00d5821249> in <module>()
    ----> 1 df2.union(df)#エラー！！
    

    ~/spark-2.3.0-bin-hadoop2.7/python/pyspark/sql/dataframe.py in union(self, other)
       1336         Also as standard in SQL, this function resolves columns by position (not by name).
       1337         """
    -> 1338         return DataFrame(self._jdf.union(other._jdf), self.sql_ctx)
       1339 
       1340     @since(1.3)


    ~/spark-2.3.0-bin-hadoop2.7/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py in __call__(self, *args)
       1158         answer = self.gateway_client.send_command(command)
       1159         return_value = get_return_value(
    -> 1160             answer, self.gateway_client, self.target_id, self.name)
       1161 
       1162         for temp_arg in temp_args:


    ~/spark-2.3.0-bin-hadoop2.7/python/pyspark/sql/utils.py in deco(*a, **kw)
         67                                              e.java_exception.getStackTrace()))
         68             if s.startswith('org.apache.spark.sql.AnalysisException: '):
    ---> 69                 raise AnalysisException(s.split(': ', 1)[1], stackTrace)
         70             if s.startswith('org.apache.spark.sql.catalyst.analysis'):
         71                 raise AnalysisException(s.split(': ', 1)[1], stackTrace)


    AnalysisException: "Union can only be performed on tables with the same number of columns, but the first table has 4 columns and the second table has 3 columns;;\n'Union\n:- AnalysisBarrier\n:     +- LogicalRDD [name#107, age#108, hight#109, weight#110], false\n+- AnalysisBarrier\n      +- LogicalRDD [name#101, age#102, hight#103], false\n"



```python
def superunionall(newdf, olddf):
    """newdf と olddf をunionallする。

    olddfに無くてnewdfにあるカラムはolddfにnullのカラムを追加した後、unionされる。
    olddfにあってnewdfにないカラムはolddfからカラムを削除した後、unionされる。

    Args:
        newdf (DataFrame): 基準となるdf
        olddf  (DataFrame): newdfのカラムの形に変形されてからnewdfにunionallさせる。

    Returns:
        DataFrame: newdfとolddfをunionallしたDataFrame

    """
    diff = set(newdf.dtypes) - set(olddf.dtypes)
    for i in diff:
        olddf = olddf.withColumn(i[0], F.expr(" cast (null as {})".format(i[1]) ) )
    newdfcolname = [ i[0] for i in newdf.dtypes]
    olddf = olddf.select(*newdfcolname)
    unionalldf = newdf.union(olddf)
    return unionalldf
```


```python
uniondf = superunionall(df2, df)
```


```python
uniondf.show()
```

    +----+---+-----+------+
    |name|age|hight|weight|
    +----+---+-----+------+
    | Bob| 23|165.3|  60.2|
    |Jack| 30|165.3|  82.4|
    |Judy| 22|172.3|  63.4|
    | Tom| 21|175.3|  null|
    |Mary| 30|165.3|  null|
    +----+---+-----+------+