spark_on_k8s.airflow.operator_links

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

SparkOnK8SOperatorLink

Bases: BaseOperatorLink

Operator link for SparkOnK8SOperator.

It allows users to access Spark job UI and spark history UI using SparkOnK8SOperator.

Source code in spark_on_k8s/airflow/operator_links.py

class SparkOnK8SOperatorLink(BaseOperatorLink):
    """
    Operator link for SparkOnK8SOperator.

    It allows users to access Spark job UI and spark history UI using SparkOnK8SOperator.
    """

    name = "Spark Job UI"
    xcom_key = "spark_ui_link"

    @staticmethod
    def persist_spark_ui_link(
        context: Context,
        task_instance: BaseOperator,
        spark_on_k8s_service_url: str,
        namespace: str,
        spark_app_id: str,
    ):
        """
        Persist Spark UI link to XCom.
        """
        from spark_on_k8s.airflow.operators import SparkOnK8SOperator

        spark_ui_link = f"{spark_on_k8s_service_url}/webserver/ui/{namespace}/{spark_app_id}"
        context["ti"].xcom_push(
            key=SparkOnK8SOperator.XCOM_SPARK_UI_LINK,
            value=spark_ui_link,
        )

    @staticmethod
    def persist_spark_history_ui_link(
        context: Context, task_instance: BaseOperator, spark_on_k8s_service_url: str, spark_app_id: str
    ):
        """
        Persist Spark history UI link to XCom.
        """
        from spark_on_k8s.airflow.operators import SparkOnK8SOperator

        spark_history_ui_link = f"{spark_on_k8s_service_url}/webserver/ui-history/history/{spark_app_id}"
        context["ti"].xcom_push(
            key=SparkOnK8SOperator.XCOM_SPARK_UI_LINK,
            value=spark_history_ui_link,
        )

    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
        """
        Get link to Spark job UI or Spark history UI.
        """
        from spark_on_k8s.airflow.operators import SparkOnK8SOperator

        spark_ui_link = XCom.get_value(ti_key=ti_key, key=SparkOnK8SOperator.XCOM_SPARK_UI_LINK)
        if spark_ui_link:
            return spark_ui_link
        else:
            return ""

get_link(operator, *, ti_key)

Get link to Spark job UI or Spark history UI.

Source code in spark_on_k8s/airflow/operator_links.py

def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
    """
    Get link to Spark job UI or Spark history UI.
    """
    from spark_on_k8s.airflow.operators import SparkOnK8SOperator

    spark_ui_link = XCom.get_value(ti_key=ti_key, key=SparkOnK8SOperator.XCOM_SPARK_UI_LINK)
    if spark_ui_link:
        return spark_ui_link
    else:
        return ""

persist_spark_history_ui_link(context, task_instance, spark_on_k8s_service_url, spark_app_id) staticmethod

Persist Spark history UI link to XCom.

Source code in spark_on_k8s/airflow/operator_links.py

@staticmethod
def persist_spark_history_ui_link(
    context: Context, task_instance: BaseOperator, spark_on_k8s_service_url: str, spark_app_id: str
):
    """
    Persist Spark history UI link to XCom.
    """
    from spark_on_k8s.airflow.operators import SparkOnK8SOperator

    spark_history_ui_link = f"{spark_on_k8s_service_url}/webserver/ui-history/history/{spark_app_id}"
    context["ti"].xcom_push(
        key=SparkOnK8SOperator.XCOM_SPARK_UI_LINK,
        value=spark_history_ui_link,
    )

persist_spark_ui_link(context, task_instance, spark_on_k8s_service_url, namespace, spark_app_id) staticmethod

Persist Spark UI link to XCom.

Source code in spark_on_k8s/airflow/operator_links.py

@staticmethod
def persist_spark_ui_link(
    context: Context,
    task_instance: BaseOperator,
    spark_on_k8s_service_url: str,
    namespace: str,
    spark_app_id: str,
):
    """
    Persist Spark UI link to XCom.
    """
    from spark_on_k8s.airflow.operators import SparkOnK8SOperator

    spark_ui_link = f"{spark_on_k8s_service_url}/webserver/ui/{namespace}/{spark_app_id}"
    context["ti"].xcom_push(
        key=SparkOnK8SOperator.XCOM_SPARK_UI_LINK,
        value=spark_ui_link,
    )