Operator Airflow

Pengantar Apache Airflow dengan Python

Mike Metzger

Data Engineer

Operator

Mewakili satu task dalam workflow
Berjalan mandiri (biasanya)
Umumnya tidak berbagi informasi
Berbagai operator untuk tugas berbeda

@dag(
  dag_id="Example_Dag"
)
def example_dag():
  @task
  def task1():
    return "The result from task1"

  task1()

example_dag()

@task (PythonOperator)

Menjalankan fungsi Python
Fungsi Python apa pun bisa diberi dekorator @task
Dapat mengoper data antar fungsi/task

from airflow.sdk import task

@task
def printme():
    print("This goes in the logs!")

printme()

Argumen @task

Dapat meneruskan argumen ke task/fungsi seperti fungsi Python biasa

@task
def printme(name: str):
    print(f"Hi {name} - This goes in the logs!")


printme(name='DataCamp')


# Menambahkan:
#   Hi DataCamp - This goes in the logs!
# ke log Airflow

@task.bash (BashOperator)

@task.bash
def bash_example():
  return "echo 'Example!'"

bash_example()

@task.bash
def run_cleanup():
  return "runcleanup.sh"

run_cleanup()

Menjalankan perintah atau skrip Bash
Menjalankan perintah di direktori sementara
Dapat menetapkan variabel lingkungan untuk perintah

Dependensi task

Setiap Dag memiliki serangkaian task untuk diselesaikan
Dependensi task menentukan urutan jalan
Ada beberapa cara menetapkan dependensi

Airflow Dag of connected tasks showing dependency order between them

Sintaks bitshift

>> dan <<

task1() >> task2()

# task1 selesai sebelum task2 mulai


task1() >> task2() >> task3()

# task1 selesai sebelum task2 dan task2 selesai sebelum task3


task1() >> task3()
task2() >> task3()

# task1 dan task2 bisa berjalan bersamaan tapi keduanya harus selesai sebelum task3

Contoh sintaks bitshift

# Unduh data penjualan sebelum rekonsiliasi
download_sales_data() >> reconcile()

# Unduh data inventaris sebelum rekonsiliasi
download_inventory_data() >> reconcile()

Keduanya harus selesai sebelum rekonsiliasi, urutan tidak ditentukan

Ayo berlatih!

Pengantar Apache Airflow dengan Python