from airflow import DAG
from datetime import datetime, timedelta
from operators.emr_spark_operator import EMRSparkOperator
from airflow.operators import BashOperator

default_args = {
    'owner': 'mreid@mozilla.com',
    'depends_on_past': False,
    'start_date': datetime(2016, 6, 27),
    'email': ['telemetry-alerts@mozilla.com', 'mreid@mozilla.com'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=30),
}

dag = DAG('main_summary', default_args=default_args, schedule_interval='@daily')

# Make sure all the data for the given day has arrived before running.
t0 = BashOperator(task_id="delayed_start",
                  bash_command="sleep 1800",
                  dag=dag)

t1 = EMRSparkOperator(task_id="main_summary",
                      job_name="Main Summary View",
                      execution_timeout=timedelta(hours=10),
                      instance_count=10,
                      env = {"date": "{{ ds_nodash }}", "bucket": "{{ task.__class__.private_output_bucket }}"},
                      uri="https://raw.githubusercontent.com/mozilla/telemetry-airflow/master/jobs/main_summary_view.sh",
                      dag=dag)

# Wait a little while after midnight to start for a given day.
t1.set_upstream(t0)

from airflow import DAG
from datetime import datetime, timedelta
from operators.emr_spark_operator import EMRSparkOperator
from airflow.operators import BashOperator

default_args = {
    'owner': 'mreid@mozilla.com',
    'depends_on_past': False,
    'start_date': datetime(2016, 6, 25),
    'email': ['telemetry-alerts@mozilla.com', 'mreid@mozilla.com'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=30),
}

dag = DAG('main_summary', default_args=default_args, schedule_interval='@daily', max_active_runs=10)

# Make sure all the data for the given day has arrived before running.
t0 = BashOperator(task_id="delayed_start",
                  bash_command="sleep 1800",
                  dag=dag)

t1 = EMRSparkOperator(task_id="main_summary",
                      job_name="Main Summary View",
                      execution_timeout=timedelta(hours=10),
                      instance_count=10,
                      env={"date": "{{ ds_nodash }}", "bucket": "{{ task.__class__.private_output_bucket }}"},
                      uri="https://raw.githubusercontent.com/mozilla/telemetry-airflow/master/jobs/main_summary_view.sh",
                      dag=dag)

# Wait a little while after midnight to start for a given day.
t1.set_upstream(t0)

Prepare "Main Summary" job for backfill
Prepare "Main Summary" job for backfill

Set the max number of active runs so we don't overwhelm the system,
and rewind the start date by a couple of days to test that the
scheduler does the right thing.

  from airflow import DAG
  from datetime import datetime, timedelta
  from operators.emr_spark_operator import EMRSparkOperator
  from airflow.operators import BashOperator
  
  default_args = {
      'owner': 'mreid@mozilla.com',
      'depends_on_past': False,
-     'start_date': datetime(2016, 6, 27),
+     'start_date': datetime(2016, 6, 25),
      'email': ['telemetry-alerts@mozilla.com', 'mreid@mozilla.com'],
      'email_on_failure': True,
      'email_on_retry': True,
      'retries': 2,
      'retry_delay': timedelta(minutes=30),
  }
  
- dag = DAG('main_summary', default_args=default_args, schedule_interval='@daily')
+ dag = DAG('main_summary', default_args=default_args, schedule_interval='@daily', max_active_runs=10)
  
  # Make sure all the data for the given day has arrived before running.
  t0 = BashOperator(task_id="delayed_start",
                    bash_command="sleep 1800",
                    dag=dag)
  
  t1 = EMRSparkOperator(task_id="main_summary",
                        job_name="Main Summary View",
                        execution_timeout=timedelta(hours=10),
                        instance_count=10,
-                       env = {"date": "{{ ds_nodash }}", "bucket": "{{ task.__class__.private_output_bucket }}"},
+                       env={"date": "{{ ds_nodash }}", "bucket": "{{ task.__class__.private_output_bucket }}"},
                        uri="https://raw.githubusercontent.com/mozilla/telemetry-airflow/master/jobs/main_summary_view.sh",
                        dag=dag)
  
  # Wait a little while after midnight to start for a given day.
  t1.set_upstream(t0)
  
Prepare "Main Summary" job for backfill
## Code Before:
from airflow import DAG
from datetime import datetime, timedelta
from operators.emr_spark_operator import EMRSparkOperator
from airflow.operators import BashOperator

default_args = {
    'owner': 'mreid@mozilla.com',
    'depends_on_past': False,
    'start_date': datetime(2016, 6, 27),
    'email': ['telemetry-alerts@mozilla.com', 'mreid@mozilla.com'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=30),
}

dag = DAG('main_summary', default_args=default_args, schedule_interval='@daily')

# Make sure all the data for the given day has arrived before running.
t0 = BashOperator(task_id="delayed_start",
                  bash_command="sleep 1800",
                  dag=dag)

t1 = EMRSparkOperator(task_id="main_summary",
                      job_name="Main Summary View",
                      execution_timeout=timedelta(hours=10),
                      instance_count=10,
                      env = {"date": "{{ ds_nodash }}", "bucket": "{{ task.__class__.private_output_bucket }}"},
                      uri="https://raw.githubusercontent.com/mozilla/telemetry-airflow/master/jobs/main_summary_view.sh",
                      dag=dag)

# Wait a little while after midnight to start for a given day.
t1.set_upstream(t0)

## Instruction:
Prepare "Main Summary" job for backfill
## Code After:
from airflow import DAG
from datetime import datetime, timedelta
from operators.emr_spark_operator import EMRSparkOperator
from airflow.operators import BashOperator

default_args = {
    'owner': 'mreid@mozilla.com',
    'depends_on_past': False,
    'start_date': datetime(2016, 6, 25),
    'email': ['telemetry-alerts@mozilla.com', 'mreid@mozilla.com'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=30),
}

dag = DAG('main_summary', default_args=default_args, schedule_interval='@daily', max_active_runs=10)

# Make sure all the data for the given day has arrived before running.
t0 = BashOperator(task_id="delayed_start",
                  bash_command="sleep 1800",
                  dag=dag)

t1 = EMRSparkOperator(task_id="main_summary",
                      job_name="Main Summary View",
                      execution_timeout=timedelta(hours=10),
                      instance_count=10,
                      env={"date": "{{ ds_nodash }}", "bucket": "{{ task.__class__.private_output_bucket }}"},
                      uri="https://raw.githubusercontent.com/mozilla/telemetry-airflow/master/jobs/main_summary_view.sh",
                      dag=dag)

# Wait a little while after midnight to start for a given day.
t1.set_upstream(t0)
