EMR Cluster Creation using Airflow dag run, Once task is done EMR will be terminated
Absolutely, that would be the most efficient use of resources. Let me warn you: there are a lot of details in this; I'll try to list as many as would get you going. I encourage you to add your own comprehensive answer listing any problems that you encountered and the workaround (once you are through this)
Regarding cluster creation / termination
For cluster creation and termination, you have
EmrCreateJobFlowOperator
andEmrTerminateJobFlowOperator
respectivelyDon't fret if you do not use
AWS
SecretAccessKey
(and rely wholly onIAM
Roles); instantiating anyAWS
-relatedhook
oroperator
inAirflow
will automatically fall-back to underlyingEC2
's attachedIAM
RoleIf your'e NOT using the EMR-Steps API for job-submission, then you'll also have to manually sense both the above operations using
Sensors
. There's already a sensor for polling creation phase calledEmrJobFlowSensor
and you can modify it slightly to create a sensor for termination tooYou pass your cluster-config JSON in
job_flow_extra
. You can also pass configs in aConnection
's (likemy_emr_conn
)extra
param, but refrain from it because it often breaksSQLAlchemy
ORM loading (since its a bigjson
)
Regarding job submission
You either submit jobs to
Emr
using EMR-Steps API, which can be done either during cluster creation phase (within the Cluster-Configs JSON) or afterwards usingadd_job_flow_steps()
. There's even anemr_add_steps_operator()
inAirflow
which also requires anEmrStepSensor
. You can read more about it inAWS
docs and you might also have to usecommand-runner.jar
For application-specific cases (like
Hive
,Livy
), you can use their specific ways. For instance you can useHiveServer2Hook
to submit aHive
job. Here's a tricky part: Therun_job_flow()
call (made during cluster-creation phase) only gives you ajob_flow_id
(cluster-id). You'll have to use adescribe_cluster()
call usingEmrHook
to obtain the private-IP of the master node. Using this you will then be able to programmatically create aConnection
(such asHive Server 2 Thrift
connection) and use it for submitting your computations to cluster. And don't forget to delete those connections (for elegance) before completing your workflow.Finally there's the good-old bash for interacting with cluster. For this you should also pass an
EC2
key pair during cluster creation phase. Afterwards, you can programmatically create anSSH
connection and use it (with anSSHHook
orSSHOperator
) for running jobs on your cluster. Read more about SSH-stuff inAirflow
hereParticularly for submitting
Spark
jobs to remoteEmr
cluster, read this discussion
Check my implementation, DAG will create emr cluster and run the spark job against the data in s3 and terminate automatically once done.
https://beyondexperiment.com/vijayravichandran06/aws-emr-orchestrate-with-airflow/
The best way to do this is probably to have a node at the root of your Airflow DAG that creates the EMR cluster, and then another node at the very end of the DAG that spins the cluster down after all of the other nodes have completed.