pip install apache-airflow-providers-google
创建一个Google Cloud项目,并记录下项目ID、服务账号的JSON文件以及GCP区域。
在Airflow中创建一个Connections。打开Airflow Web UI并点击菜单栏中的“Admin”->“Connections”。创建一个新的Connection并填写以下信息:
Conn Id: google_cloud_default
Conn Type: Google Cloud Platform
Project Id:
Keyfile Path:
Region:
from airflow import DAG
from airflow.contrib.hooks.gcs_hook import GoogleCloudStorageHook
from airflow.utils.dates import days_ago
default_args = {
'start_date': days_ago(1),
}
with DAG('google_cloud_dag', default_args=default_args, schedule_interval='@daily') as dag:
gcs_hook = GoogleCloudStorageHook(google_cloud_storage_conn_id='google_cloud_default')
download_task = gcs_hook.download(bucket_name='my_bucket', object_name='my_object', filename='local_file')
通过设置google_cloud_storage_conn_id
参数为我们创建的Connection ID,可以使用Google Cloud服务进行操作。