[emr_add_steps_operator] Allow job_flow_id & job_flow_name (Priority is in the order of id, name.)

[seunggabi]

https://airflow.apache.org/docs/apache-airflow/1.10.12/_modules/airflow/contrib/operators/emr_add_steps_operator.html

I make a emr_add_steps_operator.
job_flow_id or job_flow_name exclusively existed in runtime.

so I want to allow existing id and name.

example code is this.

EmrAddStepsOperator(
    task_id=id,
    job_flow_id='{{ task_instance.xcom_pull(task_ids="init", key="emr_id", dag_id=task_instance.dag_id) }}',
    job_flow_name='{{ task_instance.xcom_pull(task_ids="init", key="emr_name", dag_id=task_instance.dag_id) }}',
    cluster_states=["WAITING", "RUNNING"],
    steps=f'{{{{ task_instance.xcom_pull(task_ids="init", key="return_value", dag_id=task_instance.dag_id)["{id}"] }}}}',
    do_xcom_push=True,
    dag=dag,
)

---

EmrAddStepsOperator pr code is this.

def __init__(
    self,
    job_flow_id=None,
    job_flow_name=None,
    cluster_states=None,
    aws_conn_id='aws_default',
    steps=None,
    *args, **kwargs):
    if (job_flow_id is None) and (job_flow_name is None):
        raise AirflowException('Exactly job_flow_id or job_flow_name must be existed.; Priority is in the order of job_flow_id -> job_flow_name.')

    super(EmrAddStepsOperator, self).__init__(*args, **kwargs)
    steps = steps or []
    self.aws_conn_id = aws_conn_id
    self.job_flow_id = job_flow_id
    self.job_flow_name = job_flow_name
    self.cluster_states = cluster_states
    self.steps = steps

[seunggabi]

pr: http://github.com/apache/airflow/pull/31358