Débit d'air: Comment SSH et exécuter BashOperator à partir d'un autre serveur

Est-il un moyen de ssh à autre serveur et exécuter BashOperator utilisant Airbnb du flux d'Air?
Je suis en train de lancer une ruche commande sql avec le flux d'Air, mais j'ai besoin de faire du SSH vers une autre zone afin d'exécuter la ruche shell.
Mes tâches devrait ressembler à ceci:

  1. SSH pour serveur1
  2. début de la Ruche shell
  3. exécuter Ruche de commande

Merci!

InformationsquelleAutor CMPE | 2016-09-12
  1. 26

    Je pense que j'ai juste pensé à elle:

    1. Créer une connexion SSH dans l'INTERFACE utilisateur de l'Admin > Connexion. Remarque: la connexion sera supprimé si vous réinitialisez la base de données

    2. Dans le fichier Python ajouter le suivant

      from airflow.contrib.hooks import SSHHook
sshHook = SSHHook(conn_id=<YOUR CONNECTION ID FROM THE UI>)

    3. Ajouter le SSH opérateur de tâche

      t1 = SSHExecuteOperator(
    task_id="task1",
    bash_command=<YOUR COMMAND>,
    ssh_hook=sshHook,
    dag=dag)

    Merci!

    • Remarque vous pouvez aussi importer de l'opérateur: à partir de la circulation de l'air.contrib.des opérateurs.ssh_execute_operator importation SSHExecuteOperator
    • Avec les derniers flux d'air de la version 1.10 SSHExecuteOperator est obsolète et nouveau SSHOperator doit être utilisé. Si quelqu'un est à l'aide de 1.10 puis de nouveau à l'importation devrait être from airflow.contrib.hooks.ssh_hook import SSHHook et from airflow.contrib.operators.ssh_operator import SSHOperator.
    InformationsquelleAutor CMPE
  2. 13

    Une chose à noter avec Anton réponse est que l'argument est en fait ssh_conn_id, pas conn_id pour la SSHOperator objet. Au moins dans la version 1.10.

    Un exemple rapide ressemblerait

    from datetime import timedelta, datetime
import airflow
from airflow import DAG
from airflow.contrib.operators.ssh_operator import SSHOperator
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['[email protected]'],
    'email_on_failure': False,
    'email_on_retry': False,
    'start_date': datetime.now() - timedelta(minutes=20),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}
dag = DAG(dag_id='testing_stuff',
          default_args=default_args,
          schedule_interval='0,10,20,30,40,50 * * * *',
          dagrun_timeout=timedelta(seconds=120))
# Step 1 - Dump data from postgres databases
t1_bash = """
echo 'Hello World'
"""
t1 = SSHOperator(
    ssh_conn_id='ssh_default',
    task_id='test_ssh_operator',
    command=t1_bash,
    dag=dag)
    • cela devrait être la réponse pour apache air 1.10
    • c'est joliment fait. LOL@ # Step 1 - Dump data from postgres databases
    InformationsquelleAutor user1978220