pyspark更改列顺序存入iceberg数据库

创建环境，指定catalog

def get_spark():
    os.environ.setdefault('HADOOP_USER_NAME', 'root')
    # total size of serialized results of tasks is bigger than spark.driver.maxResultSize
    # ERROR DataWritingSparkTask: Aborting commit for partition 2 (task 2, attempt 0, stage 0.0) 内存不足
    spark = SparkSession.builder \
        .config('spark.sql.debug.maxToStringFields', 2000) \
        .config('spark.debug.maxToStringFields', 2000) \
        .config('spark.driver.memory', '16g') \
        .config('spark.executor.memory', '16g') \
        .config('spark.driver.maxResultSize', '4g') \
        .config('spark.network.timeout', 180) \
        .getOrCreate()
    spark.conf.set("spark.sql.catalog.iceberg",
                   "org.apache.iceberg.spark.SparkCatalog")
    spark.conf.set("spark.sql.catalog.iceberg.type", "hive")
    spark.conf.set("spark.sql.catalog.iceberg.uri",
                   "thrift://192.168.x.xx:9083")
    spark.conf.set("spark.sql.session.timeZone", "GMT+8") 
    spark.conf.set("spark.sql.iceberg.handle-timestamp-without-timezone", True)
    return spark

列 A B 转为 B A存入数据库，不更改顺序会报错列不匹配
1.spark dataframe 创建临时视图,再转为spark dataframe

spark = get_Spark()
spark_df = spark.sql('select A from ***')
spark_df = spark_df.withColumn("B", )
spark_df.createOrReplaceTempView("test")
DF = spark.sql("select B A from test")
DF.write.insertInto('xxx',  True)

2.创建schema(不推荐)
2.1直接 spark.createDataFrame(df)可能报错 schema ValueError: Some of types cannot be determined after inferring

存在字段spark无法推断它的类型，

2.2指定每个字段的类型
数据转换可能报错IntegerType can not accept object 2.0 in type <class 'float'>

schema = StructType([
                         StructField("xx", StringType(), True),
                         StructField("xx", IntegerType(), True)
                     ])