Saya mencoba merasakan file dengan DAG aliran udara, tetapi filesensor saya selalu terjebak pada status antrian. Saya telah mencoba dengan sampel kode di bawah ini. Apakah ada yang saya lewatkan? BTW, versi aliran udara saya adalah 2.0.1. from airflow.contrib.sensors.file_sensor import FileSensor fr....
Saat ini saya menggunakan Airflow Task Flow API 2.0. Saya memiliki masalah menggabungkan penggunaan tugas taskgroup dan branchpythonoperator. Di bawah ini adalah kode saya: import airflow from airflow.models import DAG from airflow.decorators import task, dag from airflow.operators.dummy_operator im....
27 Mei 2021, 13:38
Saya telah mengatur DAG dengan parameter berikut local_tz = pendulum.timezone('US/Eastern') default_args = { 'retries': 3, 'retry_delay': timedelta(minutes=5) } dag = DAG( dag_id='some_dag', start_date=datetime(2021, 1, 8, tzinfo=local_tz), schedule_interval='0 16 8 * *', ....
26 Mei 2021, 04:40
Saya mencoba mengimplementasikan SLA dalam DAG aliran udara saya. Saya tahu bagaimana cara kerja SL, Anda menetapkan objek Timedelta dan jika tugasnya tidak dilakukan dalam durasi itu, itu akan mengirim email dan memberi tahu bahwa tugas belum dilakukan. Saya ingin beberapa fungsi yang sama, tetapi ....
Saya memiliki cluster aliran udara yang terbuat dari 3 node pekerja dengan selysexecutor dan rabbitmq untuk komunikasi. Danggs saya biasanya terdiri dari tugas-tugas yang mengunduh file, unzip mereka, mengunggahnya ke Hadoop, dll. Jadi mereka bergantung satu sama lain dan harus berjalan pada mesin t....