Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[AIRFLOW-2140] Don't require kubernetes for the SparkSubmit hook #3700

Merged
merged 1 commit into from
Aug 7, 2018
Merged

[AIRFLOW-2140] Don't require kubernetes for the SparkSubmit hook #3700

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
[AIRFLOW-2140] Don't require kubernetes for the SparkSubmit hook 
This extra dep is a quasi-breaking change when upgrading - previously
there were no deps outside of Airflow itself for this hook. Importing
the k8s libs breaks installs that aren't also using Kubernetes.

This makes the dep optional for anyone who doesn't explicitly use the
functionality
  • Loading branch information
@ashb
ashb committed Aug 5, 2018
commit 809a8a7f11e376cc4f863910e9360868348a9da1
7 changes: 5 additions & 2 deletions airflow/contrib/hooks/spark_submit_hook.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
from airflow.exceptions import AirflowException
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.contrib.kubernetes import kube_client
from kubernetes.client.rest import ApiException


class SparkSubmitHook(BaseHook, LoggingMixin):
Expand Down Expand Up @@ -136,6 +135,10 @@ def __init__(self,
self._connection = self._resolve_connection()
self._is_yarn = 'yarn' in self._connection['master']
self._is_kubernetes = 'k8s' in self._connection['master']
if self._is_kubernetes and kube_client is None:
raise RuntimeError(
"{master} specified by kubernetes dependencies are not installed!".format(
self._connection['master']))

self._should_track_driver_status = self._resolve_should_track_driver_status()
self._driver_id = None
Expand Down Expand Up @@ -559,6 +562,6 @@ def on_kill(self):

self.log.info("Spark on K8s killed with response: %s", api_response)

except ApiException as e:
except kube_client.ApiException as e:
self.log.info("Exception when attempting to kill Spark on K8s:")
self.log.exception(e)
14 changes: 13 additions & 1 deletion airflow/contrib/kubernetes/kube_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,21 @@
from airflow.configuration import conf
from six import PY2

try:
from kubernetes import config, client
from kubernetes.client.rest import ApiException
has_kubernetes = True
except ImportError as e:
# We need an exception class to be able to use it in ``except`` elsewhere
# in the code base
ApiException = BaseException
has_kubernetes = False
_import_err = e


def _load_kube_config(in_cluster, cluster_context, config_file):
from kubernetes import config, client
if not has_kubernetes:
raise _import_err
if in_cluster:
config.load_incluster_config()
else:
Expand Down