我基本上是在尝试从一个DF向另一个DF更新/添加行。 这是我的密码:
# S3
import boto3
# SOURCE
source_table = "someDynamoDbtable"
source_s3 = "s://mybucket/folder/"
# DESTINATION
destination_bucket = "s3://destination-bucket"
#Select which attributes to update/add
params = ['attributeD', 'attributeF', 'AttributeG']
#spark wrapper
glueContext = GlueContext(SparkContext.getOrCreate())
newData = glueContext.create_dynamic_frame.from_options(connection_type = "dynamodb", connection_options = {"tableName": source_table})
newValues = newData.select_fields(params)
newDF = newValues.toDF()
oldData = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options={"paths": [source_s3]}, format="orc", format_options={}, transformation_ctx="dynamic_frame")
oldDataValues = oldData.drop_fields(params)
oldDF = oldDataValues.toDF()
#makes a union of the dataframes
rebuildData = oldDF.union(newDF)
#error happens here
readyData = DynamicFrame.fromDF(rebuildData, glueContext, "readyData")
#writes new data to s3 destination, into orc files, while partitioning
glueContext.write_dynamic_frame.from_options(frame = readyData, connection_type = "s3", connection_options = {"path": destination_bucket}, format = "orc", partitionBy=['partition_year', 'partition_month', 'partition_day'])
我得到的错误是SyntaxError:invalid syntaxon linereadyData=。。。 到目前为止,我还不知道怎么了,我的头从所有的空白代码开始痛。 如果你见过类似的东西,请给我一根骨头。你知道吗
您正在执行dataframe和dynamicframe之间的并集操作。你知道吗
这将创建名为newData的动态帧和名为newDF的数据帧:
这将创建名为oldData的动态帧和名为oldDF的数据帧:
您正在对上述两个实体执行联合操作,如下所示:
应该是:
是的,所以我想我需要做的是使用外部连接。 让我解释一下:
现在,我的逻辑可能有缺陷,但到目前为止,它是工作的好我。如果有人正在寻找类似的解决方案,欢迎您使用。 我的代码已更改:
相关问题 更多 >
编程相关推荐