我正在尝试使用Glue从dynamodb提取数据集到s3。在此过程中,我想选择少量列,然后为具有缺失值的任何和所有行/列设置默认值。
我目前尝试使用“Map”功能,但它似乎没有调用我的方法。
这是我有的:
def SetDefaults(rec):
print("checking record")
for col in rec:
if not rec[col]:
rec[col] = "missing"
return rec
## Read raw(source) data from target DynamoDB
raw_data_dyf = glueContext.create_dynamic_frame_from_options("dynamodb", {"dynamodb.input.tableName" : my_dynamodb_table, "dynamodb.throughput.read.percent" : "0.50" } )
## Get the necessary columns
selected_data_dyf = ApplyMapping.apply(frame = raw_data_dyf, mappings = mappingList)
## get rid of null values
mapped_dyF = Map.apply(frame=selected_data_dyf, f=SetDefaults)
## write it all out as a csv
datasink = glueContext.write_dynamic_frame.from_options(frame=mapped_dyF , connection_type="s3", connection_options={ "path": my_train_data }, format="csv", format_options = {"writeHeader": False , "quoteChar": "-1" })
我的ApplyMapping.apply
调用正在做正确的事情,其中mappingList
由一堆定义:
mappingList.append(('gsaid', 'bigint', 'gsaid', 'bigint'))
mappingList.append(('objectid', 'bigint', 'objectid', 'bigint'))
mappingList.append(('objecttype', 'bigint', 'objecttype', 'bigint'))
我没有错误,一切都完成了。我的数据全部在s3中,但仍有许多空值,而不是我想要的“缺失”条目。
“检查记录”打印语句永远不会打印出来。我在这里错过了什么?
替代方案: