Pemecahan masalah: DAG, Operator, Koneksi, dan masalah lainnya

Pelajari cara membuat kunci rahasia untuk koneksi dan variabel Apache Airflow Anda diMengkonfigurasi koneksi Apache Airflow menggunakan AWS Secrets Manager Rahasia.

Pelajari cara menggunakan kunci rahasia untuk variabel Apache Airflow (test-variable) diMenggunakan kunci rahasia AWS Secrets Manager untuk variabel Apache Airflow.

Pelajari cara menggunakan kunci rahasia untuk koneksi Apache Airflow (myconn) diMenggunakan kunci rahasia AWS Secrets Manager untuk koneksi Apache Airflow.

Uji DAG, plugin khusus, dan dependensi Python Anda secara lokal menggunakan aws- mwaa-docker-images on. GitHub

Tambahkan entri berikut ke requirements.txt untuk lingkungan Anda.

apache-airflow-providers-snowflake==1.3.0

Tambahkan impor berikut ke DAG Anda:

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

Id Konten: snowflake_conn

Jenis Konn: Kepingan Salju

Tuan rumah: <my account>. <my region if not us-west-2>.snowflakecomputing.com

Skema: <my schema>

Masuk: <my user name>

Kata Sandi: ********

Pelabuhan: <port, if any>

Tambahan:

{
    "account": "<my account>",
    "warehouse": "<my warehouse>",
    "database": "<my database>",
    "region": "<my region if not using us-west-2 otherwise omit this line>"
}

Akses jenis koneksi yang disediakan Amazon MWAA di Apache Airflow UI diPaket penyedia Apache Airflow diinstal di lingkungan Amazon MWAA.

Akses perintah untuk membuat koneksi Apache Airflow di CLI diReferensi perintah Apache Airflow CLI.

Pelajari cara menggunakan templat koneksi di UI Apache Airflow secara bergantian untuk jenis koneksi yang tidak tersedia di UI Apache Airflow di Amazon MWAA di. Ikhtisar jenis koneksi

Periksa opsi konfigurasi Apache Airflow. Pastikan pasangan kunci-nilai yang Anda tentukan sebagai opsi konfigurasi Apache Airflow, seperti AWS Secrets Manager, telah dikonfigurasi dengan benar. Untuk mempelajari lebih lanjut, lihatSaya tidak dapat terhubung ke Manajer Rahasia.

Periksarequirements.txt. Verifikasi paket “ekstra” Airflow dan pustaka lain yang tercantum dalam paket requirements.txt Anda kompatibel dengan versi Apache Airflow Anda.

Jelajahi cara untuk menentukan dependensi Python dalam requirements.txt file, lihat. Mengelola dependensi Python di requirements.txt

Konfirmasikan bahwa grup keamanan VPC Anda mengizinkan akses masuk ke port. 5432 Port ini diperlukan untuk terhubung ke database metadata Amazon Aurora PostgreSQL untuk lingkungan Anda. Setelah aturan ini ditambahkan, beri Amazon MWAA beberapa menit, dan kesalahan dapat hilang. Untuk mempelajari lebih lanjut, lihatKeamanan di VPC Anda di Amazon MWAA.

Jika penjadwal tidak berjalan, itu mungkin karena sejumlah faktor seperti kegagalan instalasi ketergantungan, atau penjadwal yang kelebihan beban. Konfirmasikan bahwa DAG, plugin, dan persyaratan Anda berfungsi dengan benar dengan mengakses grup log yang sesuai di CloudWatch Log. Untuk mempelajari lebih lanjut, lihatPemantauan dan metrik untuk Alur Kerja Terkelola Amazon untuk Apache Airflow.

Mungkin ada sejumlah besar DAG yang didefinisikan. Kurangi jumlah DAG dan lakukan pembaruan lingkungan (seperti mengubah level log) untuk memaksa reset.

1. Aliran udara mengurai DAG apakah diaktifkan atau tidak. Jika Anda menggunakan lebih dari 50% kapasitas lingkungan Anda, Anda mungkin mulai membebani penjadwal Apache Airflow. Hal ini menyebabkan Total Waktu Penguraian yang besar dalam CloudWatch Metrik atau waktu pemrosesan DAG yang panjang di CloudWatch Log. Ada cara lain untuk mengoptimalkan konfigurasi Apache Airflow yang berada di luar cakupan panduan ini.

2. Untuk mempelajari lebih lanjut tentang praktik terbaik yang kami sarankan untuk menyesuaikan kinerja lingkungan Anda, lihatPenyetelan kinerja untuk Apache Airflow di Amazon MWAA.

Mungkin ada sejumlah besar tugas dalam antrian. Ini sering ditampilkan sebagai jumlah tugas yang besar—dan bertambah—di None negara bagian, atau sebagai jumlah besar di dalam. Queued Tasks and/or Tasks Pending CloudWatch Hal ini dapat terjadi karena alasan berikut:

1. Jika ada lebih banyak tugas untuk dijalankan daripada kapasitas lingkungan untuk dijalankan, and/or sejumlah besar tugas yang diantri sebelum penskalaan otomatis memiliki waktu untuk mendeteksi tugas dan menyebarkan pekerja tambahan.

2. Jika ada lebih banyak tugas yang harus dijalankan daripada kapasitas lingkungan untuk dijalankan, sebaiknya kurangi jumlah tugas yang dijalankan DAG Anda secara bersamaan, and/or meningkatkan minimum pekerja Apache Airflow.

3. Jika ada sejumlah besar tugas yang diantri sebelum penskalaan otomatis sempat mendeteksi dan menyebarkan pekerja tambahan, kami sarankan penerapan tugas yang mengejutkan and/or meningkatkan minimum pekerja Apache Airflow.

4. Anda dapat menggunakan perintah update-environment di AWS Command Line Interface (AWS CLI) untuk mengubah jumlah minimum atau maksimum pekerja yang berjalan di lingkungan Anda.

   aws mwaa update-environment --name MyEnvironmentName --min-workers 2 --max-workers 10

5. Untuk mempelajari lebih lanjut tentang praktik terbaik yang kami sarankan untuk menyesuaikan kinerja lingkungan Anda, lihatPenyetelan kinerja untuk Apache Airflow di Amazon MWAA.

Jika tugas Anda macet dalam status “berjalan”, Anda juga dapat menghapus tugas atau menandainya sebagai berhasil atau gagal. Hal ini memungkinkan komponen penskalaan otomatis untuk lingkungan Anda untuk mengurangi jumlah pekerja yang berjalan di lingkungan Anda. Gambar berikut menggambarkan contoh tugas yang terdampar.

1. Pilih lingkaran untuk tugas yang terdampar, lalu pilih H apus (seperti yang ditunjukkan). Ini memungkinkan Amazon MWAA menurunkan skala pekerja; jika tidak, Amazon MWAA tidak dapat menentukan DAG mana yang diaktifkan atau dinonaktifkan, dan tidak dapat mengurangi skala, jika masih ada tugas yang antri.

   

Pelajari lebih lanjut tentang siklus hidup tugas Apache Airflow di Konsep dalam panduan referensi Apache Airflow.

Uji DAG, plugin khusus, dan dependensi Python Anda secara lokal menggunakan aws- mwaa-docker-images on. GitHub

Buat skrip “transformasi” Anda.

#!/bin/bash
cp $1 $2

(opsional) Pengguna macOS dan Linux mungkin perlu menjalankan perintah berikut untuk memastikan skrip dapat dieksekusi.

chmod 777 transform_test.sh

Tambahkan skrip ke plugins.zip Anda.

zip plugins.zip transform_test.sh

Ikuti langkah-langkah di Unggah plugins.zip ke Amazon S3.

Ikuti langkah-langkah dalam Men entukan versi plugins.zip di konsol Amazon MWAA.

Buat DAG berikut.

from airflow import DAG
from airflow.providers.amazon.aws.operators.s3_file_transform import S3FileTransformOperator
from airflow.utils.dates import days_ago
import os

DAG_ID = os.path.basename(__file__).replace(".py", "")

with DAG (dag_id=DAG_ID, schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
    file_transform = S3FileTransformOperator(
        task_id='file_transform',
        transform_script='/usr/local/airflow/plugins/transform_test.sh',
        source_s3_key='s3://amzn-s3-demo-bucket/files/input.txt',
        dest_s3_key='s3://amzn-s3-demo-bucket/files/output.txt'
    )

Ikuti langkah-langkah di Meng unggah kode DAG ke Amazon S3.