如何无需上传S3脚本，通过Boto3向EMR并行提交PySpark任务？

可行性说明 #

完全可行。EMR支持通过spark-submit的-e参数直接执行内嵌的PySpark代码字符串，无需提前将脚本上传至S3。你可以把read_write这类函数的逻辑直接写在Boto3代码里，转换成字符串后嵌入EMR步骤的命令中，实现无脚本文件的任务提交。

实现思路 #

• 将PySpark函数的代码转换成多行字符串，处理好引号、换行等格式问题
• 利用spark-submit -e参数执行内嵌代码，替代引用S3脚本路径的方式
• 通过字符串格式化替换动态参数（如输入输出路径）
• 保持原有Boto3并行提交步骤的逻辑，只需修改每个步骤的Args部分

当前实现代码（上传S3脚本方式） #

import boto3

emr_client = boto3.client('emr', region_name='us-east-1')

# 依赖提前上传到S3的PySpark脚本
step_args = [
    'spark-submit',
    's3://my-bucket/pyspark_scripts/read_write.py',
    '--input-path', 's3://my-bucket/input',
    '--output-path', 's3://my-bucket/output'
]

steps = [
    {
        'Name': 'Read-Write Task 1',
        'ActionOnFailure': 'CONTINUE',
        'HadoopJarStep': {
            'Jar': 'command-runner.jar',
            'Args': step_args
        }
    }
]

# 并行提交多个步骤示例
response = emr_client.add_job_flow_steps(
    JobFlowId='j-XXXXXXXXX',
    Steps=steps
)

期望实现代码（内嵌PySpark函数方式） #

import boto3

emr_client = boto3.client('emr', region_name='us-east-1')

# 直接定义PySpark逻辑的字符串，注意引号转义
pyspark_template = """
from pyspark.sql import SparkSession

def read_write(input_path, output_path):
    spark = SparkSession.builder.appName("InlineReadWrite").getOrCreate()
    df = spark.read.parquet(input_path)
    df.write.mode("overwrite").parquet(output_path)
    spark.stop()

# 执行函数
read_write("{input_path}", "{output_path}")
"""

# 构造多个并行任务的步骤列表
parallel_steps = []
# 任务1参数
task1_input = 's3://my-bucket/input/task1'
task1_output = 's3://my-bucket/output/task1'
task1_code = pyspark_template.format(input_path=task1_input, output_path=task1_output)
parallel_steps.append({
    'Name': 'Inline Read-Write Task 1',
    'ActionOnFailure': 'CONTINUE',
    'HadoopJarStep': {
        'Jar': 'command-runner.jar',
        'Args': ['spark-submit', '--master', 'yarn', '-e', task1_code]
    }
})

# 任务2参数（可快速复制添加更多并行任务）
task2_input = 's3://my-bucket/input/task2'
task2_output = 's3://my-bucket/output/task2'
task2_code = pyspark_template.format(input_path=task2_input, output_path=task2_output)
parallel_steps.append({
    'Name': 'Inline Read-Write Task 2',
    'ActionOnFailure': 'CONTINUE',
    'HadoopJarStep': {
        'Jar': 'command-runner.jar',
        'Args': ['spark-submit', '--master', 'yarn', '-e', task2_code]
    }
})

# 并行提交所有步骤
response = emr_client.add_job_flow_steps(
    JobFlowId='j-XXXXXXXXX',
    Steps=parallel_steps
)

注意事项 #

• 代码字符串中的引号要统一：比如外层用双引号包裹，内部代码用单引号，避免语法错误；如果用f-string，需要把代码中的大括号转写成{{}}
• 复杂逻辑建议保持多行字符串格式，提升可读性
• 并行任务的Name必须唯一，避免EMR步骤识别冲突
• 确保EMR集群的IAM角色拥有访问目标S3路径的权限
• 若代码过长超过命令行参数长度限制，可通过Boto3动态生成临时S3文件（无需手动上传），再用spark-submit引用该临时文件路径

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