Membuat koneksi SSH menggunakan SSHOperator - Amazon Managed Workflows for Apache Airflow
VersiPrasyaratIzinPersyaratanSalin kunci rahasia Anda ke Amazon S3Buat koneksi Apache Airflow baruContoh kode

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Membuat koneksi SSH menggunakan SSHOperator

Contoh berikut menjelaskan bagaimana Anda dapat menggunakan SSHOperator dalam grafik asiklik terarah (DAG) untuk menyambung ke EC2 instans Amazon jarak jauh dari lingkungan Alur Kerja Terkelola Amazon untuk Apache Airflow. Anda dapat menggunakan pendekatan serupa untuk terhubung ke instans jarak jauh apa pun dengan akses SSH.

Dalam contoh berikut, Anda mengunggah kunci rahasia SSH (.pem) ke dags direktori lingkungan Anda di Amazon S3. Kemudian, Anda menginstal dependensi yang diperlukan menggunakan requirements.txt dan membuat koneksi Apache Airflow baru di UI. Terakhir, Anda menulis DAG yang membuat koneksi SSH ke instance jarak jauh.

Versi

Anda dapat menggunakan contoh kode pada halaman ini dengan Apache Airflow v2 di Python 3.10 dan Apache Airflowv3 di Python 3.11.

Prasyarat

Untuk menggunakan kode sampel di halaman ini, Anda memerlukan yang berikut:

  • Lingkungan Amazon MWAA.

  • Kunci rahasia SSH. Contoh kode mengasumsikan Anda memiliki EC2 instans Amazon dan .pem di Wilayah yang sama dengan lingkungan Amazon MWAA Anda. Jika Anda tidak memiliki kunci, lihat Membuat atau mengimpor key pair di Panduan EC2 Pengguna Amazon.

Izin

Tidak diperlukan izin tambahan untuk menggunakan contoh kode di halaman ini.

Persyaratan

Tambahkan parameter berikut requirements.txt untuk menginstal apache-airflow-providers-ssh paket di server web. Setelah lingkungan Anda diperbarui dan Amazon MWAA berhasil menginstal dependensi, Anda akan mendapatkan jenis koneksi SSH baru di UI.

-c https://raw.githubusercontent.com/apache/airflow/constraints-Airflow-version/constraints-Python-version.txt
apache-airflow-providers-ssh
catatan

-cmendefinisikan kendala URL di. requirements.txt Ini memastikan bahwa Amazon MWAA menginstal versi paket yang benar untuk lingkungan Anda.

Salin kunci rahasia Anda ke Amazon S3

Gunakan AWS Command Line Interface perintah berikut untuk menyalin .pem kunci Anda ke dags direktori lingkungan Anda di Amazon S3.

aws s3 cp your-secret-key.pem s3://amzn-s3-demo-bucket/dags/

Amazon MWAA menyalin kontendags, termasuk .pem kunci, ke /usr/local/airflow/dags/ direktori lokal, Dengan melakukan ini, Apache Airflow dapat mengakses kunci.

Buat koneksi Apache Airflow baru

Untuk membuat koneksi SSH baru menggunakan Apache Airflow UI

  1. Buka halaman Lingkungan di konsol Amazon MWAA.

  2. Dari daftar lingkungan, pilih Open Airflow UI untuk lingkungan Anda.

  3. Pada halaman Apache Airflow UI, pilih Admin dari bilah navigasi utama untuk memperluas daftar dropdown, lalu pilih Connections.

  4. Pada halaman Daftar Koneksi, pilih +, atau Tambahkan tombol catatan baru untuk menambahkan koneksi baru.

  5. Pada halaman Add Connection, tambahkan informasi berikut:

    1. Untuk Connection Id, masukkanssh_new.

    2. Untuk Jenis Koneksi, pilih SSH dari daftar dropdown.

      catatan

      Jika jenis koneksi SSH tidak tersedia dalam daftar, Amazon MWAA belum menginstal paket yang diperlukan. apache-airflow-providers-ssh Perbarui requirements.txt file Anda untuk menyertakan paket ini, lalu coba lagi.

    3. Untuk Host, masukkan alamat IP untuk EC2 instance Amazon yang ingin Anda sambungkan. Misalnya, 12.345.67.89.

    4. Untuk Nama Pengguna, masukkan ec2-user jika Anda terhubung ke EC2 instans Amazon. Nama pengguna Anda mungkin berbeda, tergantung pada jenis instance jarak jauh yang ingin Anda sambungkan Apache Airflow.

    5. Untuk Ekstra, masukkan pasangan kunci-nilai berikut dalam format JSON:

      { "key_file": "/usr/local/airflow/dags/your-secret-key.pem" }

      Pasangan kunci-nilai ini menginstruksikan Apache Airflow untuk mencari kunci rahasia di direktori lokal. /dags

Contoh kode

DAG berikut menggunakan SSHOperator untuk menyambung ke EC2 instance Amazon target Anda, lalu menjalankan perintah hostname Linux untuk mencetak nama instance. Anda dapat memodifikasi DAG untuk menjalankan perintah atau skrip apa pun pada instance jarak jauh.

  1. Buka terminal, dan arahkan ke direktori tempat kode DAG Anda disimpan. Misalnya:

    cd dags

  2. Salin isi contoh kode berikut dan simpan secara lokal sebagaissh.py.

    from airflow.decorators import dag
from datetime import datetime
from airflow.providers.ssh.operators.ssh import SSHOperator

@dag(
    dag_id="ssh_operator_example",
    schedule_interval=None,     
    start_date=datetime(2022, 1, 1),
    catchup=False,
    )
def ssh_dag():
    task_1=SSHOperator(
        task_id="ssh_task",
        ssh_conn_id='ssh_new',
        command='hostname',
    )

my_ssh_dag = ssh_dag()

  3. Jalankan AWS CLI perintah berikut untuk menyalin DAG ke bucket lingkungan Anda, lalu picu DAG menggunakan Apache Airflow UI. 

    aws s3 cp your-dag.py s3://your-environment-bucket/dags/

  4. Jika berhasil, Anda akan mendapatkan output yang mirip dengan yang berikut di log tugas ssh_task di ssh_operator_example DAG:

    [2022-01-01, 12:00:00 UTC] {{base.py:79}} INFO - Using connection to: id: ssh_new. Host: 12.345.67.89, Port: None,
Schema: , Login: ec2-user, Password: None, extra: {'key_file': '/usr/local/airflow/dags/your-secret-key.pem'}
[2022-01-01, 12:00:00 UTC] {{ssh.py:264}} WARNING - Remote Identification Change is not verified. This won't protect against Man-In-The-Middle attacks [2022-01-01, 12:00:00 UTC] {{ssh.py:270}} WARNING - No Host Key Verification. This won't protect against Man-In-The-Middle attacks 
[2022-01-01, 12:00:00 UTC] {{transport.py:1819}} INFO - Connected (version 2.0, client OpenSSH_7.4) 
[2022-01-01, 12:00:00 UTC] {{transport.py:1819}} INFO - Authentication (publickey) successful!
[2022-01-01, 12:00:00 UTC] {{ssh.py:139}} INFO - Running command: hostname
[2022-01-01, 12:00:00 UTC]{{ssh.py:171}} INFO - ip-123-45-67-89.us-west-2.compute.internal
[2022-01-01, 12:00:00 UTC] {{taskinstance.py:1280}} INFO - Marking task as SUCCESS. dag_id=ssh_operator_example, task_id=ssh_task, execution_date=20220712T200914, start_date=20220712T200915, end_date=20220712T200916