Versions Compared

Old Version 1

changes.mady.by.user Unknown User (dimberman)

Saved on

compared with

New Version Current

changes.mady.by.user Jarek Potiuk

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

Background:

Kubernetes is a container-based cluster management system designed by google for easy application deployment. Companies such as Airbnb, Bloomberg, Palantir, and Google use kubernetes fast growing open-source platform which provides container-centric infrastructure. Conceived by Google in 2014, and leveraging over a decade of experience running containers at scale internally, it is one of the fastest moving projects on GitHub with 1000+ contributors and 40,000+ commits. Kubernetes has first class support on Google Cloud Platform, Amazon Web Services, and Microsoft Azure. Kubernetes is an open-source platform designed to automate deploying, scaling, and operating application containers, and is widely used by organizations across the world for a variety of large-scale solutions including serving, stateful applications, and increasingly - data science , ETL, and app deployment. Integrating airflow into Kubernetes and ETL workloads.

While traditional environments like YARN-based hadoop clusters have used Oozie, newer data and ML pipelines built on Kubernetes are increasingly using Airflow for orchestrating and scheduling DAGs. Adding native Kubernetes support into Airflow would increase the would increase viable use cases for airflow, promote airflow as a de facto add a mature and well understood workflow scheduler for to the Kubernetes ecosystem, and create possibilities for improved security and robustness within airflow in the future. 

Kubernetes Executor:

Kubernetes Api:

We will communicate with Kubernetes using the Kubernetes python client. This client will allow us to create, monitor, and kill jobs. Users will be required to either run their airflow instances within the kubernetes cluster, or provide an address to link the API to the cluster.  

Launching Jobs:

Unlike the current MesosExecutor, which uses pickle to serialize DAGs and send them to pre-built slaves, the KubernetesExecutor will launch a new temporary worker job for each task. Each job will have contain a full airflow deployment and will run an an airflow run <dag_id> <task_id> id> command. This design has two major benefits over the previous system. The first benefit is that dynamically creating airflow workers simplifies the cluster set-up. Users will not need to pre-build airflow workers or consider how the nodes will communicate with eachothereach other. The second benefit is that dynamically creating pods allows for a highly elastic system that can easily scale to large workloads while not wasting resources during periods of low usage.

Monitoring jobs:

When we create the Kubernetes jobs, we will maintain a mapping of job_id -> job key. Using these job ids we can use the read_namespaced_jobs endpoint to consistently query kubernetes for the status of running jobs. Upon recieving a failure or success status from the API, the executor can forward the given state to the scheduler to show in the UI. By using the airflow batch job API (as opposed to launching pods), we get an assurance that any failed kubernetes job can retry a pre-set number of times before the executor kills the task.

 

 

1
2
3
for job_id in current_jobs:
    status = api.read_namespaced_job(job_id, namespace).status
    process_status(job_id, key, status)

 

We will watch jobs using the Kubernetes Watch API. This API will allow us to passively watch all events on a namespace, filtered by label. We can contain the watchers on separate threads which can use event handling to handle failures from airflow pods.


Sharing Dags:

To encourage a wide array of potential storage options for airflow users, we will take advantage of kubernetes persistent volume claims. With these claims, users will be allowed to use Kubernetes Persistent Volumes. The PV/PVC abstractions allow Kubernetes to encapsulate a wide variety of distributed storage options such as github, EBS, cinder, NFS, and glusterFS. We will offer a few initial options (such as github and cinder), but will also create a "kubernetes_volume" plugin for users that wish to use other distributed file systemstwo modes for DAG storage: git-mode and persistent volume mode. Git mode is the least scalable, yet easiest to setup DAG storage system. This system will simply pull your DAGS from github in an init container for usage by the airflow pod. This case is primarily recommended for development/testing, yet will still work for small production cases. The persistent volume mode, on the other hand, takes advantage of an existing kubernetes structure called a “persistent volume.” This API will allow users to treat external systems like S3, NFS, and cinder as if they were directories in the local file system. This system is recommended for larger DAG folders in production settings.


Security:

Kubernetes offers multiple inherent security benefits that would allow airflow users to safely run their jobs with minimal risk. By running airflow instances in non-default namespaces, administrators can populate those namespaces with only the secrets required to access data that is allowed for a user or role-account. We could also further restrict access using airflows' multi-tenancy abilities and kerberos integration.

Kubernetes Operator

 

The Kubernetes operator will have a very straightforward implementation. In the same way that DAG folders are all placed within the $AIRFLOW_HOME/dags folder, kubernetes yaml files should be placed in a $AIRFLOW_HOME/kub-yaml folder. This means that a user only needs to identify the name of the yaml file to launch the kubernetes job.


Generating kubernetes pods require a fair amount of unavoidable configuration. To minimize this complexity to the user while still allowing for a high amount of flexibility we have created a KubernetesPodOperatorFactory class. This factory class will prevent anti-patterns like forcing the user to create classes with more than 5 starting parameters or depend on kw-arguments.


Code Block
languagepy
class KubernetesPodOperatorFactory:
    def __init__(
        self,
		trigger_dag_id,
        image,
        cmds
    ):
    def add_env_variables(self, env):
    def add_secrets(self, secrets):
    def add_labels(self, labels):
    def add_name(self, name):
    def set_namespace(self, namespace):
	def set_upstream(self, operator)
    def generate(self):

 

PodOperator
def __init__(
        self,
        trigger_dag_id,
        yaml_name,
        *args, **kwargs):
    super(KubernetesOperator, self).__init__(*args, **kwargs)
    self.yaml_url = yaml_url
    self.trigger_dag_id = trigger_dag_id
    self.container_id = self.task_id
Another option that will give more flexibility for users that do not want to use yamls will be to offer first class kubernetes classes that will create and launch pods/jobs at will.

 

class Pod():
    def __init__(
            self,
            image,
            envs = {},
            cmds = [],
            secrets = [],
            labels = {},
            node_selectors = {},
            kube_req_factory = None,
            name = None,
            namespace = 'default',
            result = None):
 
    def launch_pod(self):
    def _execution_finished(self):
 
class PodOperator(BaseOperator):

 

 

Questions posed by the airflow team:
...
If an airflow worker fails it might be useful to keep the kubernetes worker reserved and preserved in it's same state for debugging purposes
  • Kubernetes is primarily meant to run stateless applications. To our current knowledge there is no way to preserve state for a kubernetes jobpods can retain state and logs, and we can use etcd to preserve some additional state as well (through a CRD in future).  
Other interesting points:
The Airflow Kubernetes executor should try to respect the resources that are set in tasks for
  scheduling when hitting the kubernetes API
Future work
Spark-On-K8s integration:
Teams at Google, Palantir, and many others are currently nearing release for a beta for spark that would run natively on kubernetes. This application would allow users to submit spark-submit commands to a resource manager that can dynamically spawn spark clusters for data processing. A seperate spark-on-k8s hook can be developed to sit within the SparkSubmitOperator depending on user configurations.