对 Airflow/Python 等非常陌生，但似乎无法弄清楚我需要做什么来解决这个问题..
Airflow 在 Puckel/Docker 上运行
完整的错误是：
Broken DAG : [/usr/local/airflow/dags/xxxx.py] No module named 'airflow.contrib.operators.gsc_to_gcs'
在python代码中，我写了：
from airflow.contrib.operators.gcs_to_gcs import GoogleCloudStorageToGoogleCloudStorageOperator
我猜我需要安装gcs_to_gcs 模块，但我不知道该怎么做。
任何具体说明将不胜感激:-)
【问题讨论】：
gcs_to_gcs 运算符在 puckel docker 映像正在使用的气流版本中尚不存在。
GoogleCloudStorageToGoogleCloudStorageOperator 在 v1.9.0 中不可用，因此您必须从 here 复制文件并从 here 复制相关挂钩，并将其粘贴到 Python 环境中相应位置的 Airflow 文件夹中.请按照以下步骤操作：
运行以下代码以查找 Apache Airflow 在您的计算机上的存储位置：
pip show apache-airflow
这应该在您的终端上产生以下输出：
Name: apache-airflow
Version: 2.0.0.dev0+incubating
Summary: Programmatically author, schedule and monitor data pipelines
Home-page: http://airflow.incubator.apache.org/
Author: Apache Software Foundation
Author-email: dev@airflow.incubator.apache.org
License: Apache License 2.0
Location: /Users/kaxil/anaconda2/lib/python2.7/site-packages
Requires: iso8601, bleach, gunicorn, sqlalchemy-utc, markdown, flask-caching, alembic, croniter, flask-wtf, requests, tabulate, psutil, jinja2, gitpython, python-nvd3, sqlalchemy, dill, flask, pandas, pendulum, flask-login, funcsigs, flask-swagger, flask-admin, lxml, python-dateutil, pygments, werkzeug, tzlocal, python-daemon, setproctitle, zope.deprecation, flask-appbuilder, future, configparser, thrift
Required-by:
Location: 之后的路径是您的 Apache Airflow 目录
现在克隆 git repo 以获取这两个文件：
# Clone the git repo to `airflow-temp` folder
git clone https://github.com/apache/incubator-airflow airflow-temp
# Copy the hook from the cloned repo to where Apache Airflow is located
# Replace LINK_TO_SITE_PACKAGES_DIR with the path you found above
cp airflow-temp/airflow/contrib/hooks/gcs_hook.py LINK_TO_SITE_PACKAGES_DIR/airflow/contrib/hooks/
# For example: for me, it would be
cp airflow-temp/airflow/contrib/hooks/gcs_hook.py /Users/kaxil/anaconda2/lib/python2.7/site-packages/airflow/contrib/hooks/
# Do the same with operator file
cp airflow-temp/airflow/contrib/operators/gcs_to_gcs.py LINK_TO_SITE_PACKAGES_DIR/airflow/contrib/operators/
# For example: for me, it would be
cp airflow-temp/airflow/contrib/operators/gcs_to_gcs.py /Users/kaxil/anaconda2/lib/python2.7/site-packages/airflow/contrib/operators/
重新运行 Airflow webserver 和 scheduler，现在应该可以工作了。
【讨论】：
我知道这是一个老问题，但我只是尝试使用相同的运算符并收到相同的消息，因为 Cloud-Composer 仍然不支持 GoogleCloudStorageToGoogleCloudStorageOperator。
我设法通过使用简单的 BashOperator 的解决方法实现了我所需要的
from airflow.operators.bash_operator import BashOperator
with models.DAG(
dag_name,
schedule_interval=timedelta(days=1),
default_args=default_dag_args) as dag:
copy_files = BashOperator(
task_id='copy_files',
bash_command='gsutil -m cp <Source Bucket> <Destination Bucket>'
)
非常简单，可以根据需要创建文件夹并重命名文件。
【讨论】：