title: Create and use Connections in Airflow operator at runtime
body: <p><em>Note: This is NOT a duplicate of</em></p>
  
  <ul>
  <li><a href="https://stackoverflow.com/questions/51312425/export-environment-variables-at-runtime-with-airflow">Export environment variables at runtime with airflow</a></li>
  <li><a href="https://stackoverflow.com/questions/50643106/set-airflow-env-vars-at-runtime">Set Airflow Env Vars at Runtime</a></li>
  </ul>
  
  <hr>
  
  <p>I have to trigger certain tasks at remote systems from my <code>Airflow</code> <code>DAG</code>. The straight-forward way to achieve this is <a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/contrib/hooks/ssh_hook.py" rel="noreferrer"><code>SSHHook</code></a>.</p>
  
  <p>The problem is that the remote system is an <code>EMR</code> cluster which is itself created at runtime (by an <em>upstream task</em>) using <a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/contrib/operators/emr_create_job_flow_operator.py" rel="noreferrer"><code>EmrCreateJobFlowOperator</code></a>. So while I can get hold of <code>job_flow_id</code> of the launched EMR cluster (<a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/contrib/example_dags/example_emr_job_flow_manual_steps.py#L78" rel="noreferrer">using <code>XCOM</code></a>), what I need is to an <a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/contrib/hooks/ssh_hook.py#L61" rel="noreferrer"><code>ssh_conn_id</code></a> to be passed to each downstream task.</p>
  
  <hr>
  
  <p>Looking at the <a href="https://incubator-airflow.readthedocs.io/en/latest/concepts.html#connections" rel="noreferrer">docs</a> and <a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/hooks/base_hook.py#L80" rel="noreferrer">code</a>, it is evident that Airflow will try to look up for this connection (using <code>conn_id</code>) in <em>db</em> and <em>environment variables</em>, so now the problem boils down to being able to set either of these two properties at runtime (from <em>within</em> an <code>operator</code>).</p>
  
  <p>This seems a rather common problem because if this isn't achievable then the utility of <code>EmrCreateJobFlowOperator</code> would be severely hampered; but I haven't come across any example demonstrating it.</p>
  
  <hr>
  
  <ul>
  <li>Is it possible to create (and also destroy) either of these from within an Airflow operator?
  
  <ol>
  <li><a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/models.py#L631" rel="noreferrer"><em>Connection</em></a> (persisted in Airflow's db)</li>
  <li><em>Environment Variable</em> (should be accessible to all downstream tasks and not just current task as told <a href="https://stackoverflow.com/questions/51312425/export-environment-variables-at-runtime-with-airflow#comment89615294_51312425">here</a>)</li>
  </ol></li>
  <li>If not, what are my options?</li>
  </ul>
  
  <hr>
  
  <p>I'm on</p>
  
  <ul>
  <li><code>Airflow v1.10</code></li>
  <li><code>Python 3.6.6</code></li>
  <li><code>emr-5.15</code> (can upgrade if required)</li>
  </ul>