Remote SSH Command Execution Using CLI

See also questions close to this topic

• Airflow task takes 7 minutes to execute

  I am starting to use Airflow and my first workflow involves moving a file from GCP to S3 (back and forth).

  The task doing the work (and the whole DAG) finishes successfully but it takes 7 minutes as can be seen on the logs below for the file transfer (and I guess some authentication and protocol stuff).

  [2018-09-19 13:58:34,498] {logging_mixin.py:95} INFO - [2018-09-19 13:58:34,496] {credentials.py:1032} INFO - Found credentials in shared credentials file: ~/.aws/credentials

  [2018-09-19 14:05:55,920] {logging_mixin.py:95} INFO - [2018-09-19 14:05:55,920] {gcp_api_base_hook.py:84} INFO - Getting connection using google.auth.default() since no key file is defined for hook.

  In the same DAG there is a task to do the complementary task, file transfer from S3 to GCP and that is fast ( less than 1 minute ).

  from __future__ import print_function
  
  from builtins import range
  from datetime import datetime
  import airflow
  from airflow.operators import OmegaFileSensor
  from airflow.operators.python_operator import PythonOperator
  from airflow.contrib.operators.s3_to_gcs_operator import S3ToGoogleCloudStorageOperator
  from airflow.contrib.operators.gcs_to_s3 import GoogleCloudStorageToS3Operator
  from airflow.contrib.sensors.gcs_sensor import GoogleCloudStorageObjectSensor
  from airflow.models import DAG
  
  import time
  from pprint import pprint
  
  S3_BUCKET = 'data-preprod-redshift-exports'
  # S3_OBJECT = 'airflow/seattlecheckoutsbytitle.zip' # 2GB
  S3_OBJECT = 'airflow/cnpjqsa.zip' # 400Mb   
  # S3_OBJECT = '/airflow/chicagobusinesslicensesandowners.zip' # 100 Mb  
  
  GCS_BUCKET = 'ds_de_airflow' 
  
  args = {
      'owner': 'airflow',
      'start_date': datetime(2018,9,18)#,
      #'execution_timeout':None,
      #'dagrun_timeout': None
  }
  
  def print_context(ds, **kwargs):
      pprint(kwargs)
      print(ds)
      return 'Whatever you return gets printed in the logs'
  
  with DAG( dag_id='a_second', default_args=args, schedule_interval=None) as dag:
  
      run_this = PythonOperator(
          task_id='run_this',
          provide_context=True,
          python_callable=print_context
      )
  
      s3_to_gcs_op = S3ToGoogleCloudStorageOperator( 
          task_id = 's3_to_gcs_op', 
          bucket = S3_BUCKET,
          prefix = S3_OBJECT, 
          dest_gcs_conn_id = 'google_cloud_default',
          dest_gcs = 'gs://ds_de_airflow/Task1_upload/', 
          replace = False
      )
  
      # for some reason this takes no less than 7 minutes (tried 3 times) 
      gcs_to_s3_op = GoogleCloudStorageToS3Operator(
          task_id = 'gcs_to_s3_op', 
          bucket = GCS_BUCKET,
          prefix = 'Task1_upload',
          delimiter = 'fileGCS.txt',
          google_cloud_storage_conn_id ='google_cloud_default',
          dest_aws_conn_id = 'aws_default',
          dest_s3_key = 's3://data-preprod-redshift-exports/airflow/',
          replace = False
      )
  
      gcs_sensor = GoogleCloudStorageObjectSensor(
          task_id = 'gcs_sensor',
          bucket = GCS_BUCKET,
          object = 'Task1_upload/airflow/fileS3.txt'  # this is not the most interesting file to sensor for but for now...
      )
  
      run_this >> s3_to_gcs_op >> gcs_sensor >> gcs_to_s3_op
  

  We have install airflow in the Google Cloud Shell, with the default db engine (1 thread).

  The question is: how to reduce the execution time of that 7 min task to something more sensible?
• when trying to run dag and Airflow returning status code2

  Hey I am really stuggling here with airflow on this. I have two other dags running no probllem however for some reasons these dags are failing and I'm not getting much back that makes sense to me in the logs. I can run this task with no issues as the airflow user through the command line but for some reason when I try doing it through the dag I get exit code 2. Has anyone some pointers on this? Thank you

  [2018-09-18 20:08:18,886] {base_task_runner.py:95} INFO - Subtask: airflow: error: unrecognized arguments: run pull_dcm_today_data pull_dcm_today_data 2018-09-18T19:00:00 --job_id 8 --raw -sd DAGS_FOLDER/run_pull_dcm_today.py

  [2018-09-18 20:08:22,579] {jobs.py:2083} INFO - Task exited with return code 2
• How do I regularly deactivate inactive DAGs?

  As DAGs get deleted over time they will still appear on the web UI, so I'm interested in running a process to regularly remove inactive DAGs to keep the UI as uncluttered as possible.

  I notice that in airflow.models.DagBag there is a method called deactivate_inactive_dags() that appears to be this functionality, but I see it is not called anywhere in code. How is this method intended to be used?
• Airflow - triggered DAG but DAG state is None. And tasks within DAG don't run

  I have a DAG in Airflow that isn't paused. No matter how I trigger it (through CLI or through the web UI), the tasks within the DAG never runs. They get stuck with no status. This is how it looks like in the web UI.

  

  Within DAG view:

  Here is an example of how triggering it through CLI doesn't work either.

  $ airflow trigger_dag reporting_tables_weekly_
  ...<output truncated>
  [2018-09-17 13:09:10,215] {cli.py:185} INFO - Created <DagRun reporting_tables_weekly_ 
  @ 2018-09-17 13:09:10: manual__2018-09-17T13:09:10, externally triggered: True>
  
  $ ~/airflow$ airflow dag_state reporting_tables_weekly_ "2018-09-17"
  ...
  None
  

  This is what Task Instance Details show:

  I've run the DAGs using airflow test and both tasks run fine and generates the expected output.

  The scheduler is running and isn't under heavy load (currently have only 3 concurrent tasks running). I've also deleted and recreated the DAG but the problem persists. Can anyone help please?
• How to restart a failed task on Airflow and keep the failure record?

  My DAG looks like this:

  task1 >> task2 >> task3

  and task2 has failed. I would like to restart the dagrun from the last failure point which is task2, and I understand I can use "clear" for that.

  The issue is that when I use clear the execution attempt is cleared and looking back I have no record of this failure anymore.

  I'm wondering if I can rerun, but still keep the failure history.
• use airflow hive operator and output to a text file

  HI i want to execute hive query using airflow hive operator and output the result to a file. I don't want to use INSERT OVERWRITE here.

  # hive_ex = HiveOperator(
  #     task_id='hive-ex',
  #     hql='/sql/hive-ex.sql',
  #     hiveconfs={
  #         'DAY': '{{ ds }}',
  #         'YESTERDAY': '{{ yesterday_ds }}',
  #         'OUTPUT': '{{ file_path }}'+'csv',
  #     },
  #     dag=dag
  # )
  

  What is the best way to do this?

  I know how to do this using bash operator,but want to know if we can use hive operator

  hive_ex = BashOperator(
      task_id='hive-ex',
      bash_command='hive -f hive.sql -DAY={{ ds }} >> {{ file_path }} 
      /file_{{ds}}.json',
      dag=dag
  )