DataFrame转DynamicFrame时出现'str'对象无_jvm属性报错如何解决

报错根因 #

你遇到的AttributeError: 'str' object has no attribute '_jvm'报错，是调用DynamicFrame.fromDF方法时传入的第二个参数ctx不是合法的GlueContext对象，而是字符串类型的值导致的。

现有代码的两个错误 #

• 缩进错误：return语句写在了extractCustomFields函数作用域之外，函数内部定义的rec变量无法被外部访问，同时此时读取到的ctx是作用域外的非预期字符串值
• 参数传递错误：如果该函数是作为Glue动态帧的转换规则传入Map.apply方法，自定义转换函数默认仅接收DynamicRecord单个入参，显式传入的ctx参数会拿到错误的字符串值

正确实现代码 #

你可以把GlueContext定义为全局变量，或者在转换函数内部直接获取当前GlueContext实例，参考代码如下：

from awsglue.context import GlueContext
from pyspark.context import SparkContext
from awsglue.dynamicframe import DynamicFrame

# 全局初始化GlueContext
sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)

def extractCustomFields(record):
    rec = record.toDF()
    rec = rec.withColumn("lastname", rec["customfields"][0].value)
    # 函数内直接使用全局的glueContext完成转换
    return DynamicFrame.fromDF(rec, glueContext, "recordTransform").first()

# 调用示例：对动态帧执行自定义转换
# final_dyf = Map.apply(frame=original_dyf, f=extractCustomFields)

额外注意事项 #

如果你的处理逻辑是对全量数据做列扩展，不需要逐行转换为DataFrame再转回来，直接对原动态帧整体转DataFrame处理后再转动态帧即可，性能会远高于逐行处理：

# 批量处理性能更优
df = original_dyf.toDF()
df = df.withColumn("lastname", df["customfields"][0].value)
final_dyf = DynamicFrame.fromDF(df, glueContext, "recordTransform")

内容的提问来源于stack exchange，提问作者Miroslav Petrovic