在将 DF 创建到 tempView 后运行 Spark Glue 作业时出错

问题描述

 Explanation

当我从动态框架创建 DF 时，它工作正常，我可以写 数据帧回到动态帧但是当我将数据帧转换为
createOrReplaceTempView 然后它向我抛出这个错误。列数 是相同的，从源到目的地没有任何变化

请帮我看看出了什么问题

ERROR

IllegalArgumentException: "requirement Failed: The number of columns doesn't 
match.\nOld column names (2): name,id\nNew column names (0): "

pysparkglueCode

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.dynamicframe import DynamicFrame
from awsglue.context import glueContext
from awsglue.job import Job
from pyspark.sql.types import *
from pyspark.sql.functions import *

from pyspark import SparkConf,SparkContext
from pyspark.sql import *



 args = getResolvedOptions(sys.argv,['TempDir','JOB_NAME'])

sc = SparkContext()
glueContext = glueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'],args)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database =   
"mygluedatabaSEOregon",table_name = "dev_glue_poc_gluetable",redshift_tmp_dir =  
args["TempDir"],transformation_ctx = "datasource0")

applymapping1 = ApplyMapping.apply(frame = datasource0,mappings = [("name","string","name","string"),("id","int","id","int")],transformation_ctx =   
"applymapping1")

selectfields2 = SelectFields.apply(frame = applymapping1,paths = ["name","id"],transformation_ctx = "selectfields2")

getoutput = selectfields2.toDF() #.select('name','id')

getoutput.createOrReplaceTempView("info")

sqlData = spark.sql("select name,id from info")

outputsql = sqlData.toDF()

getoutputDFY = DynamicFrame.fromDF(outputsql,glueContext,"getoutputDFY")


resolvechoice3 = ResolveChoice.apply(frame = getoutputDFY,choice = 
"MATCH_CATALOG",database = "mygluedatabaSEOregon",table_name = 
"dev_glue_poc_importfromglue",transformation_ctx = "resolvechoice3")



resolvechoice4 = ResolveChoice.apply(frame = resolvechoice3,choice = "make_cols",transformation_ctx = "resolvechoice4")



  datasink5 = glueContext.write_dynamic_frame.from_catalog(frame = resolvechoice4,table_name = "dev_glue_poc_importfromglue",redshift_tmp_dir = args["TempDir"],transformation_ctx = "datasink5")
  job.commit()

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）