基于EMR PySpark从S3加载CSV至Redshift的报错与IAM配置咨询

一、关于自定义IAM角色的配置答复 #

• 绑定在Redshift集群上的ROLE1，是Redshift服务侧用来访问S3执行COPY/UNLOAD操作的角色，不需要配置到EMR集群创建流程中。两类服务的IAM权限链路完全独立：
  • EMR侧的资源访问权限由JobFlowRole（EC2实例角色）、ServiceRole（EMR服务角色）控制，负责访问S3上的Jar包、脚本、输入CSV、日志文件
  • Redshift侧的ROLE1仅在Redshift集群内部执行S3访问时生效，和EMR运行时的权限逻辑无关联
• 你当前PySpark代码采用原生JDBC直连Redshift的写入方式，整个流程是Spark通过网络端口直接传数据到Redshift，完全不会触发Redshift侧的S3访问逻辑，因此ROLE1在当前链路中根本不会被调用到，无需额外在EMR侧做绑定

二、当前任务报错的根因定位 #

日志中展示的“应用退出码1”是YARN层面的通用失败返回，没有打印Python代码的具体堆栈，所有问题都来自配置错误

集群创建代码的硬伤 #

• 区域参数填写不合法：region_name='us ' 不是有效的AWS区域标识，必须替换为实际部署的区域值
• 日志路径配置不完整：LogUri="s3:///new-log" 缺失S3桶名，会导致EMR无法生成步骤运行日志，你自然看不到Python侧的具体报错
• spark-submit参数配置错误：
  • 多余的sudo参数：EMR步骤本身以root权限执行，额外加sudo会直接导致命令执行异常
  • 依赖包版本不兼容：你使用的EMR 5.32.0版本自带Spark 3.0.1，对应Scala 2.12，但你引入的spark-redshift_2.10是针对Scala 2.10编译的，会直接触发类加载失败
  • 依赖冗余：你的代码采用原生JDBC写Redshift，根本不需要引入spark-redshift、spark-avro、minimal-json这些包，多引入的不兼容包只会增加报错概率

PySpark代码的逻辑问题 #

• Schema配置冲突：读取CSV时同时开启inferSchema=true和手动传入自定义Schema，Spark会优先采用自动推导的Schema结果，手动定义的规则完全不生效，两种配置同时存在会导致解析逻辑混乱
• 类型定义不规范：你定义的DecimalType没有指定精度和标度，使用默认类型会和Redshift表的字段类型不匹配，触发类型转换报错
• 网络连通性风险：未确认EMR所在子网、安全组是否允许出站连接Redshift的5439端口，网络不通会直接触发连接失败，返回退出码1

三、修正后的配置参考 #

EMR创建流程参数修正 #

cluster_id = connection.run_job_flow(
    Name='Spark_DIF1',
    # 补全合法的S3日志路径
    LogUri="s3://你的实际日志桶/new-log",
    ReleaseLabel='emr-5.32.0',
    Applications=[{'Name': 'Spark'}],
    Instances={
        'InstanceGroups': [
            {
                'Name': "Master",
                'Market': 'ON_DEMAND',
                'InstanceRole': 'MASTER',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 1,
            },
            {
                'Name': "Slave nodes",
                'Market': 'ON_DEMAND',
                'InstanceRole': 'CORE',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 2,
            }
        ],
        'KeepJobFlowAliveWhenNoSteps': False,
        'TerminationProtected': False,
        'Ec2SubnetId': 'subnet-5b673f01',
    },
    Steps=[
        {
            'Name': 'Run Step',
            'ActionOnFailure': 'TERMINATE_CLUSTER',
            'HadoopJarStep': {
                'Jar': 'command-runner.jar',
                # 移除错误参数与冗余依赖
                'Args': [
                    'spark-submit',
                    '--master', 'yarn',
                    '--deploy-mode', 'cluster',
                    '--jars','s3a://你的实际桶/jars/RedshiftJDBC42-no-awssdk-1.2.51.1078.jar',
                    '--py-files', 's3a://你的实际桶/S3Redshift.py',
                    's3a://你的实际桶/S3Redshift.py'
                ]
            }
        }
    ], 
    VisibleToAllUsers=True,
    JobFlowRole='EMR_EC2_DefaultRole',
    ServiceRole='EMR_DefaultRole',
    Tags=[{'Key': 'tag_name_1', 'Value': 'tab_value_1'}],
)

PySpark代码修正 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, DecimalType, StringType

spark = SparkSession.builder.appName('app').getOrCreate()
log4jLogger = spark._jvm.org.apache.log4j
logger = log4jLogger.LogManager.getLogger(__name__)

# 为DecimalType指定明确的精度与标度
Input_schema1 = StructType([
    StructField("sepal_length", DecimalType(3,1), True),
    StructField("sepal_width", DecimalType(3,1), True),
    StructField("petal_length", DecimalType(3,1), True),
    StructField("petal_width", DecimalType(3,1), True),
    StructField("species", StringType(), True)
])
# 移除inferSchema参数，直接使用自定义Schema
lookup_df = spark.read \
              .option("header", "true") \
              .schema(Input_schema1) \
              .csv("s3://mltest11/iris.csv")
lookup_df.createOrReplaceTempView("dif_tbl")
final_df= spark.sql("select * from dif_tbl")

final_df.write.format('jdbc').options(
      url='jdbc:redshift://你的Redshift集群地址:5439/dev',
      driver='com.amazon.redshift.jdbc42.Driver',
      dbtable='dev.table',
      user='你的Redshift账号',
      password='你的Redshift密码').mode('overwrite').save()

spark.stop()

四、排查技巧 #

• 将步骤的ActionOnFailure从CONTINUE调整为TERMINATE_CLUSTER，补全合法的LogUri后，就可以在S3日志路径下找到steps/目录中的Python应用完整堆栈，直接定位代码层面的报错点
• 临时将KeepJobFlowAliveWhenNoSteps设为True，步骤运行失败后集群不会自动终止，可以通过Master节点的YARN UI查看应用的stderr输出，快速定位具体错误
• 确认EMR实例的安全组出站规则，允许访问Redshift的5439端口，优先走VPC内网连通，不要依赖公网访问Redshift的配置

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