title: SparkSubmitOperator raises “dag_id could not be found" errors when run
body: <p>So I've got a Spark job to pull some domains from AWS, then three different jobs that each take said domains and extract various data from the sites. For some reason this workflow stalls at ImportS3CrawlData with the following error:</p>
  
  <pre><code>[2018-03-22 13:37:02,762] {models.py:1428} INFO - Executing &lt;Task(SparkSubmitOperator): ImportCrawlJob&gt; on 2018-03-22 13:37:00
  [2018-03-22 13:37:02,763] {base_task_runner.py:115} INFO - Running: ['bash', '-c', 'sudo -H -u hdfs airflow run dag_extract_jobs ImportCrawlJob 2018-03-22T13:37:00 --job_id 21 --raw -sd DAGS_FOLDER/run_extract_jobs.py --cfg_path /tmp/tmpir3e3r32']
  [2018-03-22 13:37:04,194] {base_task_runner.py:98} INFO - Subtask: [2018-03-22 13:37:04,193] {__init__.py:45} INFO - Using executor SequentialExecutor
  [2018-03-22 13:37:04,356] {base_task_runner.py:98} INFO - Subtask: [2018-03-22 13:37:04,356] {models.py:189} INFO - Filling up the DagBag from /home/airflow/airflow/dags/run_extract_jobs.py
  [2018-03-22 13:37:04,451] {base_task_runner.py:98} INFO - Subtask: Traceback (most recent call last):
  [2018-03-22 13:37:04,451] {base_task_runner.py:98} INFO - Subtask:   File "/usr/bin/airflow", line 27, in &lt;module&gt;
  [2018-03-22 13:37:04,451] {base_task_runner.py:98} INFO - Subtask:     args.func(args)
  [2018-03-22 13:37:04,452] {base_task_runner.py:98} INFO - Subtask:   File "/usr/lib/python3.5/site-packages/airflow/bin/cli.py", line 353, in run
  [2018-03-22 13:37:04,452] {base_task_runner.py:98} INFO - Subtask:     dag = get_dag(args)
  [2018-03-22 13:37:04,452] {base_task_runner.py:98} INFO - Subtask:   File "/usr/lib/python3.5/site-packages/airflow/bin/cli.py", line 130, in get_dag
  [2018-03-22 13:37:04,452] {base_task_runner.py:98} INFO - Subtask:     'parse.'.format(args.dag_id))
  [2018-03-22 13:37:04,452] {base_task_runner.py:98} INFO - Subtask: airflow.exceptions.AirflowException: dag_id could not be found: dag_extract_jobs. Either the dag did not exist or it failed to parse.
  </code></pre>
  
  <p>The code for run_extract_jobs.py can be found below, with sensitive/unnecessary bits removed.</p>
  
  <pre><code># Parameters to initialize Spark:
  access_id = Variable.get("AWS_ACCESS_KEY")
  bucket_name = 'cb-scrapinghub'
  secret_key = Variable.get("AWS_SECRET_KEY")
  timestamp = datetime.now().strftime("%Y-%m-%d-%H:%M:%S")
  
  
  default_args = {
      'owner': 'airflow',
      'depends_on_past': False,
      'retries': 1,
      'retry_delay': timedelta(minutes=5),
  }
  
  DAG = DAG(
      dag_id='dag_extract_jobs',
      description='Run Extract Jobs',
      schedule_interval='@once',
      start_date=datetime(2018, 1, 1),
      catchup=False,
      default_args=default_args,
  )
  
  # Spark Job that runs ImportS3CrawlData:
  importCrawlJob = SparkSubmitOperator(
      task_id='ImportCrawlJob',
      ...
      run_as_user='hdfs',
      dag=DAG,
  )
  
  # Spark Job that runs ExtractAboutText:
  extractAboutText = SparkSubmitOperator(
      task_id='ExtractAboutText',
      ...
      run_as_user='hdfs',
      dag=DAG
  )
  extractAboutText.set_upstream(importCrawlJob)
  
  # Spark Job that runs ExtractCompanyInfo:
  extractCompanyInfo = SparkSubmitOperator(
      task_id='ExtractCompanyInfo',
      ...
      run_as_user='hdfs',
      dag=DAG
  )
  extractCompanyInfo.set_upstream(importCrawlJob)
  
  # Spark Job that runs ExtractWebPeople:
  extractWebPeople = SparkSubmitOperator(
      task_id='ExtractWebPeople',
      ...
      run_as_user='hdfs',
      dag=DAG
  )
  extractWebPeople.set_upstream(importCrawlJob)
  </code></pre>
  
  <p>I've made sure Airflow and Spark are both up to date. My dag folder is set up correctly. Airflow can run the tutorial files just fine.</p>
  
  <p>I've been messing with this for days and am totally baffled. Thanks in advance for any help.</p>