如何在AWS Glue中将DataFrame转换为Dynamic Frame？

现有代码优化建议 #

你当前代码中调用fromDF的方式不规范，该方法是DynamicFrame类的静态方法，不需要用实例调用，优化后的标准写法如下：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame

glueContext = GlueContext(SparkContext.getOrCreate())

# 从数据目录加载数据
students = glueContext.create_dynamic_frame.from_catalog(database="example_db", table_name="samp_csv")

# 转成DataFrame做转换
students_df = students.toDF()
students_transformed_df = students_df.withColumnRenamed("state","County")\
    .withColumnRenamed("capital","cap")\
    .drop("municipal","metropolitan")

# 标准方式转回DynamicFrame
students_trans = DynamicFrame.fromDF(students_transformed_df, glueContext, "students_trans")

# 写入S3
glueContext.write_dynamic_frame.from_options(frame = students_trans,
              connection_type = "s3",
              connection_options = {"path": "s3://kingb/target/"},
              format = "csv")

其他可选转换方案 #

除了上述官方最推荐的标准转换方法外，还有两种可替代的实现方式：

• 方法1：通过RDD中转生成DynamicFrame
  适合需要对RDD做自定义底层处理的场景，步骤是先把DataFrame转成RDD，再通过GlueContext的RDD加载接口生成DynamicFrame：

# 转换后的DataFrame转RDD
students_rdd = students_transformed_df.rdd
# 从RDD生成DynamicFrame，可复用DataFrame原有schema
students_trans = glueContext.create_dynamic_frame.from_rdd(
    rdd=students_rdd,
    name="students_trans",
    schema=students_transformed_df.schema
)

• 方法2：直接用Glue原生转换避免互转
  如果只是做改列名、删列这类简单转换，不需要转成PySpark DataFrame，直接用Glue自带的Transform算子操作DynamicFrame即可，避免互转开销，整体运行效率更高：

from awsglue.transforms import ApplyMapping, DropFields

# 直接修改列名，不需要转DF
students_rename = ApplyMapping.apply(
    frame=students,
    mappings=[
        ("state", "string", "County", "string"),
        ("capital", "string", "cap", "string"),
        # 其余需要保留的列按上述格式补充即可
        ("id", "int", "id", "int"),
        ("name", "string", "name", "string")
    ]
)

# 直接删除不需要的列
students_trans = DropFields.apply(frame=students_rename, paths=["municipal", "metropolitan"])

内容的提问来源于stack exchange，提问作者Abiodun Adeoye