dag14_crick_seqrun_transfer.py 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147
  1. from datetime import timedelta
  2. from airflow.models import DAG,Variable
  3. from airflow.utils.dates import days_ago
  4. from airflow.contrib.operators.ssh_operator import SSHOperator
  5. from airflow.contrib.hooks.ssh_hook import SSHHook
  6. from airflow.operators.bash_operator import BashOperator
  7. from airflow.operators.python_operator import PythonOperator
  8. from airflow.contrib.operators.sftp_operator import SFTPOperator
  9. from igf_airflow.utils.dag14_crick_seqrun_transfer_utils import check_and_transfer_run_func
  10. #from igf_airflow.utils.dag14_crick_seqrun_transfer_utils import extract_tar_file_func
  11. FTP_SEQRUN_SERVER = Variable.get('crick_ftp_seqrun_hostname')
  12. FTP_CONFIG_FILE = Variable.get('crick_ftp_config_file_wells')
  13. FTP_BASE_PATH = Variable.get('crick_ftp_base_path')
  14. SEQRUN_BASE_PATH = Variable.get('seqrun_base_path')
  15. HPC_SEQRUN_BASE_PATH = Variable.get('hpc_seqrun_path')
  16. args = {
  17. 'owner': 'airflow',
  18. 'start_date': days_ago(2),
  19. 'retries': 1,
  20. 'retry_delay': timedelta(minutes=5),
  21. 'provide_context': True,
  22. 'email_on_failure': False,
  23. 'email_on_retry': False,
  24. 'catchup': False,
  25. 'max_active_runs': 1,
  26. }
  27. ## SSH HOOK
  28. #orwell_ssh_hook = \
  29. # SSHHook(
  30. # key_file=Variable.get('hpc_ssh_key_file'),
  31. # username=Variable.get('hpc_user'),
  32. # remote_host=Variable.get('orwell_server_hostname'))
  33. dag = \
  34. DAG(
  35. dag_id='dag14_crick_seqrun_transfer',
  36. schedule_interval=None,
  37. default_args=args,
  38. tags=['ftp', 'hpc', 'orwell', 'wells'])
  39. with dag:
  40. ## TASK
  41. # not working on HPC
  42. check_and_transfer_run = \
  43. PythonOperator(
  44. task_id='check_and_transfer_run',
  45. dag=dag,
  46. pool='crick_ftp_pool',
  47. queue='hpc_4G',
  48. params={'ftp_seqrun_server': FTP_SEQRUN_SERVER,
  49. 'hpc_seqrun_base_path': HPC_SEQRUN_BASE_PATH,
  50. 'ftp_config_file': FTP_CONFIG_FILE},
  51. python_callable=check_and_transfer_run_func)
  52. ## TASK
  53. #extract_tar_file = \
  54. # PythonOperator(
  55. # task_id='extract_tar_file',
  56. # dag=dag,
  57. # queue='hpc_4G',
  58. # params={'hpc_seqrun_base_path': HPC_SEQRUN_BASE_PATH},
  59. # python_callable=extract_tar_file_func)
  60. #check_and_transfer_run = \
  61. # BashOperator(
  62. # task_id='check_and_transfer_run',
  63. # dag=dag,
  64. # pool='crick_ftp_pool',
  65. # do_xcom_push=False,
  66. # queue='wells',
  67. # params={'ftp_seqrun_server': FTP_SEQRUN_SERVER,
  68. # 'seqrun_base_path': HPC_SEQRUN_BASE_PATH,
  69. # 'ftp_config_file': FTP_CONFIG_FILE},
  70. # bash_command="""
  71. # python /rds/general/user/igf/home/data2/airflow_test/github/data-management-python/scripts/ftp_seqrun_transfer/transfer_seqrun_from_crick.py \
  72. # -f {{ params.ftp_seqrun_server }} \
  73. # -s {{ dag_run.conf["seqrun_id"] }} \
  74. # -d {{ params.seqrun_base_path }} \
  75. # -c {{ params.ftp_config_file }}
  76. # """)
  77. #check_and_transfer_run = \
  78. # SFTPOperator(
  79. # task_id='check_and_transfer_run',
  80. # dag=dag,
  81. # pool='crick_ftp_pool',
  82. # queue='hpc_4G',
  83. # ssh_conn_id="crick_sftp_conn",
  84. # params={'seqrun_base_path': HPC_SEQRUN_BASE_PATH,
  85. # 'ftp_base_path': FTP_BASE_PATH},
  86. # local_filepath="{{ params.seqrun_base_path }}/{{ dag_run.conf['seqrun_id'] }}.tar.gz",
  87. # remote_filepath="{{ params.ftp_base_path }}/{{ dag_run.conf['seqrun_id'] }}.tar.gz",
  88. # operation='get'
  89. # )
  90. # TASK
  91. extract_tar_file = \
  92. BashOperator(
  93. task_id='extract_tar_file',
  94. dag=dag,
  95. do_xcom_push=False,
  96. queue='hpc_4G',
  97. params={'seqrun_base_path': HPC_SEQRUN_BASE_PATH},
  98. bash_command="""
  99. cd {{ params.seqrun_base_path }};
  100. if [ -d temp_{{ dag_run.conf["seqrun_id"] }} ];
  101. then
  102. echo "Seqrun dir exists";
  103. exit 1;
  104. else
  105. mkdir -p temp_{{ dag_run.conf["seqrun_id"] }};
  106. tar \
  107. --no-same-owner \
  108. --no-same-permissions \
  109. --owner=igf \
  110. -xzf {{ dag_run.conf["seqrun_id"] }}.tar.gz \
  111. -C temp_{{ dag_run.conf["seqrun_id"] }};
  112. find temp_{{ dag_run.conf["seqrun_id"] }} \
  113. -type d \
  114. -exec chmod 700 {} \;
  115. chmod -R u+r temp_{{ dag_run.conf["seqrun_id"] }};
  116. chmod -R u+w temp_{{ dag_run.conf["seqrun_id"] }};
  117. fi
  118. """
  119. )
  120. ## TASK
  121. move_seqrun_dir = \
  122. BashOperator(
  123. task_id='move_seqrun_dir',
  124. dag=dag,
  125. do_xcom_push=False,
  126. queue='hpc_4G',
  127. params={'seqrun_base_path': SEQRUN_BASE_PATH},
  128. bash_command="""
  129. cd {{ params.seqrun_base_path }};
  130. if [ -d {{ dag_run.conf["seqrun_id"] }} ];
  131. then
  132. echo "Seqrun dir exists";
  133. exit 1;
  134. fi
  135. ls temp_{{ dag_run.conf["seqrun_id"] }}/camp/stp/sequencing/inputs/instruments/sequencers/{{ dag_run.conf["seqrun_id"] }};
  136. move temp_{{ dag_run.conf["seqrun_id"] }}/camp/stp/sequencing/inputs/instruments/sequencers/{{ dag_run.conf["seqrun_id"] }} {{ params.seqrun_base_path }}/{{ dag_run.conf["seqrun_id"] }};
  137. """
  138. )
  139. ## PIPELINE
  140. check_and_transfer_run >> extract_tar_file >> move_seqrun_dir