spark_on_k8s.api.app

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.

delete_app(namespace, app_id, force=False) async

This endpoint deletes a spark application by its app_id.

It deletes the driver pod, then the Kubernetes garbage collector will delete the executor pods and the other resources.

Source code in spark_on_k8s/api/app.py

@router.delete(
    "/{namespace}/{app_id}",
    summary="Delete a spark application",
)
async def delete_app(namespace: str, app_id: str, force: bool = False):
    """This endpoint deletes a spark application by its app_id.

    It deletes the driver pod, then the Kubernetes garbage collector
    will delete the executor pods and the other resources.
    """
    namespace_scope().require(namespace)
    async_spark_app_manager = AsyncSparkAppManager(
        k8s_client_manager=KubernetesClientSingleton.client_manager
    )
    try:
        await async_spark_app_manager.delete_app(namespace=namespace, app_id=app_id, force=force)
        return Response(status_code=200)
    except Exception as e:
        count_kubernetes_error(e)
        if (error := _target_error(e)) is not None:
            raise error from e
        return handle_exception(e, 500)

kill_app(namespace, app_id) async

This endpoint kills a spark application by its app_id.

It sends a SIGTERM to the init process of the driver pod (PID 1).

Source code in spark_on_k8s/api/app.py

@router.patch(
    "/{namespace}/{app_id}",
    summary="Kill a spark application",
)
async def kill_app(namespace: str, app_id: str):
    """This endpoint kills a spark application by its app_id.

    It sends a SIGTERM to the init process of the driver pod (PID 1).
    """
    namespace_scope().require(namespace)
    async_spark_app_manager = AsyncSparkAppManager(
        k8s_client_manager=KubernetesClientSingleton.client_manager
    )
    try:
        await async_spark_app_manager.kill_app(namespace=namespace, app_id=app_id)
        return Response(status_code=200)
    except Exception as e:
        count_kubernetes_error(e)
        if (error := _target_error(e)) is not None:
            raise error from e
        return handle_exception(e, 500)