将缺少的列添加到 AWS Glue DataFrame

Question

我正在使用 Glue 读取 DynamoDB 表，由于动态架构，可能会出现某些列不存在的情况。 使用以下代码添加它们可以正常工作，但如果我需要添加多个列，我不确定如何使函数动态化。

# add missing columns if not available
def AddCustRegName(r):
    r["customerRegistrationName"] = ""  # add column with empty string.
    return r

if addCustRegName:
    case_df_final = Map.apply(frame=case_df_final, f=AddCustRegName)

有什么建议吗？

以下代码因以下错误而失败

# add missing columns if not available
def AddColumn(r, col):
    r[col] = ""  # add column with empty string.
    return r

case_df_final = Map.apply(frame=case_df_final, f=AddColumn(case_df_final ,'accessoryTaxIncluded'))

case_df_final.toDF().printSchema()

执行第 6 行失败：case_df_final = Map.apply(frame=case_df_final, f=AddColumn(case_df_final ,'accessoryTaxIncluded')) Traceback (最 最近通话最后）：文件 “/tmp/zeppelin_pyspark-4928209310219195923.py”，第 375 行，在 exec(code, _zcUserQueryNameSpace) File "", line 6, in File "", line 3, in AddColumn TypeError: 'DynamicFrame' 对象不支持项目分配

Answer 1

Map 传入的函数只能有一个参数：

f – 应用于 DynamicFrame 中所有 DynamicRecords 的函数。 该函数必须将 DynamicRecord 作为参数并返回一个新的 映射产生的DynamicRecord（必需）。

但是，您可以在 pyspark 数据框而不是 DynamicFrame 上执行此操作：

from pyspark.sql import functions as F

def AddColumn(sdf, new_col):
    return sdf.withColumn(new_col, F.lit(""))

case_sdf_final = AddColumn(case_df_final.toDF(), "accessoryTaxIncluded")

case_sdf_final.printSchema()

或者，如果您有要添加的列列表，您可以像这样使用functools.reduce：

import functools

new_cols = ["customerRegistrationName", "accessoryTaxIncluded"]

case_sdf_final = functools.reduce(
    lambda acc, c: AddColumn(acc, c),
    new_cols,
    case_df_final.toDF()
)

case_sdf_final.printSchema()

然后回到DynamicFrame：

case_df_final = DynamicFrame.fromDF(case_sdf_final, glueContext, "case_df_final")