submit-zipped-pyspark-job-to-emr.py

import boto3
from os.path import join


def lambda_handler(event, context):
    emr = boto3.client('emr')
    version = 'latest'
    main_path = join('s3://<artifacts-bucket-name>', version, 'main.py')
    modules_path = join('s3://<artifacts-bucket-name>', version, 'module_seed.zip')

    job_parameters = {
        'job_name': '<your-job-name>',
        'input_path': 's3://<raw-data-path>',
        'output_path': 's3://<processed-data-path>',
        'spark_config': {
            '--executor-memory': '1G',
            '--driver-memory': '2G'
        }
    }

    step_args = [
        "/usr/bin/spark-submit",
        '--py-files', modules_path,
        main_path, str(job_parameters)
    ]

    step = {
        "Name": job_parameters['job_name'],
        'ActionOnFailure': 'CONTINUE',
        'HadoopJarStep': {
            'Jar': 's3://us-east-1.elasticmapreduce/libs/script-runner/script-runner.jar',
            'Args': step_args
        }
    }

    action = emr.add_job_flow_steps(JobFlowId='<emr-cluster-id>', Steps=[step])
    return action