AWS Glue中Delta Lake配置及ETL代码请求与报错排查

关键前提 #

确保你的Glue作业已配置参数：{"--datalake-formats": "delta"}，且使用Glue 3.0+版本（原生支持Delta Lake，避免额外依赖问题）。

完整可复制代码 #

import sys
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext

# 解析作业参数（需在作业配置中传入JOB_NAME和S3_OUTPUT_PATH）
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'S3_OUTPUT_PATH'])

# 初始化Glue与Spark上下文
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# 配置Delta Lake核心Spark参数（必须添加）
spark.conf.set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
spark.conf.set("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

# 初始化Glue作业
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# --------------------------
# 替换为你的ETL业务逻辑
# 示例1：读取S3 CSV文件
df = spark.read.csv("s3://your-input-bucket/input-data/", header=True, inferSchema=True)

# 示例2：读取Glue数据目录中的表
# df = glueContext.create_dynamic_frame.from_catalog(
#     database="your-database", 
#     table_name="your-table"
# ).toDF()
# --------------------------

# 以Delta格式写入目标S3路径
df.write.format("delta").mode("overwrite").save(args['S3_OUTPUT_PATH'])

# 提交作业
job.commit()

错误原因与解决 #

1. ModuleNotFoundError: No module named 'delta'

   • 检查作业参数是否正确配置--datalake-formats delta，Glue会自动加载官方集成的Delta依赖，无需手动执行pip install delta
   • 确认使用Glue 3.0+版本，低版本需额外配置依赖包

2. NameError: name 'configure_spark_with_delta_pip' is not defined

   • 该函数是Delta社区版的初始化方式，在AWS Glue托管环境中无需使用，Glue已通过作业参数集成Delta依赖，只需配置上述Spark参数即可

作业配置补充 #

• 在Glue作业的「作业参数」中添加：
  • 键：--datalake-formats，值：delta
• 确保作业IAM角色拥有目标S3路径的读写权限、Glue数据目录访问权限

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