如何使用 BashOprator 在 Airflow 中使用 Airflow AWS 连接凭证将文件从 AWS s3 存储桶传输到 GCS

Question

当我使用两个云时，我的任务是将进入 s3 存储桶的文件同步到 gcs 存储桶。为了实现这一点，我正在使用 GCP composer (Airflow) 服务来安排这个 rsync 操作来同步文件。我正在使用 Airflow 连接 (aws_default) 来存储 AWS 访问密钥和秘密访问密钥。一切正常，但问题是我能够在日志中看到凭据，这再次暴露了凭据，我什至不想在日志中显示它们。如果有任何方法可以使凭据不显示在日志中，请提供帮助。

import airflow
import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.hooks.base_hook import BaseHook
from datetime import timedelta, datetime

START_TIME = datetime.utcnow() - timedelta(hours=1)

default_args = {
    'owner': 'airflow',
    'depends_on_past': True,
    'wait_for_downstream': True,
    'start_date': START_TIME,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=3)
}

aws_connection = BaseHook.get_connection('aws_default')

bash_env = {
        "AWS_ACCESS_KEY_ID": aws_connection.login,
        "AWS_SECRET_ACCESS_KEY": aws_connection.password
}

rsync_command = ''' 
    set -e; 
    export AWS_ACCESS_KEY_ID="%s"; 
    export AWS_SECRET_ACCESS_KEY="%s"; 
''' %(bash_env.get('AWS_ACCESS_KEY_ID'), bash_env.get('AWS_SECRET_ACCESS_KEY')) \
+  '''
    gsutil -m rsync -r -n s3://aws_bucket/{{ execution_date.strftime('%Y/%m/%d/%H') }}/ gs://gcp_bucket/good/test/
'''

dag = DAG(
    'rsync',
    default_args=default_args,
    description='This dag is for gsutil rsync from s3 buket to gcs storage',
    schedule_interval=timedelta(minutes=20),
    dagrun_timeout=timedelta(minutes=15)
    )


s3_sync = BashOperator(
    task_id='gsutil_s3_gcp_sync',
    bash_command=rsync_command,
    dag=dag,
    depends_on_past=False,
    execution_timeout=timedelta(hours=1),
    )

Answer 1

我建议将凭据放在与 Airflow 分开的 boto 配置文件中。更多配置文件here

它有一个凭证部分：

[Credentials]
  aws_access_key_id
  aws_secret_access_key
  gs_access_key_id
  gs_host
  gs_host_header
  gs_json_host
  gs_json_host_header
  gs_json_port
  gs_oauth2_refresh_token
  gs_port
  gs_secret_access_key
  gs_service_client_id
  gs_service_key_file
  gs_service_key_file_password
  s3_host
  s3_host_header
  s3_port