Как использовать учетные данные подключения Airflow AWS в Airflow с помощью BashOprator для передачи файлов из корзины AWS s3 в GCS - PullRequest
Новая
       14

Как использовать учетные данные подключения Airflow AWS в Airflow с помощью BashOprator для передачи файлов из корзины AWS s3 в GCS

0 голосов
/ 24 апреля 2020

Поскольку я работаю с двумя облаками, моя задача - rsyn c файлы, поступающие в корзину s3 в корзину gcs. Для этого я использую сервис GCP composer (Airflow), где я планирую эту операцию rsyn c для синхронизации c файлов. Я использую соединение 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),
    )

1 Ответ

1 голос
/ 25 апреля 2020

Я бы предложил поместить учетные данные в конфигурационный файл boto отдельно от Airflow. Больше информации о конфигурационном файле здесь

Имеет раздел учетных данных: 

[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
...