create an EMR step function inside jupyter notebook via python code

user_14221331 0
Viewed 435

Right now, I transferring a lot of small individual spark parquet files from EMR to S3. The way I am currently doing this is by creating a step function, which is what AWS recommends, in the clusters Steps UI, below is an example

JAR location :command-runner.jar
Main class :None
Arguments :/usr/bin/s3-dist-cp --src=/refoutput --dest=s3://***-us-east-1/bens-flattened-step/refs
Action on failure:Continue

I actually need to change parts of the --dest parameter depending on the files that are being uploaded. What I am trying to do, is to create the same step function, except with python code that I can put into my Jupyter notebook, as opposed to doing this with the UI. Is this possible? Also, spark.parquet.write('s3 path') causes S3 to hang when there are a lot of small files, so that is not a viable solution.

amazon-emr aws-step-functions apache-spark amazon-web-services python
1 Answers

From Notebook, you can use boto3 to list cluster and using the Cluster-ID, you can submit spark steps.

First. to install boto3, you can install pip package as

sc.install_pypi_package("boto3")
#You can check it by using sc.list_packages()

With boto3, you can either list the cluster Ids

import boto3

boto3 = boto3.session.Session(region_name='us-east-2')
emr = boto3.client('emr')

page_iterator = emr.get_paginator('list_clusters').paginate(
    ClusterStates=['RUNNING','WAITING']
)

for page in page_iterator:
    for item in page['Clusters']:
        print(item['Id'])

or simpler

response = emr.list_clusters(
    ClusterStates=[
        'RUNNING','WAITING',
    ],
   
)
print(response)

You can filter the cluster as well for example with date arguments to list_cluster

CreatedAfter=datetime(2019, 11, 11),
CreatedBefore=datetime(2020, 1, 1)

Not having cluster-id in hand, you can submit the step with add_job_flow_steps

For example, a step can be

newsteps =[
                 {
            'Name': 'AWS S3 Copy',
            'ActionOnFailure': 'CONTINUE',
            'HadoopJarStep': {
                'Jar': 'command-runner.jar',
                'Args':["aws","s3","cp","s3://XXX/","/home/hadoop/copy/","--recursive"],
                        }
              }
              ]

Adding the step with the cluster-id you retrieved (e,g j-xxxxxxxxxxx)

step_response = emr.add_job_flow_steps(JobFlowId="j-xxxxxxxxxxxxxxx", Steps=newsteps)

step_ids = step_response['StepIds']

print("Step IDs:", step_ids)
user_36808310
Related