title: Google Cloud Composer, airflow job cannot recognize installed PyPi packages
body: <p>I am working on Airflow using Google Cloud Composer. Here is the dag file:</p>
  
  <pre><code>from airflow import DAG
  from airflow.operators.bash_operator import BashOperator
  from datetime import datetime, timedelta
  
  dag = DAG(
      'hello_world',
      description='Simple DAG',
      start_date=datetime.now() - timedelta(days=1),
      schedule_interval='@once'
  )
  
  hello = BashOperator(
      task_id='hello_world',
      bash_command='python3 /home/airflow/gcs/dags/dependencies/helper.py',
      dag=dag
  )
  </code></pre>
  
  <p>It basically runs <code>helper.py</code> inside folder <code>/dags/dependencies/</code> which is located in Google Cloud Storage, the DAG bag directory.</p>
  
  <p><code>helper.py</code> contains the following code:</p>
  
  <pre><code>from fastavro import writer
  import io
  import logging
  
  
  def greetings():
      buffer = io.BytesIO()
      age = 24
      schema = {
          'doc': "cockroach",
          'name': "table",
          'namespace': "cockroach",
          'type': "record",
          'fields': [{'name': 'age', 'type': ['null', 'int']}]
      }
      writer(buffer, schema=schema, records=[{"age": 24}])
      logging.info("Hello {}".format(name))
      return "Hello {}".format(name)
  </code></pre>
  
  <p>It raises an error <code>ModuleNotFoundError: No module named 'fastavro'</code>:</p>
  
  <pre><code>[2019-01-11 04:01:57,388] {base_task_runner.py:98} INFO - Subtask: [2019-01-11 04:01:57,386] {bash_operator.py:101} INFO - Traceback (most recent call last):
  [2019-01-11 04:01:57,389] {base_task_runner.py:98} INFO - Subtask: [2019-01-11 04:01:57,388] {bash_operator.py:101} INFO - File "/home/airflow/gcs/dags/dependencies/helper.py", line 1, in &lt;module&gt;
  [2019-01-11 04:01:57,389] {base_task_runner.py:98} INFO - Subtask: [2019-01-11 04:01:57,388] {bash_operator.py:101} INFO - from fastavro import writer
  [2019-01-11 04:01:57,390] {base_task_runner.py:98} INFO - Subtask: [2019-01-11 04:01:57,389] {bash_operator.py:101} INFO - ModuleNotFoundError: No module named 'fastavro'
  [2019-01-11 04:01:58,154] {base_task_runner.py:98} INFO - Subtask: [2019-01-11 04:01:58,152] {bash_operator.py:105} INFO - Command exited with return code 1
  [2019-01-11 04:01:58,214] {base_task_runner.py:98} INFO - Subtask: Traceback (most recent call last):
  [2019-01-11 04:01:58,214] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/bin/airflow", line 6, in &lt;module&gt;
  [2019-01-11 04:01:58,214] {base_task_runner.py:98} INFO - Subtask:     exec(compile(open(__file__).read(), __file__, 'exec'))
  [2019-01-11 04:01:58,215] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/airflow/airflow/bin/airflow", line 27, in &lt;module&gt;
  [2019-01-11 04:01:58,215] {base_task_runner.py:98} INFO - Subtask:     args.func(args)
  [2019-01-11 04:01:58,215] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/airflow/airflow/bin/cli.py", line 392, in run
  [2019-01-11 04:01:58,215] {base_task_runner.py:98} INFO - Subtask:     pool=args.pool,
  [2019-01-11 04:01:58,215] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/airflow/airflow/utils/db.py", line 50, in wrapper
  [2019-01-11 04:01:58,216] {base_task_runner.py:98} INFO - Subtask:     result = func(*args, **kwargs)
  [2019-01-11 04:01:58,216] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/airflow/airflow/models.py", line 1492, in _run_raw_task
  [2019-01-11 04:01:58,216] {base_task_runner.py:98} INFO - Subtask:     result = task_copy.execute(context=context)
  [2019-01-11 04:01:58,219] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/airflow/airflow/operators/bash_operator.py", line 109, in execute
  [2019-01-11 04:01:58,219] {base_task_runner.py:98} INFO - Subtask:     raise AirflowException("Bash command failed")
  [2019-01-11 04:01:58,220] {base_task_runner.py:98} INFO - Subtask: airflow.exceptions.AirflowException: Bash command failed
  </code></pre>
  
  <p>Although, I have installed the <code>fastavro</code> in PyPi packages from Google Composer Environment:
  <a href="https://i.stack.imgur.com/rjT6e.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/rjT6e.png" alt="enter image description here"></a></p>
  
  <p>Anybody knows how to solve it?</p>