Operator extra link 정의하기

Operator에 extra link를 추가하는 방법을 설명하는 문서예요. 플러그인이나 provider 패키지로 extra link를 정의하면 Grid 뷰의 태스크 상세 페이지에 표시돼요. 컨텍스트 의존 링크, 기존 operator에 링크 추가·오버라이드하는 방법까지 코드와 함께 살펴볼게요.

출처: 문서

본문

Operator에 extra link를 추가하고 싶다면 플러그인이나 provider 패키지를 통해 정의할 수 있어요. Extra link는 Grid 뷰의 태스크 상세 페이지에 표시돼요.

다음 코드는 Plugins를 통해 operator에 extra link를 추가하는 방법을 보여줘요:

from airflow.sdk import BaseOperator
from airflow.sdk import BaseOperatorLink
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.plugins_manager import AirflowPlugin

class GoogleLink(BaseOperatorLink):
    name = "Google"

    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey):
        return "https://www.google.com"

class MyFirstOperator(BaseOperator):
    operator_extra_links = (GoogleLink(),)

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def execute(self, context):
        self.log.info("Hello World!")

# Defining the plugin class
class AirflowExtraLinkPlugin(AirflowPlugin):
    name = "extra_link_plugin"
    operator_extra_links = [
        GoogleLink(),
    ]

커스텀 Airflow Provider나 Airflow operator를 통해 정의된 extra link는 태스크 실행 중에 xcom으로 XCom 테이블에 푸시돼요. Grid 뷰에서 표시할 때는 이 xcom이 검색되어 표시돼요.

또한 Airflow 플러그인이나 Airflow providers를 통해 모든 operator에 사용할 수 있는 전역 operator extra link를 추가할 수도 있어요. 자세한 내용은 플러그인 인터페이스와 Providers에서 확인할 수 있어요.

커뮤니티 provider를 통해 사용할 수 있는 모든 extra link는 Extra Links에서 볼 수 있어요.

Airflow 플러그인이나 커스텀 provider를 통해 기존 operator에 extra link를 추가(또는 오버라이드)할 수도 있어요.

예를 들어 다음 Airflow 플러그인은 GCSToS3Operator operator를 사용하는 모든 태스크에 Operator Link를 추가해요.

기존 Operator에 Operator Links 추가하기 (Adding Operator Links to Existing Operators) — plugins/extra_link.py:

from airflow.sdk import BaseOperator, BaseOperatorLink
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.plugins_manager import AirflowPlugin
from airflow.providers.amazon.aws.transfers.gcs_to_s3 import GCSToS3Operator

class S3LogLink(BaseOperatorLink):
    name = "S3"

    # Add list of all the operators to which you want to add this extra link
    # Example: operators = [GCSToS3Operator, GCSToBigQueryOperator]
    operators = [GCSToS3Operator]

    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey):
        # Invalid bucket name because upper case letters and underscores are used
        # This will not be a valid bucket in any region
        bucket_name = "Invalid_Bucket_Name"
        return "https://s3.amazonaws.com/airflow-logs/{bucket_name}/{dag_id}/{task_id}/{run_id}".format(
            bucket_name=bucket_name,
            dag_id=operator.dag_id,
            task_id=operator.task_id,
            run_id=ti_key.run_id,
        )

# Defining the plugin class
class AirflowExtraLinkPlugin(AirflowPlugin):
    name = "extra_link_plugin"
    operator_extra_links = [
        S3LogLink(),
    ]

기존 Operator의 Operator Links 오버라이드하기 (Overriding Operator Links of Existing Operators):

플러그인을 통해 operator의 내장 링크를 바꾸는 것도 가능해요. 예를 들어 BigQueryExecuteQueryOperator는 Google Cloud Console로 가는 링크를 포함하는데, 그 링크를 바꾸고 싶다면 이렇게 할 수 있어요:

from airflow.sdk import BaseOperator, BaseOperatorLink
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.plugins_manager import AirflowPlugin
from airflow.providers.google.cloud.operators.bigquery import BigQueryOperator

# Change from https to http just to display the override
BIGQUERY_JOB_DETAILS_LINK_FMT = "http://console.cloud.google.com/bigquery?j={job_id}"

class BigQueryDatasetLink(BaseGoogleLink):
    """
    Helper class for constructing BigQuery Dataset Link.
    """

    name = "BigQuery Dataset"
    key = "bigquery_dataset"
    format_str = BIGQUERY_DATASET_LINK_FMT

    @staticmethod
    def persist(
        context: Context,
        task_instance: BaseOperator,
        dataset_id: str,
        project_id: str,
    ):
        task_instance.xcom_push(
            context,
            key=BigQueryDatasetLink.key,
            value={"dataset_id": dataset_id, "project_id": project_id},
        )

# Defining the plugin class
class AirflowExtraLinkPlugin(AirflowPlugin):
    name = "extra_link_plugin"
    operator_extra_links = [
        BigQueryDatasetLink(),
    ]

Providers를 통해 Operator Links 추가하기 (Adding Operator Links via Providers)

Providers에서 설명했듯이, 자신만의 Airflow Provider를 만들 때 extra link 기능을 제공하는 operator 목록을 지정할 수 있어요. 이는 Provider 패키지의 메타데이터에 저장된 provider-info 정보에 operator 클래스 이름을 포함시켜 이루어져요:

provider-info 딕셔너리에 필요한 예시 메타데이터(현재 apache-airflow-providers-google provider가 반환하는 메타데이터의 일부)예요:

extra-links:
  - airflow.providers.google.cloud.links.bigquery.BigQueryDatasetLink
  - airflow.providers.google.cloud.links.bigquery.BigQueryTableLink

extra link가 있는 operator는 원하는 만큼 많이 포함할 수 있어요.

더 알아보기 (Learn more)