Remote SSH Command Execution Using CLI

See also questions close to this topic

• airflow v1.10.1 - SSHOperator does not log the ssh command

  Since we upgraded airflow from v1.8.1 to v1.10.1 the ssh command that is executed by the SSHOperator is not showing in the logs anymore. now you have the log;

  Subtask: Output:
         <the output of your ssh command>
  

  before you got the command + the output of the executed command. Before you had the log;

  Subtask: Running command: <your ssh command>
  Subtask: Output:
         <the output of your ssh command>
  

  any help on making the SSHOperator log the command again? it's good to have the command in the logs to debug your task.
• Airflow file name from s3 using s3KeySensor

  I'm using Airflow to schedule my spark jobs on files using AWS s3. So far, I was able to run a python operator right after the file was added to the data storage (using s3KeySensor). I would like to get the name of the file that was added, is it possible using airflow's s3keysensor?

  Thanks, Matar Peles
• Create hive partition based on time zone

  I'm trying to materialize hive table based on file that are stored as parquet in GCS, with path like gs://abc/dt=02-02-2019/hr=02(physical partition based on UTC)

  Now I want to create two hive table where the logical partition is based on timezone, say one for UTC and other for CET, how can I partition such that date and hour based partition picks the dt and hr value based on timezone. Also it would be great if it can also accommodate for day-light saving etc.

  I am using airflow to create external hive table.
• create tasks in Airflow by looping over a list and pass arguments

  edit: This will work, I defined ex_func_airflow(var_1 = i) which was causing the issue

  I would like to create tasks in airflow by looping on a list.

  tabs = [1,2,3,4,5]
  for i in tabs:
      task = PythonOperator(
      task_id = name,
      provide_context=False,
      op_args  = [i],
      python_callable=ex_func_airflow,
      dag=dag)
      task_0 >> task >> task_1
  

  When this is run in airflow the argument that is passed is always the last element in that list.

  So i'm essentially running:

  ex_func_airflow(6) 
  

  five times instead of running

  ex_func_airflow(1)
  ex_func_airflow(2)
  ex_func_airflow(3)
  

  ..etc.

  How would can I pass the correct arguments for each task?
• How to persist a value during airflow retry

  Everytime a dag triggers a task I would like to generate a value and keep the value during retry (if the task fails to run for any reason)

  I try to declare a variable inside the dag (basically set current time to the variable) When the task fails and it retries, it set the variable with latest time again
• dag_id could not be found: dag_id. Either the dag did not exist or it failed to parse after upgrading airflow from 1.7.3 to 1.10.1

  I upgraded the airflow version from 1.7.3 to 1.10.1. After up-gradation of the scheduler, webserver and workers, the dags have stopped working showing below error on scheduler-

  Either the dag did not exist or it failed to parse.

  I have not made any changes to the config. While investigating the issue the scheduler logs shows the issue. Earlier the scheduler run the task as -

  Adding to queue: airflow run <dag_id> <task_id> <execution_date> --local -sd DAGS_FOLDER/<dag_filename.py>

  While now it is running with absolute path -

  Adding to queue: airflow run <dag_id> <task_id> <execution_date> --local -sd /<PATH_TO_DAGS_FOLDER>/<dag_filename.py>

  PATH_TO_DAGS_FOLDER is like /home/<user>/Airflow/dags...

  which is same as what it is pushing it to workers by since worker is running on some other user it is not able to find the dag location specified.

  I am not sure how to tell the worker to look in it's own airflow home dir and not the scheduler one?

  I am using mysql as backend and rabbitmq for message passing.