Cloud Composer 1 dalam mode pasca-pemeliharaan. Google tidak merilis update lebih lanjut untuk Cloud Composer 1, termasuk versi baru Airflow, perbaikan bug, dan update keamanan. Sebaiknya rencanakan migrasi ke Cloud Composer 2.

Halaman ini diterjemahkan oleh Cloud Translation API.

Mengakses Airflow REST API

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Apache Airflow memiliki antarmuka REST API yang dapat Anda gunakan untuk melakukan tugas mendapatkan informasi tentang operasi dan tugas DAG, memperbarui DAG, mendapatkan Konfigurasi Airflow, menambahkan dan menghapus koneksi, serta mencantumkan pengguna.

Untuk contoh penggunaan Airflow REST API dengan fungsi Cloud Run, lihat Memicu DAG dengan Cloud Functions.

Versi Airflow REST API

Versi Airflow REST API berikut tersedia di Cloud Composer 2:

Penggunaan Airflow 2 REST API stabil. REST API eksperimental tidak digunakan lagi oleh Airflow.
Anda masih dapat menggunakan REST API eksperimental di Airflow 2 jika mengaktifkannya melalui penggantian konfigurasi Airflow, seperti dijelaskan lebih lanjut.

Sebelum memulai

Aktifkan API Cloud Composer.

Mengaktifkan API

Mengaktifkan Airflow REST API stabil

REST API stabil sudah diaktifkan secara default di Airflow 2.

Cloud Composer menggunakan backend autentikasi API.

Otorisasi berfungsi dengan cara standar yang disediakan oleh Airflow. Saat pengguna baru diotorisasi melalui API, akun pengguna akan mendapatkan peran Op secara default.

Anda dapat mengaktifkan atau menonaktifkan REST API stabil, atau mengubah pengguna default peran dengan mengganti konfigurasi Airflow berikut opsi:

Bagian	Kunci	Nilai	Catatan
`api`	(Airflow 2.2.5 dan yang lebih lama) `auth_backend` (Airflow 2.3.0 dan yang lebih baru) `auth_backends`	`airflow.composer.api.backend.composer_auth`	Untuk menonaktifkan REST API stabil, ubah ke `airflow.api.auth.backend.deny_all`
`api`	`composer_auth_user_registration_role`	`Op`	Anda dapat menentukan peran lainnya.

Mengaktifkan Airflow REST API eksperimental

Secara default, fitur autentikasi API dinonaktifkan dalam Compute Engine API. Server web Airflow menolak semua permintaan yang Anda buat.

Untuk mengaktifkan fitur otentikasi API dan API eksperimental Airflow 2, ganti opsi konfigurasi Airflow berikut:

Bagian	Kunci	Nilai	Catatan
`api`	(Airflow 2.2.5 dan yang lebih lama) `auth_backend` (Airflow 2.3.0 dan yang lebih baru) `auth_backends`	`airflow.api.auth.backend.default`	Nilai defaultnya adalah `airflow.composer.api.backend.composer_auth`.
`api`	`enable_experimental_api`	`True`	Nilai defaultnya adalah `False`.

Setelah Anda menetapkan opsi konfigurasi ini ke airflow.api.auth.backend.default, server web Airflow menerima semua API permintaan tanpa otentikasi. Meskipun server web Airflow itu sendiri tidak memerlukan otentikasi, tapi masih dilindungi oleh Identity-Aware Proxy yang menyediakan lapisan otentikasinya sendiri.

Mengizinkan panggilan API ke Airflow REST API menggunakan Kontrol Akses Webserver

Bergantung pada metode yang digunakan untuk memanggil Airflow REST API, metode pemanggil dapat menggunakan alamat IPv4 atau IPv6. Ingatlah untuk berhenti memblokir Traffic IP ke Airflow REST API menggunakan Webserver Access Control.

Gunakan opsi konfigurasi default, yaitu All IP addresses have access (default) jika Anda tidak yakin dari alamat IP mana panggilan Anda ke Airflow REST API akan dikirimkan.

Melakukan panggilan ke Airflow REST API

Bagian ini menyediakan contoh skrip Python yang dapat Anda gunakan untuk memicu DAG dengan Airflow REST API yang stabil.

Masukkan isi contoh berikut ke dalam file bernama composer2_airflow_rest_api.py, lalu berikan URL UI Airflow Anda, yaitu nama DAG, dan konfigurasi run DAG di nilai variabel.

from __future__ import annotations

from typing import Any

import google.auth
from google.auth.transport.requests import AuthorizedSession
import requests


# Following GCP best practices, these credentials should be
# constructed at start-up time and used throughout
# https://cloud.google.com/apis/docs/client-libraries-best-practices
AUTH_SCOPE = "https://www.googleapis.com/auth/cloud-platform"
CREDENTIALS, _ = google.auth.default(scopes=[AUTH_SCOPE])


def make_composer2_web_server_request(
    url: str, method: str = "GET", **kwargs: Any
) -> google.auth.transport.Response:
    """
    Make a request to Cloud Composer 2 environment's web server.
    Args:
      url: The URL to fetch.
      method: The request method to use ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT',
        'PATCH', 'DELETE')
      **kwargs: Any of the parameters defined for the request function:
                https://github.com/requests/requests/blob/master/requests/api.py
                  If no timeout is provided, it is set to 90 by default.
    """

    authed_session = AuthorizedSession(CREDENTIALS)

    # Set the default timeout, if missing
    if "timeout" not in kwargs:
        kwargs["timeout"] = 90

    return authed_session.request(method, url, **kwargs)


def trigger_dag(web_server_url: str, dag_id: str, data: dict) -> str:
    """
    Make a request to trigger a dag using the stable Airflow 2 REST API.
    https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html

    Args:
      web_server_url: The URL of the Airflow 2 web server.
      dag_id: The DAG ID.
      data: Additional configuration parameters for the DAG run (json).
    """

    endpoint = f"api/v1/dags/{dag_id}/dagRuns"
    request_url = f"{web_server_url}/{endpoint}"
    json_data = {"conf": data}

    response = make_composer2_web_server_request(
        request_url, method="POST", json=json_data
    )

    if response.status_code == 403:
        raise requests.HTTPError(
            "You do not have a permission to perform this operation. "
            "Check Airflow RBAC roles for your account."
            f"{response.headers} / {response.text}"
        )
    elif response.status_code != 200:
        response.raise_for_status()
    else:
        return response.text




if __name__ == "__main__":
    # TODO(developer): replace with your values
    dag_id = "your-dag-id"  # Replace with the ID of the DAG that you want to run.
    dag_config = {
        "your-key": "your-value"
    }  # Replace with configuration parameters for the DAG run.
    # Replace web_server_url with the Airflow web server address. To obtain this
    # URL, run the following command for your environment:
    # gcloud composer environments describe example-environment \
    #  --location=your-composer-region \
    #  --format="value(config.airflowUri)"
    web_server_url = (
        "https://example-airflow-ui-url-dot-us-central1.composer.googleusercontent.com"
    )

    response_text = trigger_dag(
        web_server_url=web_server_url, dag_id=dag_id, data=dag_config
    )

    print(response_text)

Misalnya, konfigurasi di bawah ini salah

  web_server_url = (
    "https://example-airflow-ui-url-dot-us-central1.composer.googleusercontent.com/"
  )

Mengakses Airflow REST API menggunakan akun layanan

Database Airflow membatasi panjang kolom email hingga 64 karakter. Akun layanan terkadang memiliki alamat email yang lebih panjang dari 64 karakter. Pengguna Airflow tidak dapat dibuat untuk layanan tersebut akun dengan cara biasa. Jika tidak ada pengguna Airflow untuk layanan tersebut mengakses Airflow REST API akan menghasilkan error HTTP 401 dan 403.

Sebagai solusinya, Anda dapat melakukan prapendaftaran pengguna Airflow untuk akun layanan. Kepada melakukannya, gunakan accounts.google.com:NUMERIC_USER_ID sebagai nama pengguna, dan string unik sebagai email.

Untuk mendapatkan NUMERIC_USER_ID untuk akun layanan, jalankan:

gcloud iam service-accounts describe \
  SA_NAME@PROJECT_ID.iam.gserviceaccount.com \
  --format="value(oauth2ClientId)"

Ganti:

SA_NAME dengan nama akun layanan.
PROJECT_ID dengan ID project.

Buat pengguna Airflow dengan peran Op untuk akun layanan:
UI Airflow
1. Buka UI Airflow.
2. Buka Security > List Users, lalu klik Tambahkan data baru. Pengguna Airflow Anda harus memiliki peran Admin untuk membuka halaman ini.
3. Tentukan accounts.google.com:NUMERIC_USER_ID sebagai pengguna nama. Ganti NUMERIC_USER_ID dengan ID pengguna yang diperoleh di langkah sebelumnya.
4. Tentukan ID unik sebagai email. Anda dapat menggunakan string unik.
5. Tentukan peran untuk pengguna tersebut. Misalnya, Op.
6. Pastikan kotak centang Is Active? sudah dicentang.
7. Tentukan nama depan dan nama belakang pengguna. Anda dapat menggunakan {i>string<i}.
8. Klik Simpan.
gcloud

Jalankan perintah Airflow CLI berikut:
```
gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    users create -- \
    -u accounts.google.com:NUMERIC_USER_ID \
    -e UNIQUE_ID  \
    -f UNIQUE_ID \
    -l - -r Op --use-random-password
```
Ganti:
- ENVIRONMENT_NAME dengan nama lingkungan.
- LOCATION dengan region tempat lingkungan berada.
- NUMERIC_USER_ID dengan ID pengguna yang diperoleh di langkah sebelumnya.
- UNIQUE_ID dengan ID untuk pengguna Airflow. Anda dapat menggunakan string unik.
Setelah Anda membuat pengguna Airflow untuk akun layanan, pemanggil diautentikasi karena akun layanan diakui sebagai pengguna yang telah melakukan pradaftar, dan login ke Airflow.
Catatan: Untuk konfigurasi penggabungan identitas tenaga kerja: Perlu diingat bahwa akun layanan dianggap sebagai Akun Google, meskipun Anda menggunakan tenaga kerja penggabungan identitas. Saat berinteraksi dengan UI Airflow melalui akun layanan, pastikan Anda menggunakan URL yang ditujukan untuk Akun Google. Lihat Mengakses UI Airflow.

Menskalakan komponen REST API Airflow

Endpoint Airflow REST API dan UI Airflow dijalankan di dalam komponen, yaitu Airflow Webserver. Jika Anda menggunakan REST API secara intensif, pertimbangkan meningkatkan parameter CPU dan memori untuk menyesuaikan resource Airflow Webserver dengan beban yang diharapkan.