From 598595b7ec05040617939d0ab4fff466f8d44484 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Fri, 29 Jun 2018 10:30:13 +0200 Subject: [PATCH 01/15] Drop snakebite in favour of hdfs3. --- airflow/hooks/hdfs_hook.py | 126 +++++++++++++++------------------- setup.py | 2 +- tests/hooks/test_hdfs_hook.py | 112 ++++++++++++++++++++++++++++++ 3 files changed, 168 insertions(+), 72 deletions(-) create mode 100644 tests/hooks/test_hdfs_hook.py diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index 597b7c4f7ec83..83b3fe1c25df7 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -17,85 +17,69 @@ # specific language governing permissions and limitations # under the License. -from six import PY2 +import hdfs3 +from hdfs3.utils import MyNone -from airflow import configuration -from airflow.exceptions import AirflowException from airflow.hooks.base_hook import BaseHook -snakebite_imported = False -if PY2: - from snakebite.client import Client, HAClient, Namenode, AutoConfigClient - snakebite_imported = True +class HDFSHook(BaseHook): + """Hook for interacting with HDFS using the hdfs3 library. + By default hdfs3 loads its configuration from `core-site.xml` and + `hdfs-site.xml` if these files can be found in any of the typical + locations. The hook loads `host` and `port` parameters from the + hdfs connection (if given) and extra configuration parameters can be + supplied using the `pars` key in extra JSON. See the hdfs3 documentation + for more details. -class HDFSHookException(AirflowException): - pass + :param str hdfs_conn_id: Connection ID to fetch parameters from. + :param bool autoconf: Whether to use autoconfig to discover + configuration options from the hdfs XML configuration files. + """ + def __init__(self, hdfs_conn_id=None, autoconf=True): + super().__init__(None) -class HDFSHook(BaseHook): - """ - Interact with HDFS. This class is a wrapper around the snakebite library. - - :param hdfs_conn_id: Connection id to fetch connection info - :type hdfs_conn_id: str - :param proxy_user: effective user for HDFS operations - :type proxy_user: str - :param autoconfig: use snakebite's automatically configured client - :type autoconfig: bool - """ - def __init__(self, hdfs_conn_id='hdfs_default', proxy_user=None, - autoconfig=False): - if not snakebite_imported: - raise ImportError( - 'This HDFSHook implementation requires snakebite, but ' - 'snakebite is not compatible with Python 3 ' - '(as of August 2015). Please use Python 2 if you require ' - 'this hook -- or help by submitting a PR!') self.hdfs_conn_id = hdfs_conn_id - self.proxy_user = proxy_user - self.autoconfig = autoconfig + self._autoconf = autoconf + + self._conn = None def get_conn(self): - """ - Returns a snakebite HDFSClient object. - """ - # When using HAClient, proxy_user must be the same, so is ok to always - # take the first. - effective_user = self.proxy_user - autoconfig = self.autoconfig - use_sasl = configuration.conf.get('core', 'security') == 'kerberos' - - try: - connections = self.get_connections(self.hdfs_conn_id) - - if not effective_user: - effective_user = connections[0].login - if not autoconfig: - autoconfig = connections[0].extra_dejson.get('autoconfig', - False) - hdfs_namenode_principal = connections[0].extra_dejson.get( - 'hdfs_namenode_principal') - except AirflowException: - if not autoconfig: - raise - - if autoconfig: - # will read config info from $HADOOP_HOME conf files - client = AutoConfigClient(effective_user=effective_user, - use_sasl=use_sasl) - elif len(connections) == 1: - client = Client(connections[0].host, connections[0].port, - effective_user=effective_user, use_sasl=use_sasl, - hdfs_namenode_principal=hdfs_namenode_principal) - elif len(connections) > 1: - nn = [Namenode(conn.host, conn.port) for conn in connections] - client = HAClient(nn, effective_user=effective_user, - use_sasl=use_sasl, - hdfs_namenode_principal=hdfs_namenode_principal) - else: - raise HDFSHookException("conn_id doesn't exist in the repository " - "and autoconfig is not specified") - - return client + if self._conn is None: + if self.hdfs_conn_id is None: + self._conn = hdfs3.HDFileSystem(autoconf=self._autoconf) + else: + params = self.get_connection(self.hdfs_conn_id) + + # Extract hadoop parameters from extra. + hdfs_pars = params.extra_dejson.get('pars', {}) + + # Collect extra parameters to pass to kwargs. + extra_kws = {} + if params.login: + extra_kws['user'] = params.login + + # Build connection. + self._conn = hdfs3.HDFileSystem( + host=params.host or MyNone, + port=params.port or MyNone, + pars=hdfs_pars, + autoconf=self._autoconf, + **extra_kws) + + return self._conn + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + def close(self): + """Closes the HDFSHook and any underlying connections.""" + + if self._conn is not None: + self._conn.disconnect() + self._conn = None diff --git a/setup.py b/setup.py index 1cbe29b64ad05..23278036419f2 100644 --- a/setup.py +++ b/setup.py @@ -183,7 +183,7 @@ def write_version(filename=os.path.join(*['airflow', ] github_enterprise = ['Flask-OAuthlib>=0.9.1'] google_auth = ['Flask-OAuthlib>=0.9.1'] -hdfs = ['snakebite>=2.7.8'] +hdfs = ['hdfs3>=0.3.0'] hive = [ 'hmsclient>=0.1.0', 'pyhive>=0.6.0', diff --git a/tests/hooks/test_hdfs_hook.py b/tests/hooks/test_hdfs_hook.py new file mode 100644 index 0000000000000..2de8b30a9dae6 --- /dev/null +++ b/tests/hooks/test_hdfs_hook.py @@ -0,0 +1,112 @@ +# -*- coding: utf-8 -*- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +import unittest + +import mock +from hdfs3.utils import MyNone + +from airflow.hooks.hdfs_hook import HDFSHook, hdfs3 + + +class TestHDFSHook(unittest.TestCase): + """ + Tests for the Hdfs3Hook class. + + Note that the HDFileSystem class is mocked in most of these tests + to avoid the requirement of having a local HDFS instance for testing. + """ + + def setUp(self): + self._mock_fs = mock.Mock() + + self._mocked_hook = HDFSHook() + self._mocked_hook._conn = self._mock_fs + + @mock.patch.object(hdfs3, 'HDFileSystem') + @mock.patch.object(HDFSHook, 'get_connection') + def test_get_conn(self, conn_mock, hdfs3_mock): + """Tests get_conn call without ID.""" + + with HDFSHook() as hook: + hook.get_conn() + + conn_mock.assert_not_called() + hdfs3_mock.assert_called_once_with(autoconf=True) + + @mock.patch.object(hdfs3, 'HDFileSystem') + @mock.patch.object(HDFSHook, 'get_connection') + def test_get_conn_no_autoconf(self, conn_mock, hdfs3_mock): + """Tests get_conn call without ID and autoconf = False.""" + + with HDFSHook(autoconf=False) as hook: + hook.get_conn() + + conn_mock.assert_not_called() + hdfs3_mock.assert_called_once_with(autoconf=False) + + @mock.patch.object(hdfs3, 'HDFileSystem') + @mock.patch.object(HDFSHook, 'get_connection') + def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): + """Tests get_conn call with ID.""" + + conn_mock.return_value = mock.Mock( + host='namenode', + login='hdfs_user', + port=8020, + extra_dejson={'pars': {'dfs.namenode.logging.level': 'info'}}) + + with HDFSHook(hdfs_conn_id='hdfs_default') as hook: + hook.get_conn() + + conn_mock.assert_called_once_with('hdfs_default') + + hdfs3_mock.assert_called_once_with( + host='namenode', + port=8020, + pars={'dfs.namenode.logging.level': 'info'}, + user='hdfs_user', + autoconf=True) + + @mock.patch.object(hdfs3, 'HDFileSystem') + @mock.patch.object(HDFSHook, 'get_connection') + def test_get_conn_with_empty_conn(self, conn_mock, hdfs3_mock): + """Tests get_conn call with empty connection.""" + + conn_mock.return_value = mock.Mock( + host='', + login='', + port='', + extra_dejson={}) + + with HDFSHook(hdfs_conn_id='hdfs_default') as hook: + hook.get_conn() + + conn_mock.assert_called_once_with('hdfs_default') + + hdfs3_mock.assert_called_once_with( + host=MyNone, + port=MyNone, + pars={}, + autoconf=True) + + +if __name__ == '__main__': + unittest.main() From 56e52e42ab13f8f96c684fe62552e92e86e398ca Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Fri, 29 Jun 2018 15:33:55 +0200 Subject: [PATCH 02/15] Update HdfsSensor for the new HdfsHook class. --- airflow/hooks/hdfs_hook.py | 28 +++- airflow/sensors/hdfs_sensor.py | 156 +++++++++++----------- tests/core.py | 152 ---------------------- tests/hooks/test_hdfs_hook.py | 20 +-- tests/sensors/test_hdfs_sensor.py | 209 +++++++++++++++++++++--------- 5 files changed, 262 insertions(+), 303 deletions(-) diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index 83b3fe1c25df7..411a7697d5169 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -17,13 +17,15 @@ # specific language governing permissions and limitations # under the License. +import warnings + import hdfs3 from hdfs3.utils import MyNone from airflow.hooks.base_hook import BaseHook -class HDFSHook(BaseHook): +class HdfsHook(BaseHook): """Hook for interacting with HDFS using the hdfs3 library. By default hdfs3 loads its configuration from `core-site.xml` and @@ -83,3 +85,27 @@ def close(self): if self._conn is not None: self._conn.disconnect() self._conn = None + + +class _DeprecationHelper(object): + def __init__(self, new_target, message, category=PendingDeprecationWarning): + self._message = message + self._new_target = new_target + self._category = category + + def _warn(self): + warnings.warn(self._message, category=self._category) + + def __call__(self, *args, **kwargs): + self._warn() + return self._new_target(*args, **kwargs) + + def __getattr__(self, attr): + self._warn() + return getattr(self._new_target, attr) + + +HDFSHook = _DeprecationHelper( + HdfsHook, + message='The `HDFSHook` has been renamed to `HdfsHook`. Support for ' + 'the old naming will be dropped in a future version of Airflow.') diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index 3eb5145ec9668..1261e6629ada0 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -17,107 +17,109 @@ # specific language governing permissions and limitations # under the License. -import re -import sys -from builtins import str +import posixpath +import warnings from airflow import settings -from airflow.hooks.hdfs_hook import HDFSHook +from airflow.hooks.hdfs_hook import HdfsHook from airflow.sensors.base_sensor_operator import BaseSensorOperator from airflow.utils.decorators import apply_defaults -from airflow.utils.log.logging_mixin import LoggingMixin class HdfsSensor(BaseSensorOperator): """ Waits for a file or folder to land in HDFS """ - template_fields = ('filepath',) + + template_fields = ('file_path',) ui_color = settings.WEB_COLORS['LIGHTBLUE'] @apply_defaults def __init__(self, - filepath, + file_path, hdfs_conn_id='hdfs_default', - ignored_ext=None, - ignore_copying=True, file_size=None, - hook=HDFSHook, + ignored_ext=('_COPYING_', ), + ignore_copying=None, + hook=HdfsHook, *args, **kwargs): + + if ignore_copying is not None: + warnings.warn( + 'The ignore_copying argument is no longer used and will ' + 'be removed in the next version of Airflow.', + category=PendingDeprecationWarning) + super(HdfsSensor, self).__init__(*args, **kwargs) - if ignored_ext is None: - ignored_ext = ['_COPYING_'] - self.filepath = filepath self.hdfs_conn_id = hdfs_conn_id + self.file_path = file_path + self.file_size = file_size - self.ignored_ext = ignored_ext - self.ignore_copying = ignore_copying + self.ignored_ext = set(ignored_ext) + self.hook = hook - @staticmethod - def filter_for_filesize(result, size=None): - """ - Will test the filepath result and test if its size is at least self.filesize - - :param result: a list of dicts returned by Snakebite ls - :param size: the file size in MB a file should be at least to trigger True - :return: (bool) depending on the matching criteria - """ - if size: - log = LoggingMixin().log - log.debug( - 'Filtering for file size >= %s in files: %s', - size, map(lambda x: x['path'], result) - ) - size *= settings.MEGABYTE - result = [x for x in result if x['length'] >= size] - log.debug('HdfsSensor.poke: after size filter result is %s', result) - return result - - @staticmethod - def filter_for_ignored_ext(result, ignored_ext, ignore_copying): - """ - Will filter if instructed to do so the result to remove matching criteria - - :param result: list of dicts returned by Snakebite ls - :type result: list[dict] - :param ignored_ext: list of ignored extensions - :type ignored_ext: list - :param ignore_copying: shall we ignore ? - :type ignore_copying: bool - :return: list of dicts which were not removed - :rtype: list[dict] - """ - if ignore_copying: - log = LoggingMixin().log - regex_builder = r"^.*\.(%s$)$" % '$|'.join(ignored_ext) - ignored_extensions_regex = re.compile(regex_builder) - log.debug( - 'Filtering result for ignored extensions: %s in files %s', - ignored_extensions_regex.pattern, map(lambda x: x['path'], result) - ) - result = [x for x in result if not ignored_extensions_regex.match(x['path'])] - log.debug('HdfsSensor.poke: after ext filter result is %s', result) - return result + @property + def filepath(self): + warnings.warn( + 'The `filepath` property has been renamed to `file_path`. ' + 'Support for the old accessor will be dropped in the next ' + 'version of Airflow.', + category=PendingDeprecationWarning) + return self.file_path def poke(self, context): - sb = self.hook(self.hdfs_conn_id).get_conn() - self.log.info('Poking for file {self.filepath}'.format(**locals())) + self.log.info('Poking for file %s', self.file_path) + hdfs_conn = self.hook(self.hdfs_conn_id).get_conn() + try: - # IMOO it's not right here, as there no raise of any kind. - # if the filepath is let's say '/data/mydirectory', - # it's correct but if it is '/data/mydirectory/*', - # it's not correct as the directory exists and sb does not raise any error - # here is a quick fix - result = [f for f in sb.ls([self.filepath], include_toplevel=False)] - self.log.debug('HdfsSensor.poke: result is %s', result) - result = self.filter_for_ignored_ext( - result, self.ignored_ext, self.ignore_copying - ) - result = self.filter_for_filesize(result, self.file_size) - return bool(result) - except Exception: - e = sys.exc_info() - self.log.debug("Caught an exception !: %s", str(e)) + file_paths = hdfs_conn.glob(self.file_path) + except IOError: + # File path doesn't exist yet. return False + + self.log.info('Files matching pattern: %s', file_paths) + + if self.file_size: + file_paths = self._filter_for_size( + hdfs_conn, file_paths, min_size=self.file_size) + self.log.info('Files after filtering for size: %s', file_paths) + + if self.ignored_ext: + file_paths = self._filter_with_ext( + file_paths, exts=self.ignored_ext) + self.log.info('Files after filtering for extensions: %s', + file_paths) + + return self._not_empty(file_paths) + + @staticmethod + def _not_empty(iterable): + try: + next(iterable) + return True + except StopIteration: + return False + + @staticmethod + def _filter_for_size(hdfs_conn, file_paths, min_size): + """Filters file paths for a minimum file size.""" + + min_size_mb = min_size * settings.MEGABYTE + + for file_path in file_paths: + info = hdfs_conn.info(file_path) + if info['kind'] == 'file' and info['size'] > min_size_mb: + yield file_path + + @staticmethod + def _filter_with_ext(file_paths, exts): + """Filters any files with the given extensions.""" + + for file_path in file_paths: + # Get file extension without preceding '.'. + file_ext = posixpath.splitext(file_path)[1][1:] + + if file_ext not in exts: + yield file_path diff --git a/tests/core.py b/tests/core.py index 0fbe29952e66d..f7e3edb2cefea 100644 --- a/tests/core.py +++ b/tests/core.py @@ -2647,158 +2647,6 @@ def check_for_path(self, hdfs_path): return hdfs_path -class FakeSnakeBiteClientException(Exception): - pass - - -class FakeSnakeBiteClient(object): - - def __init__(self): - self.started = True - - def ls(self, path, include_toplevel=False): - """ - the fake snakebite client - :param path: the array of path to test - :param include_toplevel: to return the toplevel directory info - :return: a list for path for the matching queries - """ - if path[0] == '/datadirectory/empty_directory' and not include_toplevel: - return [] - elif path[0] == '/datadirectory/datafile': - return [{ - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, - 'length': 0, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/datafile' - }] - elif path[0] == '/datadirectory/empty_directory' and include_toplevel: - return [{ - 'group': u'supergroup', - 'permission': 493, - 'file_type': 'd', - 'access_time': 0, - 'block_replication': 0, - 'modification_time': 1481132141540, - 'length': 0, - 'blocksize': 0, - 'owner': u'hdfs', - 'path': '/datadirectory/empty_directory' - }] - elif path[0] == '/datadirectory/not_empty_directory' and include_toplevel: - return [{ - 'group': u'supergroup', - 'permission': 493, - 'file_type': 'd', - 'access_time': 0, - 'block_replication': 0, - 'modification_time': 1481132141540, - 'length': 0, - 'blocksize': 0, - 'owner': u'hdfs', - 'path': '/datadirectory/empty_directory' - }, { - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, - 'length': 0, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/not_empty_directory/test_file' - }] - elif path[0] == '/datadirectory/not_empty_directory': - return [{ - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, - 'length': 0, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/not_empty_directory/test_file' - }] - elif path[0] == '/datadirectory/not_existing_file_or_directory': - raise FakeSnakeBiteClientException - elif path[0] == '/datadirectory/regex_dir': - return [{ - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, 'length': 12582912, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/regex_dir/test1file' - }, { - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, - 'length': 12582912, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/regex_dir/test2file' - }, { - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, - 'length': 12582912, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/regex_dir/test3file' - }, { - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, - 'length': 12582912, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/regex_dir/copying_file_1.txt._COPYING_' - }, { - 'group': u'supergroup', - 'permission': 420, - 'file_type': 'f', - 'access_time': 1481122343796, - 'block_replication': 3, - 'modification_time': 1481122343862, - 'length': 12582912, - 'blocksize': 134217728, - 'owner': u'hdfs', - 'path': '/datadirectory/regex_dir/copying_file_3.txt.sftp' - }] - else: - raise FakeSnakeBiteClientException - - -class FakeHDFSHook(object): - def __init__(self, conn_id=None): - self.conn_id = conn_id - - def get_conn(self): - client = FakeSnakeBiteClient() - return client - - class ConnectionTest(unittest.TestCase): def setUp(self): configuration.load_test_config() diff --git a/tests/hooks/test_hdfs_hook.py b/tests/hooks/test_hdfs_hook.py index 2de8b30a9dae6..e783f801f376c 100644 --- a/tests/hooks/test_hdfs_hook.py +++ b/tests/hooks/test_hdfs_hook.py @@ -23,7 +23,7 @@ import mock from hdfs3.utils import MyNone -from airflow.hooks.hdfs_hook import HDFSHook, hdfs3 +from airflow.hooks.hdfs_hook import HdfsHook, hdfs3 class TestHDFSHook(unittest.TestCase): @@ -37,33 +37,33 @@ class TestHDFSHook(unittest.TestCase): def setUp(self): self._mock_fs = mock.Mock() - self._mocked_hook = HDFSHook() + self._mocked_hook = HdfsHook() self._mocked_hook._conn = self._mock_fs @mock.patch.object(hdfs3, 'HDFileSystem') - @mock.patch.object(HDFSHook, 'get_connection') + @mock.patch.object(HdfsHook, 'get_connection') def test_get_conn(self, conn_mock, hdfs3_mock): """Tests get_conn call without ID.""" - with HDFSHook() as hook: + with HdfsHook() as hook: hook.get_conn() conn_mock.assert_not_called() hdfs3_mock.assert_called_once_with(autoconf=True) @mock.patch.object(hdfs3, 'HDFileSystem') - @mock.patch.object(HDFSHook, 'get_connection') + @mock.patch.object(HdfsHook, 'get_connection') def test_get_conn_no_autoconf(self, conn_mock, hdfs3_mock): """Tests get_conn call without ID and autoconf = False.""" - with HDFSHook(autoconf=False) as hook: + with HdfsHook(autoconf=False) as hook: hook.get_conn() conn_mock.assert_not_called() hdfs3_mock.assert_called_once_with(autoconf=False) @mock.patch.object(hdfs3, 'HDFileSystem') - @mock.patch.object(HDFSHook, 'get_connection') + @mock.patch.object(HdfsHook, 'get_connection') def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): """Tests get_conn call with ID.""" @@ -73,7 +73,7 @@ def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): port=8020, extra_dejson={'pars': {'dfs.namenode.logging.level': 'info'}}) - with HDFSHook(hdfs_conn_id='hdfs_default') as hook: + with HdfsHook(hdfs_conn_id='hdfs_default') as hook: hook.get_conn() conn_mock.assert_called_once_with('hdfs_default') @@ -86,7 +86,7 @@ def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): autoconf=True) @mock.patch.object(hdfs3, 'HDFileSystem') - @mock.patch.object(HDFSHook, 'get_connection') + @mock.patch.object(HdfsHook, 'get_connection') def test_get_conn_with_empty_conn(self, conn_mock, hdfs3_mock): """Tests get_conn call with empty connection.""" @@ -96,7 +96,7 @@ def test_get_conn_with_empty_conn(self, conn_mock, hdfs3_mock): port='', extra_dejson={}) - with HDFSHook(hdfs_conn_id='hdfs_default') as hook: + with HdfsHook(hdfs_conn_id='hdfs_default') as hook: hook.get_conn() conn_mock.assert_called_once_with('hdfs_default') diff --git a/tests/sensors/test_hdfs_sensor.py b/tests/sensors/test_hdfs_sensor.py index 26adeaa3ab60b..12c1e4b453ff4 100644 --- a/tests/sensors/test_hdfs_sensor.py +++ b/tests/sensors/test_hdfs_sensor.py @@ -16,76 +16,159 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -import unittest from datetime import timedelta +import fnmatch +import unittest + +import mock + +from airflow import models +from airflow.sensors.hdfs_sensor import HdfsSensor, HdfsHook -from airflow import configuration -from airflow.exceptions import AirflowSensorTimeout -from airflow.sensors.hdfs_sensor import HdfsSensor -from airflow.utils.timezone import datetime -from tests.core import FakeHDFSHook -configuration.load_test_config() +class MockHdfs3Client(object): + """Mock hdfs3 client for testing purposes.""" -DEFAULT_DATE = datetime(2015, 1, 1) -TEST_DAG_ID = 'unit_test_dag' + def __init__(self, file_details): + self._file_details = { + entry['name']: entry for entry in file_details + } + + def glob(self, pattern): + """Returns glob of files matching pattern.""" + return fnmatch.filter(self._file_details.keys(), pattern) + + def info(self, file_path): + """Returns info for given file path.""" + + try: + return self._file_details[file_path] + except KeyError: + raise IOError() class HdfsSensorTests(unittest.TestCase): + """Tests for the HdfsSensor class.""" def setUp(self): - self.hook = FakeHDFSHook - - def test_legacy_file_exist(self): - """ - Test the legacy behaviour - :return: - """ - # When - task = HdfsSensor(task_id='Should_be_file_legacy', - filepath='/datadirectory/datafile', - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) - task.execute(None) - - # Then - # Nothing happens, nothing is raised exec is ok - - def test_legacy_file_exist_but_filesize(self): - """ - Test the legacy behaviour with the filesize - :return: - """ - # When - task = HdfsSensor(task_id='Should_be_file_legacy', - filepath='/datadirectory/datafile', - timeout=1, - file_size=20, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) - - # When - # Then - with self.assertRaises(AirflowSensorTimeout): - task.execute(None) - - def test_legacy_file_does_not_exists(self): - """ - Test the legacy behaviour - :return: - """ - task = HdfsSensor(task_id='Should_not_be_file_legacy', - filepath='/datadirectory/not_existing_file_or_directory', - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) - - # When - # Then - with self.assertRaises(AirflowSensorTimeout): - task.execute(None) + file_details = [ + { + 'kind': 'directory', + 'name': '/data/empty', + 'size': 0 + }, + { + 'kind': 'directory', + 'name': '/data/not_empty', + 'size': 0 + }, + { + 'kind': 'file', + 'name': '/data/not_empty/small.txt', + 'size': 10 + }, + { + 'kind': 'file', + 'name': '/data/not_empty/large.txt', + 'size': 10000000 + }, + { + 'kind': 'file', + 'name': '/data/not_empty/file.txt._COPYING_' + } + ] + + self._mock_client = MockHdfs3Client(file_details) + self._mock_params = models.Connection(conn_id='hdfs_default') + + # Setup mock for get_connection. + patcher = mock.patch.object( + HdfsHook, 'get_connection', return_value=self._mock_params) + self.addCleanup(patcher.stop) + patcher.start() + + # Setup mock for get_conn. + patcher = mock.patch.object( + HdfsHook, 'get_conn', return_value=self._mock_client) + self.addCleanup(patcher.stop) + patcher.start() + + self._default_task_kws = { + 'timeout': 1, + 'retry_delay': timedelta(seconds=1), + 'poke_interval': 1 + } + + def test_existing_file(self): + """Tests poking for existing file.""" + + task = HdfsSensor(task_id='existing_file', + file_path='/data/not_empty/small.txt', + **self._default_task_kws) + self.assertTrue(task.poke(context={})) + + def test_existing_file_glob(self): + """Tests poking for existing file with glob.""" + + task = HdfsSensor(task_id='existing_file', + file_path='/data/not_empty/*.txt', + **self._default_task_kws) + self.assertTrue(task.poke(context={})) + + def test_nonexisting_file(self): + """Tests poking for non-existing file.""" + + task = HdfsSensor(task_id='nonexisting_file', + file_path='/data/not_empty/random.txt', + **self._default_task_kws) + self.assertFalse(task.poke(context={})) + + def test_nonexisting_file_glob(self): + """Tests poking for non-existing file with glob.""" + + task = HdfsSensor(task_id='existing_file', + file_path='/data/not_empty/*.xml', + **self._default_task_kws) + self.assertFalse(task.poke(context={})) + + def test_nonexisting_path(self): + """Tests poking for non-existing path.""" + + task = HdfsSensor(task_id='nonexisting_path', + file_path='/data/not_empty/small.txt', + **self._default_task_kws) + + with mock.patch.object(self._mock_client, 'glob', side_effect=IOError): + self.assertFalse(task.poke(context={})) + + def test_file_filter_size_small(self): + """Tests poking for file while filtering for file size (too small).""" + + task = HdfsSensor(task_id='existing_file_too_small', + file_path='/data/not_empty/small.txt', + file_size=1, + **self._default_task_kws) + self.assertFalse(task.poke(context={})) + + def test_file_filter_size_large(self): + """Tests poking for file while filtering for file size (large).""" + + task = HdfsSensor(task_id='existing_file_large', + file_path='/data/not_empty/large.txt', + file_size=1, + **self._default_task_kws) + self.assertTrue(task.poke(context={})) + + def test_file_filter_ext(self): + """Tests poking for file while filtering for extension.""" + + task = HdfsSensor(task_id='existing_file_large', + file_path='/data/not_empty/f*', + ignored_ext=('_COPYING_', ), + **self._default_task_kws) + self.assertFalse(task.poke(context={})) + + +if __name__ == '__main__': + unittest.main() From 4792d04fc10b9d490cbc209b32d636435417328c Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Fri, 29 Jun 2018 15:52:03 +0200 Subject: [PATCH 03/15] Clean comment. --- tests/hooks/test_hdfs_hook.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/hooks/test_hdfs_hook.py b/tests/hooks/test_hdfs_hook.py index e783f801f376c..e0371c85b4133 100644 --- a/tests/hooks/test_hdfs_hook.py +++ b/tests/hooks/test_hdfs_hook.py @@ -16,7 +16,6 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -# import unittest From 9855bf4a2645359eda1320ded06d1f00fc8cfe6c Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Fri, 29 Jun 2018 22:40:11 +0200 Subject: [PATCH 04/15] Generalize filtering to filter functions. --- airflow/contrib/sensors/hdfs_sensor.py | 30 ++---- airflow/sensors/hdfs_sensor.py | 123 ++++++++++++++++++------- 2 files changed, 97 insertions(+), 56 deletions(-) diff --git a/airflow/contrib/sensors/hdfs_sensor.py b/airflow/contrib/sensors/hdfs_sensor.py index 832b81b8e5f25..853d5af6a476d 100644 --- a/airflow/contrib/sensors/hdfs_sensor.py +++ b/airflow/contrib/sensors/hdfs_sensor.py @@ -20,35 +20,17 @@ class HdfsSensorRegex(HdfsSensor): - def __init__(self, - regex, - *args, - **kwargs): - super(HdfsSensorRegex, self).__init__(*args, **kwargs) - self.regex = regex + """HdfsSensor subclass that filters using a specific regex.""" - def poke(self, context): - """ - poke matching files in a directory with self.regex + def __init__(self, regex, *args, **kwargs): - :return: Bool depending on the search criteria - """ - sb = self.hook(self.hdfs_conn_id).get_conn() - self.log.info( - 'Poking for {self.filepath} to be a directory ' - 'with files matching {self.regex.pattern}'. - format(**locals()) - ) - result = [f for f in sb.ls([self.filepath], include_toplevel=False) if - f['file_type'] == 'f' and - self.regex.match(f['path'].replace('%s/' % self.filepath, ''))] - result = self.filter_for_ignored_ext(result, self.ignored_ext, - self.ignore_copying) - result = self.filter_for_filesize(result, self.file_size) - return bool(result) + kwargs['filters'] = [lambda conn, fp: regex.match(fp) is not None] + super(HdfsSensorRegex, self).__init__(*args, **kwargs) class HdfsSensorFolder(HdfsSensor): + """HdfsSensor subclass that filters specifically for directories.""" + def __init__(self, be_empty=False, *args, diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index 1261e6629ada0..fa1c980f8173d 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -17,6 +17,7 @@ # specific language governing permissions and limitations # under the License. +import functools import posixpath import warnings @@ -26,30 +27,70 @@ from airflow.utils.decorators import apply_defaults +def deprecated_args(renamed=None, dropped=None): + """Decorator for the deprecation of renamed/removed keyword arguments. + + Wraps functions with changed keyword arguments (either renamed to new + arguments or dropped entirely). Functions calls with deprecated arguments + raise appropriate warnings. For removed arguments, any given values are + ignored (outside of the warning). For renamed arguments, values are + transparently proxied to their new argument names. + """ + + def decorator(function): + + @functools.wraps(function) + def wrapper(*args, **kwargs): + new_kwargs = {} + + for key in kwargs: + if key in dropped: + warnings.warn( + 'Argument {!r} is no longer supported and will be' + 'removed in a future version of Airflow.'.format(key), + category=DeprecationWarning) + elif key in renamed: + warnings.warn( + 'Argument {!r} has been renamed to {!r}. The old name ' + 'will no longer be supported in a future version of ' + 'Airflow.'.format(key, renamed[key]), + category=DeprecationWarning) + + new_kwargs[renamed[key]] = kwargs[key] + else: + new_kwargs[key] = kwargs[key] + + return function(*args, **new_kwargs) + return wrapper + return decorator + + class HdfsSensor(BaseSensorOperator): """ - Waits for a file or folder to land in HDFS + Waits for a file or folder to land in HDFS. """ template_fields = ('file_path',) ui_color = settings.WEB_COLORS['LIGHTBLUE'] + @deprecated_args(renamed={'filepath': 'file_path'}, + dropped={'ignore_copying'}) @apply_defaults def __init__(self, - file_path, + file_path=None, hdfs_conn_id='hdfs_default', + filters=None, + hook=HdfsHook, file_size=None, ignored_ext=('_COPYING_', ), - ignore_copying=None, - hook=HdfsHook, *args, **kwargs): - if ignore_copying is not None: - warnings.warn( - 'The ignore_copying argument is no longer used and will ' - 'be removed in the next version of Airflow.', - category=PendingDeprecationWarning) + if filters is None: + filters = [] + + filters = filters + self._default_filters( + min_size=file_size, ignored_exts=ignored_ext) super(HdfsSensor, self).__init__(*args, **kwargs) self.hdfs_conn_id = hdfs_conn_id @@ -59,6 +100,7 @@ def __init__(self, self.ignored_ext = set(ignored_ext) self.hook = hook + self._filters = filters @property def filepath(self): @@ -81,45 +123,62 @@ def poke(self, context): self.log.info('Files matching pattern: %s', file_paths) - if self.file_size: - file_paths = self._filter_for_size( - hdfs_conn, file_paths, min_size=self.file_size) - self.log.info('Files after filtering for size: %s', file_paths) - - if self.ignored_ext: - file_paths = self._filter_with_ext( - file_paths, exts=self.ignored_ext) - self.log.info('Files after filtering for extensions: %s', - file_paths) + file_paths = self._apply_filters(file_paths, self._filters, hdfs_conn) + self.log.info('Filters after filtering: %s', file_paths) return self._not_empty(file_paths) + @staticmethod + def _apply_filters(file_paths, filter_funcs, conn): + """Filters file paths that fail any of the filters (i.e. any + of the filter functions returns true for a given file). + """ + + for file_path in file_paths: + for func in filter_funcs: + if not func(conn, file_path): + break + else: + yield file_path + @staticmethod def _not_empty(iterable): + """Returns true if iterable is not empty.""" + try: next(iterable) return True except StopIteration: return False + @classmethod + def _default_filters(cls, min_size, ignored_exts): + """Returns default filters.""" + + filters = [] + + if min_size is not None: + filters.append( + lambda conn, fp: cls._filter_size( + conn, fp, min_size=min_size)) + + if ignored_exts: + filters.append(lambda conn, fp: cls._filter_ext(fp, ignored_exts)) + + return filters + @staticmethod - def _filter_for_size(hdfs_conn, file_paths, min_size): - """Filters file paths for a minimum file size.""" + def _filter_size(hdfs_conn, file_path, min_size): + """Filters any files below a minimum file size.""" + info = hdfs_conn.info(file_path) min_size_mb = min_size * settings.MEGABYTE - for file_path in file_paths: - info = hdfs_conn.info(file_path) - if info['kind'] == 'file' and info['size'] > min_size_mb: - yield file_path + return info['kind'] == 'file' and info['size'] > min_size_mb @staticmethod - def _filter_with_ext(file_paths, exts): + def _filter_ext(file_path, exts): """Filters any files with the given extensions.""" - for file_path in file_paths: - # Get file extension without preceding '.'. - file_ext = posixpath.splitext(file_path)[1][1:] - - if file_ext not in exts: - yield file_path + file_ext = posixpath.splitext(file_path)[1][1:] + return file_ext not in exts From 007e8c3ee2cff42cb5d333f570ed21e3498d86a4 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Sat, 30 Jun 2018 00:10:48 +0200 Subject: [PATCH 05/15] Rename file_size arg. --- airflow/sensors/hdfs_sensor.py | 15 ++++++++------- tests/sensors/test_hdfs_sensor.py | 4 ++-- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index fa1c980f8173d..fed61f32d3789 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -73,7 +73,8 @@ class HdfsSensor(BaseSensorOperator): template_fields = ('file_path',) ui_color = settings.WEB_COLORS['LIGHTBLUE'] - @deprecated_args(renamed={'filepath': 'file_path'}, + @deprecated_args(renamed={'filepath': 'file_path', + 'file_size': 'min_size'}, dropped={'ignore_copying'}) @apply_defaults def __init__(self, @@ -81,7 +82,7 @@ def __init__(self, hdfs_conn_id='hdfs_default', filters=None, hook=HdfsHook, - file_size=None, + min_size=None, ignored_ext=('_COPYING_', ), *args, **kwargs): @@ -89,14 +90,14 @@ def __init__(self, if filters is None: filters = [] - filters = filters + self._default_filters( - min_size=file_size, ignored_exts=ignored_ext) + filters = self._setup_filters( + filters, min_size=min_size, ignored_exts=ignored_ext) super(HdfsSensor, self).__init__(*args, **kwargs) self.hdfs_conn_id = hdfs_conn_id self.file_path = file_path - self.file_size = file_size + self.min_size = min_size self.ignored_ext = set(ignored_ext) self.hook = hook @@ -152,10 +153,10 @@ def _not_empty(iterable): return False @classmethod - def _default_filters(cls, min_size, ignored_exts): + def _setup_filters(cls, filters, min_size, ignored_exts): """Returns default filters.""" - filters = [] + filters = list(filters) if min_size is not None: filters.append( diff --git a/tests/sensors/test_hdfs_sensor.py b/tests/sensors/test_hdfs_sensor.py index 12c1e4b453ff4..dd3b09cf3a780 100644 --- a/tests/sensors/test_hdfs_sensor.py +++ b/tests/sensors/test_hdfs_sensor.py @@ -147,7 +147,7 @@ def test_file_filter_size_small(self): task = HdfsSensor(task_id='existing_file_too_small', file_path='/data/not_empty/small.txt', - file_size=1, + min_size=1, **self._default_task_kws) self.assertFalse(task.poke(context={})) @@ -156,7 +156,7 @@ def test_file_filter_size_large(self): task = HdfsSensor(task_id='existing_file_large', file_path='/data/not_empty/large.txt', - file_size=1, + min_size=1, **self._default_task_kws) self.assertTrue(task.poke(context={})) From 12da2e57e390a5f3e58c4135bdf4a8e8f9445cc3 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Sun, 15 Jul 2018 11:33:29 +0200 Subject: [PATCH 06/15] Add ha parameters + kerberos config. --- airflow/hooks/hdfs_hook.py | 52 ++++++++++++++++++++++++++++++++------ 1 file changed, 44 insertions(+), 8 deletions(-) diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index 411a7697d5169..37731e8d9883a 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -22,6 +22,7 @@ import hdfs3 from hdfs3.utils import MyNone +from airflow import configuration from airflow.hooks.base_hook import BaseHook @@ -32,8 +33,34 @@ class HdfsHook(BaseHook): `hdfs-site.xml` if these files can be found in any of the typical locations. The hook loads `host` and `port` parameters from the hdfs connection (if given) and extra configuration parameters can be - supplied using the `pars` key in extra JSON. See the hdfs3 documentation - for more details. + supplied in the connections extra JSON as follows: + + { + "pars": { + "dfs.domain.socket.path": "/var/lib/hadoop-hdfs/dn_socket" + }, + "ha": { + "host": "nameservice1", + "conf": { + "dfs.nameservices": "nameservice1", + "dfs.ha.namenodes.nameservice1": "namenode113,namenode188", + "dfs.namenode.rpc-address.nameservice1.namenode113": "host1:8020", + "dfs.namenode.rpc-address.nameservice1.namenode188": "host2:8020", + "dfs.namenode.http-address.nameservice1.namenode113": "host1:50070", + "dfs.namenode.http-address.nameservice1.namenode188": "host2:50070" + } + } + } + + Here `pars` can be used to supply configuration options with the same key + names as typically contained in the XML config files, which will take + precedence over any parameters loaded from files. The `ha` configuration + section can be used to supply options for using hdfs3 in high-availability + mode. See the hdfs3 documentation for more details. + + Security modes can also be configured by defining appropriate value for + the `hadoop.security.authentication` key in `pars`. Kerberos is used + automatically if Airflow has been configured to use kerberos. :param str hdfs_conn_id: Connection ID to fetch parameters from. :param bool autoconf: Whether to use autoconfig to discover @@ -54,20 +81,29 @@ def get_conn(self): self._conn = hdfs3.HDFileSystem(autoconf=self._autoconf) else: params = self.get_connection(self.hdfs_conn_id) + extra_params = params.extra_dejson # Extract hadoop parameters from extra. - hdfs_pars = params.extra_dejson.get('pars', {}) + hdfs_params = extra_params.get("pars", {}) + + # Configure kerberos security if used by Airflow. + if configuration.conf.get("core", "security") == "kerberos": + hdfs_params["hadoop.security.authentication"] = "kerberos" + + # Extract high-availability config if given. + ha_params = extra_params.get("ha", {}) + hdfs_params.update(ha_params.get("conf", {})) # Collect extra parameters to pass to kwargs. extra_kws = {} if params.login: - extra_kws['user'] = params.login + extra_kws["user"] = params.login # Build connection. self._conn = hdfs3.HDFileSystem( - host=params.host or MyNone, + host=ha_params.get("host") or params.host or MyNone, port=params.port or MyNone, - pars=hdfs_pars, + pars=hdfs_params, autoconf=self._autoconf, **extra_kws) @@ -107,5 +143,5 @@ def __getattr__(self, attr): HDFSHook = _DeprecationHelper( HdfsHook, - message='The `HDFSHook` has been renamed to `HdfsHook`. Support for ' - 'the old naming will be dropped in a future version of Airflow.') + message="The `HDFSHook` has been renamed to `HdfsHook`. Support for " + "the old naming will be dropped in a future version of Airflow.") From 50b3fad675a6b78496548bf8195c78823c776280 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Wed, 18 Jul 2018 16:49:32 +0200 Subject: [PATCH 07/15] Update regex sensor. --- airflow/contrib/sensors/hdfs_sensor.py | 26 +- airflow/sensors/hdfs_sensor.py | 166 +++++---- tests/contrib/sensors/test_hdfs_sensor.py | 394 +++++++++++----------- tests/sensors/test_hdfs_sensor.py | 38 ++- 4 files changed, 352 insertions(+), 272 deletions(-) diff --git a/airflow/contrib/sensors/hdfs_sensor.py b/airflow/contrib/sensors/hdfs_sensor.py index 853d5af6a476d..aaa6ea79bbd63 100644 --- a/airflow/contrib/sensors/hdfs_sensor.py +++ b/airflow/contrib/sensors/hdfs_sensor.py @@ -16,16 +16,36 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. + +import posixpath + from airflow.sensors.hdfs_sensor import HdfsSensor class HdfsSensorRegex(HdfsSensor): """HdfsSensor subclass that filters using a specific regex.""" - def __init__(self, regex, *args, **kwargs): + def __init__(self, file_pattern, regex, *args, **kwargs): + if not self._is_pattern(file_pattern): + # If file path is not a pattern, we assume it is a directory + # containing files that we want to match the regex against. + # This matches the legacy behaviour of the sensor. + file_pattern = posixpath.join(file_pattern, '*') + + def _filter_regex(_, file_path): + file_name = posixpath.basename(file_path) + return regex.match(file_name) is not None + + super(HdfsSensorRegex, self).__init__( + *args, + file_pattern=file_pattern, + extra_filters=[_filter_regex], + **kwargs) - kwargs['filters'] = [lambda conn, fp: regex.match(fp) is not None] - super(HdfsSensorRegex, self).__init__(*args, **kwargs) + @staticmethod + def _is_pattern(path_): + """Checks if given path contains any glob patterns.""" + return '*' in path_ or '[' in path_ class HdfsSensorFolder(HdfsSensor): diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index fed61f32d3789..05502f4c7cd2e 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -65,6 +65,29 @@ def wrapper(*args, **kwargs): return decorator +def deprecated(new_name=None): + """This is a decorator which can be used to mark functions + as deprecated. It will result in a warning being emitted + when the function is used. + """ + + def decorator(function): + @functools.wraps(function) + def wrapper(*args, **kwargs): + if new_name: + message = ("{} has been deprecated and will be replaced by " + "{} in a future version of Airflow." + .format(function.__name__, new_name)) + else: + message = ("{} has been deprecated and will be removed " + "in a future version of Airflow." + .format(function.__name__)) + warnings.warn(message, category=DeprecationWarning) + return function(*args, **kwargs) + return wrapper + return decorator + + class HdfsSensor(BaseSensorOperator): """ Waits for a file or folder to land in HDFS. @@ -73,71 +96,124 @@ class HdfsSensor(BaseSensorOperator): template_fields = ('file_path',) ui_color = settings.WEB_COLORS['LIGHTBLUE'] - @deprecated_args(renamed={'filepath': 'file_path', - 'file_size': 'min_size'}, + @deprecated_args(renamed={'filepath': 'file_pattern', + 'file_size': 'min_size', + 'ignored_ext': 'ignore_exts'}, dropped={'ignore_copying'}) @apply_defaults def __init__(self, - file_path=None, + file_pattern, hdfs_conn_id='hdfs_default', - filters=None, - hook=HdfsHook, min_size=None, - ignored_ext=('_COPYING_', ), + ignore_exts=('_COPYING_', ), + extra_filters=None, + hook=HdfsHook, *args, **kwargs): + super(HdfsSensor, self).__init__(*args, **kwargs) - if filters is None: - filters = [] - - filters = self._setup_filters( - filters, min_size=min_size, ignored_exts=ignored_ext) + default_filters = self._default_filters( + min_size=min_size, ignore_exts=ignore_exts) + filters = default_filters + (extra_filters or []) - super(HdfsSensor, self).__init__(*args, **kwargs) - self.hdfs_conn_id = hdfs_conn_id - self.file_path = file_path + self._file_pattern = file_pattern + self._conn_id = hdfs_conn_id - self.min_size = min_size - self.ignored_ext = set(ignored_ext) + self._min_size = min_size + self._ignore_exts = set(ignore_exts) - self.hook = hook + self._hook = hook(hdfs_conn_id) self._filters = filters + @property + def file_pattern(self): + """File pattern (glob) that the sensor matches against.""" + return self._file_pattern + + @property + def conn_id(self): + """Connection ID used by the sensor.""" + return self._conn_id + + @deprecated(new_name="file_pattern") @property def filepath(self): - warnings.warn( - 'The `filepath` property has been renamed to `file_path`. ' - 'Support for the old accessor will be dropped in the next ' - 'version of Airflow.', - category=PendingDeprecationWarning) - return self.file_path + return self.file_pattern + + @deprecated(new_name="conn_id") + @property + def hdfs_conn_id(self): + return self.conn_id + + @deprecated() + @property + def min_size(self): + return self._min_size + + @deprecated() + @property + def ignored_ext(self): + return self._ignore_exts + + @classmethod + def _default_filters(cls, min_size=None, ignore_exts=None): + filters = [] + + if min_size is not None: + def _size_filter(hook, file_path): + return cls._filter_size(hook, file_path, min_size=min_size) + filters.append(_size_filter) + + if ignore_exts: + def _ext_filter(hook, file_path): + return cls._filter_ext(hook, file_path, exts=ignore_exts) + filters.append(_ext_filter) + + return filters + + @staticmethod + def _filter_size(hook, file_path, min_size): + """Filters any files below a minimum file size.""" + + info = hook.get_conn().info(file_path) + min_size_mb = min_size * settings.MEGABYTE + + return info['kind'] == 'file' and info['size'] > min_size_mb + + @staticmethod + def _filter_ext(_, file_path, exts): + """Filters any files with the given extensions.""" + + file_ext = posixpath.splitext(file_path)[1][1:] + return file_ext not in exts def poke(self, context): - self.log.info('Poking for file %s', self.file_path) - hdfs_conn = self.hook(self.hdfs_conn_id).get_conn() + self.log.info('Poking for file pattern %s', self.file_pattern) + hdfs_conn = self._hook.get_conn() try: - file_paths = hdfs_conn.glob(self.file_path) + file_paths = hdfs_conn.glob(self.file_pattern) except IOError: # File path doesn't exist yet. return False self.log.info('Files matching pattern: %s', file_paths) - file_paths = self._apply_filters(file_paths, self._filters, hdfs_conn) + file_paths = self._apply_filters( + file_paths, self._filters, hook=self._hook) self.log.info('Filters after filtering: %s', file_paths) return self._not_empty(file_paths) @staticmethod - def _apply_filters(file_paths, filter_funcs, conn): + def _apply_filters(file_paths, filter_funcs, hook): """Filters file paths that fail any of the filters (i.e. any of the filter functions returns true for a given file). """ for file_path in file_paths: for func in filter_funcs: - if not func(conn, file_path): + if not func(hook, file_path): break else: yield file_path @@ -151,35 +227,3 @@ def _not_empty(iterable): return True except StopIteration: return False - - @classmethod - def _setup_filters(cls, filters, min_size, ignored_exts): - """Returns default filters.""" - - filters = list(filters) - - if min_size is not None: - filters.append( - lambda conn, fp: cls._filter_size( - conn, fp, min_size=min_size)) - - if ignored_exts: - filters.append(lambda conn, fp: cls._filter_ext(fp, ignored_exts)) - - return filters - - @staticmethod - def _filter_size(hdfs_conn, file_path, min_size): - """Filters any files below a minimum file size.""" - - info = hdfs_conn.info(file_path) - min_size_mb = min_size * settings.MEGABYTE - - return info['kind'] == 'file' and info['size'] > min_size_mb - - @staticmethod - def _filter_ext(file_path, exts): - """Filters any files with the given extensions.""" - - file_ext = posixpath.splitext(file_path)[1][1:] - return file_ext not in exts diff --git a/tests/contrib/sensors/test_hdfs_sensor.py b/tests/contrib/sensors/test_hdfs_sensor.py index b03b738686ed8..7036f7360a025 100644 --- a/tests/contrib/sensors/test_hdfs_sensor.py +++ b/tests/contrib/sensors/test_hdfs_sensor.py @@ -16,186 +16,172 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. + import logging +import mock import unittest import re from datetime import timedelta +from airflow import models from airflow.contrib.sensors.hdfs_sensor import HdfsSensorFolder, HdfsSensorRegex from airflow.exceptions import AirflowSensorTimeout +from airflow.hooks.hdfs_hook import HdfsHook +from tests.sensors.test_hdfs_sensor import MockHdfs3Client + + +class HdfsSensorRegexTests(unittest.TestCase): + """Tests for the HdfsSensorRegex class.""" -class HdfsSensorFolderTests(unittest.TestCase): def setUp(self): - from tests.core import FakeHDFSHook - self.hook = FakeHDFSHook - self.log = logging.getLogger() - self.log.setLevel(logging.DEBUG) + file_details = [ + { + 'kind': 'directory', + 'name': '/data', + 'size': 0 + }, + { + 'kind': 'file', + 'name': '/data/test1file', + 'size': 2000000 + }, + { + 'kind': 'file', + 'name': '/data/copying._COPYING_', + 'size': 2000000 + } + ] - def test_should_be_empty_directory(self): - """ - test the empty directory behaviour - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - task = HdfsSensorFolder(task_id='Should_be_empty_directory', - filepath='/datadirectory/empty_directory', - be_empty=True, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) + self._mock_client, self._mock_params = \ + MockHdfs3Client.from_file_details(file_details, test_instance=self) - # When - task.execute(None) + self._default_task_kws = { + 'timeout': 1, + 'retry_delay': timedelta(seconds=1), + 'poke_interval': 1 + } - # Then - # Nothing happens, nothing is raised exec is ok + def test_should_match_regex(self): + """Tests example where files should match regex.""" - def test_should_be_empty_directory_fail(self): - """ - test the empty directory behaviour - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - task = HdfsSensorFolder(task_id='Should_be_empty_directory_fail', - filepath='/datadirectory/not_empty_directory', - be_empty=True, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) + regex = re.compile("test[1-2]file") + task = HdfsSensorRegex(task_id='should_match_the_regex', + file_pattern='/data/*', + regex=regex, + **self._default_task_kws) + self.assertTrue(task.poke(context={})) - # When - # Then - with self.assertRaises(AirflowSensorTimeout): - task.execute(None) - - def test_should_be_a_non_empty_directory(self): - """ - test the empty directory behaviour - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - task = HdfsSensorFolder(task_id='Should_be_non_empty_directory', - filepath='/datadirectory/not_empty_directory', - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) + def test_should_match_regex_dir(self): + """Tests example where files should match regex with dir path.""" - # When - task.execute(None) + regex = re.compile("test[1-2]file") + task = HdfsSensorRegex(task_id='should_match_the_regex', + file_pattern='/data', + regex=regex, + **self._default_task_kws) + self.assertTrue(task.poke(context={})) - # Then - # Nothing happens, nothing is raised exec is ok + def test_should_not_match_regex(self): + """Tests example where files should match regex.""" - def test_should_be_non_empty_directory_fail(self): - """ - test the empty directory behaviour - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - task = HdfsSensorFolder(task_id='Should_be_empty_directory_fail', - filepath='/datadirectory/empty_directory', - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) + regex = re.compile("^IDoNotExist") + task = HdfsSensorRegex(task_id='should_not_match_the_regex', + file_pattern='/data/*', + regex=regex, + **self._default_task_kws) + self.assertFalse(task.poke(context={})) - # When - # Then - with self.assertRaises(AirflowSensorTimeout): - task.execute(None) + def test_should_match_regex_and_size(self): + """Tests example with matching regex and sufficient file size.""" + regex = re.compile("test[1-2]file") + task = HdfsSensorRegex(task_id='should_match_the_regex_and_size', + file_pattern='/data/*', + regex=regex, + min_size=1, + **self._default_task_kws) + self.assertTrue(task.poke(context={})) -class HdfsSensorRegexTests(unittest.TestCase): - def setUp(self): - from tests.core import FakeHDFSHook - self.hook = FakeHDFSHook - self.log = logging.getLogger() - self.log.setLevel(logging.DEBUG) + def test_should_match_regex_not_size(self): + """Tests example with matching regex but too small file size.""" - def test_should_match_regex(self): - """ - test the empty directory behaviour - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - compiled_regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='Should_match_the_regex', - filepath='/datadirectory/regex_dir', - regex=compiled_regex, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) + regex = re.compile("test[1-2]file") + task = HdfsSensorRegex(task_id='should_match_the_regex_but_not_size', + file_pattern='/data/*', + regex=regex, + min_size=10, + **self._default_task_kws) + self.assertFalse(task.poke(context={})) - # When - task.execute(None) + def test_should_match_regex_not_ext(self): + """Tests example with matching regex but wrong ext.""" - # Then - # Nothing happens, nothing is raised exec is ok + regex = re.compile("test[1-2]file") + task = HdfsSensorRegex(task_id='should_match_the_regex_but_not_size', + file_pattern='/data/*', + regex=regex, + min_size=10, + **self._default_task_kws) + self.assertFalse(task.poke(context={})) - def test_should_not_match_regex(self): - """ - test the empty directory behaviour - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - compiled_regex = re.compile("^IDoNotExist") - task = HdfsSensorRegex(task_id='Should_not_match_the_regex', - filepath='/datadirectory/regex_dir', + compiled_regex = re.compile("copying.*") + task = HdfsSensorRegex(task_id='should_match_the_regex_but_not_ext', + file_pattern='/data/*', regex=compiled_regex, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) + ignore_exts=['_COPYING_'], + **self._default_task_kws) + self.assertFalse(task.poke(context={})) + + +class HdfsSensorFolderTests(unittest.TestCase): + + def setUp(self): + file_details = [ + { + 'kind': 'directory', + 'name': '/empty', + 'size': 0 + }, + { + 'kind': 'directory', + 'name': '/not_empty', + 'size': 0 + }, + { + 'kind': 'file', + 'name': '/not_empty/test1file', + 'size': 2000000 + } + ] + + self._mock_client, self._mock_params = \ + MockHdfs3Client.from_file_details(file_details, test_instance=self) + + self._default_task_kws = { + 'timeout': 1, + 'retry_delay': timedelta(seconds=1), + 'poke_interval': 1 + } + + self._mock_client = MockHdfs3Client(file_details) + self._mock_params = models.Connection(conn_id='hdfs_default') + + def test_should_be_empty_directory(self): + """Tests example with empty directory.""" - # When - # Then - with self.assertRaises(AirflowSensorTimeout): - task.execute(None) - - def test_should_match_regex_and_filesize(self): - """ - test the file size behaviour with regex - :return: - """ # Given self.log.debug('#' * 10) self.log.debug('Running %s', self._testMethodName) self.log.debug('#' * 10) - compiled_regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='Should_match_the_regex_and_filesize', - filepath='/datadirectory/regex_dir', - regex=compiled_regex, - ignore_copying=True, - ignored_ext=['_COPYING_', 'sftp'], - file_size=10, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) + task = HdfsSensorFolder(task_id='Should_be_empty_directory', + filepath='/datadirectory/empty_directory', + be_empty=True, + timeout=1, + retry_delay=timedelta(seconds=1), + poke_interval=1, + hook=self.hook) # When task.execute(None) @@ -203,51 +189,71 @@ def test_should_match_regex_and_filesize(self): # Then # Nothing happens, nothing is raised exec is ok - def test_should_match_regex_but_filesize(self): - """ - test the file size behaviour with regex - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - compiled_regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='Should_match_the_regex_but_filesize', - filepath='/datadirectory/regex_dir', - regex=compiled_regex, - file_size=20, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) +# def test_should_be_empty_directory_fail(self): +# """ +# test the empty directory behaviour +# :return: +# """ +# # Given +# self.log.debug('#' * 10) +# self.log.debug('Running %s', self._testMethodName) +# self.log.debug('#' * 10) +# task = HdfsSensorFolder(task_id='Should_be_empty_directory_fail', +# filepath='/datadirectory/not_empty_directory', +# be_empty=True, +# timeout=1, +# retry_delay=timedelta(seconds=1), +# poke_interval=1, +# hook=self.hook) - # When - # Then - with self.assertRaises(AirflowSensorTimeout): - task.execute(None) - - def test_should_match_regex_but_copyingext(self): - """ - test the file size behaviour with regex - :return: - """ - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - compiled_regex = re.compile(r"copying_file_\d+.txt") - task = HdfsSensorRegex(task_id='Should_match_the_regex_but_filesize', - filepath='/datadirectory/regex_dir', - regex=compiled_regex, - ignored_ext=['_COPYING_', 'sftp'], - file_size=20, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) +# # When +# # Then +# with self.assertRaises(AirflowSensorTimeout): +# task.execute(None) - # When - # Then - with self.assertRaises(AirflowSensorTimeout): - task.execute(None) +# def test_should_be_a_non_empty_directory(self): +# """ +# test the empty directory behaviour +# :return: +# """ +# # Given +# self.log.debug('#' * 10) +# self.log.debug('Running %s', self._testMethodName) +# self.log.debug('#' * 10) +# task = HdfsSensorFolder(task_id='Should_be_non_empty_directory', +# filepath='/datadirectory/not_empty_directory', +# timeout=1, +# retry_delay=timedelta(seconds=1), +# poke_interval=1, +# hook=self.hook) + +# # When +# task.execute(None) + +# # Then +# # Nothing happens, nothing is raised exec is ok + +# def test_should_be_non_empty_directory_fail(self): +# """ +# test the empty directory behaviour +# :return: +# """ +# # Given +# self.log.debug('#' * 10) +# self.log.debug('Running %s', self._testMethodName) +# self.log.debug('#' * 10) +# task = HdfsSensorFolder(task_id='Should_be_empty_directory_fail', +# filepath='/datadirectory/empty_directory', +# timeout=1, +# retry_delay=timedelta(seconds=1), +# poke_interval=1, +# hook=self.hook) + +# # When +# # Then +# with self.assertRaises(AirflowSensorTimeout): +# task.execute(None) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/sensors/test_hdfs_sensor.py b/tests/sensors/test_hdfs_sensor.py index dd3b09cf3a780..f28c232f9c45b 100644 --- a/tests/sensors/test_hdfs_sensor.py +++ b/tests/sensors/test_hdfs_sensor.py @@ -35,6 +35,28 @@ def __init__(self, file_details): entry['name']: entry for entry in file_details } + @classmethod + def from_file_details(cls, file_details, test_instance, + conn_id='hdfs_default'): + """Builds mock hdsf3 client using file_details.""" + + mock_client = cls(file_details) + mock_params = models.Connection(conn_id=conn_id) + + # Setup mock for get_connection. + patcher = mock.patch.object( + HdfsHook, 'get_connection', return_value=mock_params) + test_instance.addCleanup(patcher.stop) + patcher.start() + + # Setup mock for get_conn. + patcher = mock.patch.object( + HdfsHook, 'get_conn', return_value=mock_client) + test_instance.addCleanup(patcher.stop) + patcher.start() + + return mock_client, mock_params + def glob(self, pattern): """Returns glob of files matching pattern.""" return fnmatch.filter(self._file_details.keys(), pattern) @@ -79,20 +101,8 @@ def setUp(self): } ] - self._mock_client = MockHdfs3Client(file_details) - self._mock_params = models.Connection(conn_id='hdfs_default') - - # Setup mock for get_connection. - patcher = mock.patch.object( - HdfsHook, 'get_connection', return_value=self._mock_params) - self.addCleanup(patcher.stop) - patcher.start() - - # Setup mock for get_conn. - patcher = mock.patch.object( - HdfsHook, 'get_conn', return_value=self._mock_client) - self.addCleanup(patcher.stop) - patcher.start() + self._mock_client, self._mock_params = \ + MockHdfs3Client.from_file_details(file_details, test_instance=self) self._default_task_kws = { 'timeout': 1, From 7eaca1383e6bc0ec90592966285a138a9e27e8e9 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Mon, 23 Jul 2018 14:24:17 +0200 Subject: [PATCH 08/15] Split HdfsSensor into HdfsFile- and HdfsFolder sensor. --- airflow/contrib/sensors/hdfs_sensor.py | 69 ++-- airflow/hooks/hdfs_hook.py | 24 +- airflow/sensors/hdfs_sensor.py | 323 ++++++++++--------- airflow/utils/deprecation.py | 157 ++++++++++ tests/contrib/sensors/test_hdfs_sensor.py | 247 ++++++--------- tests/sensors/test_hdfs_sensor.py | 365 +++++++++++++++++----- 6 files changed, 746 insertions(+), 439 deletions(-) create mode 100644 airflow/utils/deprecation.py diff --git a/airflow/contrib/sensors/hdfs_sensor.py b/airflow/contrib/sensors/hdfs_sensor.py index aaa6ea79bbd63..015a402b8429b 100644 --- a/airflow/contrib/sensors/hdfs_sensor.py +++ b/airflow/contrib/sensors/hdfs_sensor.py @@ -17,64 +17,49 @@ # specific language governing permissions and limitations # under the License. +from functools import partial import posixpath -from airflow.sensors.hdfs_sensor import HdfsSensor +from airflow.sensors import hdfs_sensor +from airflow.utils.deprecation import RenamedClass -class HdfsSensorRegex(HdfsSensor): +class HdfsRegexFileSensor(hdfs_sensor.HdfsFileSensor): """HdfsSensor subclass that filters using a specific regex.""" - def __init__(self, file_pattern, regex, *args, **kwargs): - if not self._is_pattern(file_pattern): + def __init__(self, pattern, regex, **kwargs): + if not self._is_pattern(pattern): # If file path is not a pattern, we assume it is a directory # containing files that we want to match the regex against. # This matches the legacy behaviour of the sensor. - file_pattern = posixpath.join(file_pattern, '*') + pattern = posixpath.join(pattern, "*") - def _filter_regex(_, file_path): - file_name = posixpath.basename(file_path) - return regex.match(file_name) is not None - - super(HdfsSensorRegex, self).__init__( - *args, - file_pattern=file_pattern, - extra_filters=[_filter_regex], - **kwargs) + super(HdfsRegexFileSensor, self).__init__( + pattern=pattern, + filters=[partial(filter_regex, regex=regex)], + **kwargs + ) @staticmethod def _is_pattern(path_): """Checks if given path contains any glob patterns.""" - return '*' in path_ or '[' in path_ + return "*" in path_ or "[" in path_ + +def filter_regex(_, file_paths, regex): + """Filters file paths for given regex.""" -class HdfsSensorFolder(HdfsSensor): - """HdfsSensor subclass that filters specifically for directories.""" + for file_path in file_paths: + if regex.match(posixpath.basename(file_path)): + yield file_path - def __init__(self, - be_empty=False, - *args, - **kwargs): - super(HdfsSensorFolder, self).__init__(*args, **kwargs) - self.be_empty = be_empty - def poke(self, context): - """ - poke for a non empty directory +HdfsSensorRegex = RenamedClass( + "HdfsSensorRegex", new_class=HdfsRegexFileSensor, old_module=__name__ +) - :return: Bool depending on the search criteria - """ - sb = self.hook(self.hdfs_conn_id).get_conn() - result = [f for f in sb.ls([self.filepath], include_toplevel=True)] - result = self.filter_for_ignored_ext(result, self.ignored_ext, - self.ignore_copying) - result = self.filter_for_filesize(result, self.file_size) - if self.be_empty: - self.log.info('Poking for filepath {self.filepath} to a empty directory' - .format(**locals())) - return len(result) == 1 and result[0]['path'] == self.filepath - else: - self.log.info('Poking for filepath {self.filepath} to a non empty directory' - .format(**locals())) - result.pop(0) - return bool(result) and result[0]['file_type'] == 'f' +HdfsSensorFolder = RenamedClass( + "HdfsSensorFolder", + new_class=hdfs_sensor.HdfsFolderSensor, + old_module=__name__ +) diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index 37731e8d9883a..c3d7e19e64865 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -24,6 +24,7 @@ from airflow import configuration from airflow.hooks.base_hook import BaseHook +from airflow.utils.deprecation import RenamedClass class HdfsHook(BaseHook): @@ -123,25 +124,4 @@ def close(self): self._conn = None -class _DeprecationHelper(object): - def __init__(self, new_target, message, category=PendingDeprecationWarning): - self._message = message - self._new_target = new_target - self._category = category - - def _warn(self): - warnings.warn(self._message, category=self._category) - - def __call__(self, *args, **kwargs): - self._warn() - return self._new_target(*args, **kwargs) - - def __getattr__(self, attr): - self._warn() - return getattr(self._new_target, attr) - - -HDFSHook = _DeprecationHelper( - HdfsHook, - message="The `HDFSHook` has been renamed to `HdfsHook`. Support for " - "the old naming will be dropped in a future version of Airflow.") +HDFSHook = RenamedClass('HDFSHook', new_class=HdfsHook, old_module=__name__) diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index 05502f4c7cd2e..7e54479e8ad23 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -19,131 +19,76 @@ import functools import posixpath -import warnings from airflow import settings from airflow.hooks.hdfs_hook import HdfsHook from airflow.sensors.base_sensor_operator import BaseSensorOperator from airflow.utils.decorators import apply_defaults +from airflow.utils.deprecation import deprecated_args, deprecated, RenamedClass -def deprecated_args(renamed=None, dropped=None): - """Decorator for the deprecation of renamed/removed keyword arguments. - - Wraps functions with changed keyword arguments (either renamed to new - arguments or dropped entirely). Functions calls with deprecated arguments - raise appropriate warnings. For removed arguments, any given values are - ignored (outside of the warning). For renamed arguments, values are - transparently proxied to their new argument names. - """ - - def decorator(function): - - @functools.wraps(function) - def wrapper(*args, **kwargs): - new_kwargs = {} - - for key in kwargs: - if key in dropped: - warnings.warn( - 'Argument {!r} is no longer supported and will be' - 'removed in a future version of Airflow.'.format(key), - category=DeprecationWarning) - elif key in renamed: - warnings.warn( - 'Argument {!r} has been renamed to {!r}. The old name ' - 'will no longer be supported in a future version of ' - 'Airflow.'.format(key, renamed[key]), - category=DeprecationWarning) - - new_kwargs[renamed[key]] = kwargs[key] - else: - new_kwargs[key] = kwargs[key] - - return function(*args, **new_kwargs) - return wrapper - return decorator - - -def deprecated(new_name=None): - """This is a decorator which can be used to mark functions - as deprecated. It will result in a warning being emitted - when the function is used. - """ - - def decorator(function): - @functools.wraps(function) - def wrapper(*args, **kwargs): - if new_name: - message = ("{} has been deprecated and will be replaced by " - "{} in a future version of Airflow." - .format(function.__name__, new_name)) - else: - message = ("{} has been deprecated and will be removed " - "in a future version of Airflow." - .format(function.__name__)) - warnings.warn(message, category=DeprecationWarning) - return function(*args, **kwargs) - return wrapper - return decorator - - -class HdfsSensor(BaseSensorOperator): - """ - Waits for a file or folder to land in HDFS. - """ - - template_fields = ('file_path',) - ui_color = settings.WEB_COLORS['LIGHTBLUE'] - - @deprecated_args(renamed={'filepath': 'file_pattern', - 'file_size': 'min_size', - 'ignored_ext': 'ignore_exts'}, - dropped={'ignore_copying'}) - @apply_defaults - def __init__(self, - file_pattern, - hdfs_conn_id='hdfs_default', - min_size=None, - ignore_exts=('_COPYING_', ), - extra_filters=None, - hook=HdfsHook, - *args, - **kwargs): - super(HdfsSensor, self).__init__(*args, **kwargs) +class HdfsFileSensor(BaseSensorOperator): + """Waits for file(s) to land in HDFS.""" + + template_fields = ("_pattern",) + ui_color = settings.WEB_COLORS["LIGHTBLUE"] + @deprecated_args( + renamed={ + "filepath": "file_pattern", + "hdfs_conn_id": "conn_id", + "file_size": "min_size", + "ignored_ext": "ignore_exts", + }, + dropped={"ignore_copying", "hook"}, + ) + @apply_defaults + def __init__( + self, + pattern, + conn_id="hdfs_default", + filters=None, + min_size=None, + ignore_exts=("_COPYING_",), + **kwargs + ): + super(HdfsFileSensor, self).__init__(**kwargs) + + # Min-size and ignore-ext filters are added via + # arguments for backwards compatibility. default_filters = self._default_filters( - min_size=min_size, ignore_exts=ignore_exts) - filters = default_filters + (extra_filters or []) + min_size=min_size, ignore_exts=ignore_exts + ) + filters = default_filters + (filters or []) - self._file_pattern = file_pattern - self._conn_id = hdfs_conn_id + self._pattern = pattern + self._conn_id = conn_id + self._filters = filters self._min_size = min_size self._ignore_exts = set(ignore_exts) - self._hook = hook(hdfs_conn_id) - self._filters = filters - @property - def file_pattern(self): + def pattern(self): """File pattern (glob) that the sensor matches against.""" - return self._file_pattern + return self._pattern @property def conn_id(self): - """Connection ID used by the sensor.""" + """ID of connection used by the sensor.""" return self._conn_id + # Deprecated properties that exist for backwards compatibility. + @deprecated(new_name="file_pattern") @property def filepath(self): - return self.file_pattern + return self._pattern @deprecated(new_name="conn_id") @property def hdfs_conn_id(self): - return self.conn_id + return self._conn_id @deprecated() @property @@ -160,70 +105,142 @@ def _default_filters(cls, min_size=None, ignore_exts=None): filters = [] if min_size is not None: - def _size_filter(hook, file_path): - return cls._filter_size(hook, file_path, min_size=min_size) - filters.append(_size_filter) + filters.append(functools.partial(filter_by_size, min_size=min_size)) if ignore_exts: - def _ext_filter(hook, file_path): - return cls._filter_ext(hook, file_path, exts=ignore_exts) - filters.append(_ext_filter) + filters.append(functools.partial(filter_for_exts, exts=ignore_exts)) return filters - @staticmethod - def _filter_size(hook, file_path, min_size): - """Filters any files below a minimum file size.""" + def poke(self, context): + with HdfsHook(self._conn_id) as hook: + conn = hook.get_conn() + + # Fetch files matching glob pattern. + self.log.info("Poking for file pattern %s", self._pattern) + + try: + file_paths = [ + fp for fp in conn.glob(self._pattern) if not conn.isdir(fp) + ] + except IOError: + # File path doesn't exist yet. + file_paths = [] + + self.log.info("Files matching pattern: %s", file_paths) + + # Filter using any provided filters. + for filter_func in self._filters: + file_paths = filter_func(hook, file_paths) + file_paths = list(file_paths) + + self.log.info("Filters after filtering: %s", file_paths) + + return bool(file_paths) + + +HdfsSensor = RenamedClass( + "HdfsSensor", + new_class=HdfsFileSensor, + old_module=__name__) + + +class HdfsFolderSensor(BaseSensorOperator): + """Waits for folders to lands in HDFS.""" + + template_fields = ("_pattern",) + ui_color = settings.WEB_COLORS["LIGHTBLUE"] - info = hook.get_conn().info(file_path) - min_size_mb = min_size * settings.MEGABYTE + def __init__( + self, + pattern, + conn_id="hdfs_default", + require_empty=False, + require_not_empty=False, + sub_pattern=None, + sub_filters=None, + **kwargs + ): + super(HdfsFolderSensor, self).__init__(**kwargs) - return info['kind'] == 'file' and info['size'] > min_size_mb + if require_empty and require_not_empty: + raise ValueError( + "Either require_empty or require_not_empty must be false, " + "as the two conditions are mutually exclusive." + ) - @staticmethod - def _filter_ext(_, file_path, exts): - """Filters any files with the given extensions.""" + self._pattern = pattern + self._conn_id = conn_id - file_ext = posixpath.splitext(file_path)[1][1:] - return file_ext not in exts + self._require_empty = require_empty + self._require_not_empty = require_not_empty + + self._sub_pattern = sub_pattern or "*" + self._sub_filters = sub_filters or [] def poke(self, context): - self.log.info('Poking for file pattern %s', self.file_pattern) - hdfs_conn = self._hook.get_conn() - - try: - file_paths = hdfs_conn.glob(self.file_pattern) - except IOError: - # File path doesn't exist yet. - return False - - self.log.info('Files matching pattern: %s', file_paths) - - file_paths = self._apply_filters( - file_paths, self._filters, hook=self._hook) - self.log.info('Filters after filtering: %s', file_paths) - - return self._not_empty(file_paths) - - @staticmethod - def _apply_filters(file_paths, filter_funcs, hook): - """Filters file paths that fail any of the filters (i.e. any - of the filter functions returns true for a given file). - """ - - for file_path in file_paths: - for func in filter_funcs: - if not func(hook, file_path): - break - else: - yield file_path - - @staticmethod - def _not_empty(iterable): - """Returns true if iterable is not empty.""" - - try: - next(iterable) - return True - except StopIteration: - return False + with HdfsHook(self._conn_id) as hook: + conn = hook.get_conn() + + # Try to expand glob pattern, returns single dir if not glob. + self.log.info("Poking for directories matching %s", self._pattern) + + try: + dir_paths = [ + path_ for path_ in conn.glob(self._pattern) if conn.isdir(path_) + ] + except IOError: + # File path doesn't exist yet. + dir_paths = [] + + self.log.info("Directories matching pattern: %s", dir_paths) + + if self._require_empty or self._require_not_empty: + self.log.info( + "Checking for files or subdirectories " "matching pattern: %s", + self._sub_pattern, + ) + + # Check if directories do/don't contain files. Returns False + # if any of the directories fail the required condition. + for dir_path in dir_paths: + sub_pattern = posixpath.join(dir_path, self._sub_pattern) + sub_paths = conn.glob(sub_pattern) + + for filter_func in self._sub_filters: + sub_paths = filter_func(hook, sub_paths) + + sub_paths = list(sub_paths) + is_empty = not sub_paths + + self.log.info( + "Sub-directories/files matching pattern: %s", sub_paths + ) + + if (self._require_empty and not is_empty) or ( + self._require_not_empty and is_empty + ): + return False + + return bool(dir_paths) + + +def filter_by_size(hook, file_paths, min_size): + """Filters any HDFS files below a minimum file size.""" + + conn = hook.get_conn() + min_size_mb = min_size * settings.MEGABYTE + + for file_path in file_paths: + info = conn.info(file_path) + + if info["kind"] == "file" and info["size"] > min_size_mb: + yield file_path + + +def filter_for_exts(_, file_paths, exts): + """Filters any HDFS files with the given extensions.""" + + for file_path in file_paths: + if posixpath.splitext(file_path)[1][1:] not in exts: + yield file_path diff --git a/airflow/utils/deprecation.py b/airflow/utils/deprecation.py new file mode 100644 index 0000000000000..8bce49cae5373 --- /dev/null +++ b/airflow/utils/deprecation.py @@ -0,0 +1,157 @@ +# -*- coding: utf-8 -*- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function +from __future__ import unicode_literals + +import functools +import warnings + + +def deprecated_args(renamed=None, dropped=None): + """Decorator for the deprecation of renamed/removed keyword arguments. + + Wraps functions with changed keyword arguments (either renamed to new + arguments or dropped entirely). Functions calls with deprecated arguments + raise appropriate warnings. For removed arguments, any given values are + ignored (outside of the warning). For renamed arguments, values are + transparently proxied to their new argument names. + + :param dict[str, str] renamed: Dict mapping old arguments to their + argument names in the new function/method. + :param list[str] dropped: List of arguments that have been removed + in the new function/method. + """ + + def decorator(function): + @functools.wraps(function) + def wrapper(*args, **kwargs): + new_kwargs = {} + + for key in kwargs: + if key in dropped: + warnings.warn( + "Argument {!r} is no longer supported and will be" + "removed in a future version of Airflow.".format(key), + category=DeprecationWarning, + ) + elif key in renamed: + warnings.warn( + "Argument {!r} has been renamed to {!r}. The old name " + "will no longer be supported in a future version of " + "Airflow.".format(key, renamed[key]), + category=DeprecationWarning, + ) + + new_kwargs[renamed[key]] = kwargs[key] + else: + new_kwargs[key] = kwargs[key] + + return function(*args, **new_kwargs) + + return wrapper + + return decorator + + +def deprecated(new_name=None): + """This is a decorator which can be used to mark functions as deprecated. + + The decorator ensures a warning is emitted whenever the function is + called to warn the user of its deprecated status. The parameter new_name + can be used to indicate a replacing function, if the deprecated function + has been renamed or replaced. + + :param str new_name: Optional name of a replacing function, if applicable. + """ + + def decorator(function): + @functools.wraps(function) + def wrapper(*args, **kwargs): + if new_name: + message = ( + "{} has been deprecated and will be replaced by " + "{} in a future version of Airflow.".format( + function.__name__, new_name + ) + ) + else: + message = ( + "{} has been deprecated and will be removed " + "in a future version of Airflow.".format(function.__name__) + ) + warnings.warn(message, category=DeprecationWarning) + return function(*args, **kwargs) + + return wrapper + + return decorator + + +class RenamedClass(object): + """Helper class used for deprecating old classes that have new names. + + For example, we can use this class to rename the (old) class + `HDFSHook` to it's new class name `HdfsHook` as follows: + + class HdfsHook(object): + ... + + HDFSHook = RenamedClass('HDFSHook', new_class=HdfsHook) + + so that old code can still use the deprecated form: + + hook = HDFSHook(...) + + which will raise an appropriate warning when called. + + :param str old_name: Name of the old class. + :param class new_class: The replacing class. + :param str old_module: Name of the module containing the old class. + """ + + def __init__(self, old_name, new_class, old_module=None): + self._old_name = old_name + self._old_module = old_module + self._new_class = new_class + + def _warn(self): + old_name = self._old_name + + if self._old_module: + old_name = self._old_module + '.' + old_name + + message = ("Class {!r} has been renamed to {!r}. Support for the old " + "class name will be removed in future versions of Airflow." + .format(old_name, _full_class_name(self._new_class))) + warnings.warn(message, category=DeprecationWarning) + + def __call__(self, *args, **kwargs): + self._warn() + return self._new_class(*args, **kwargs) + + def __getattr__(self, attr): + self._warn() + return getattr(self._new_class, attr) + + +def _full_class_name(cls): + return cls.__module__ + "." + cls.__name__ diff --git a/tests/contrib/sensors/test_hdfs_sensor.py b/tests/contrib/sensors/test_hdfs_sensor.py index 7036f7360a025..82bada8c044ce 100644 --- a/tests/contrib/sensors/test_hdfs_sensor.py +++ b/tests/contrib/sensors/test_hdfs_sensor.py @@ -17,23 +17,21 @@ # specific language governing permissions and limitations # under the License. -import logging -import mock -import unittest - +import datetime as dt import re -from datetime import timedelta +import unittest +import warnings from airflow import models -from airflow.contrib.sensors.hdfs_sensor import HdfsSensorFolder, HdfsSensorRegex -from airflow.exceptions import AirflowSensorTimeout -from airflow.hooks.hdfs_hook import HdfsHook +from airflow.contrib.sensors.hdfs_sensor import (HdfsSensorFolder, + HdfsSensorRegex, + HdfsRegexFileSensor) from tests.sensors.test_hdfs_sensor import MockHdfs3Client -class HdfsSensorRegexTests(unittest.TestCase): - """Tests for the HdfsSensorRegex class.""" +class HdfsRegexFileSensorTests(unittest.TestCase): + """Tests for the HdfsRegexFileSensor class.""" def setUp(self): file_details = [ @@ -59,7 +57,7 @@ def setUp(self): self._default_task_kws = { 'timeout': 1, - 'retry_delay': timedelta(seconds=1), + 'retry_delay': dt.timedelta(seconds=1), 'poke_interval': 1 } @@ -67,192 +65,141 @@ def test_should_match_regex(self): """Tests example where files should match regex.""" regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='should_match_the_regex', - file_pattern='/data/*', - regex=regex, - **self._default_task_kws) + task = HdfsRegexFileSensor( + task_id='should_match_the_regex', + pattern='/data/*', + regex=regex, + **self._default_task_kws) self.assertTrue(task.poke(context={})) def test_should_match_regex_dir(self): """Tests example where files should match regex with dir path.""" regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='should_match_the_regex', - file_pattern='/data', - regex=regex, - **self._default_task_kws) + task = HdfsRegexFileSensor( + task_id='should_match_the_regex', + pattern='/data', + regex=regex, + **self._default_task_kws) self.assertTrue(task.poke(context={})) def test_should_not_match_regex(self): """Tests example where files should match regex.""" regex = re.compile("^IDoNotExist") - task = HdfsSensorRegex(task_id='should_not_match_the_regex', - file_pattern='/data/*', - regex=regex, - **self._default_task_kws) + task = HdfsRegexFileSensor( + task_id='should_not_match_the_regex', + pattern='/data/*', + regex=regex, + **self._default_task_kws) self.assertFalse(task.poke(context={})) def test_should_match_regex_and_size(self): """Tests example with matching regex and sufficient file size.""" regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='should_match_the_regex_and_size', - file_pattern='/data/*', - regex=regex, - min_size=1, - **self._default_task_kws) + task = HdfsRegexFileSensor( + task_id='should_match_the_regex_and_size', + pattern='/data/*', + regex=regex, + min_size=1, + **self._default_task_kws) self.assertTrue(task.poke(context={})) def test_should_match_regex_not_size(self): """Tests example with matching regex but too small file size.""" regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='should_match_the_regex_but_not_size', - file_pattern='/data/*', - regex=regex, - min_size=10, - **self._default_task_kws) + task = HdfsRegexFileSensor( + task_id='should_match_the_regex_but_not_size', + pattern='/data/*', + regex=regex, + min_size=10, + **self._default_task_kws) self.assertFalse(task.poke(context={})) def test_should_match_regex_not_ext(self): """Tests example with matching regex but wrong ext.""" regex = re.compile("test[1-2]file") - task = HdfsSensorRegex(task_id='should_match_the_regex_but_not_size', - file_pattern='/data/*', - regex=regex, - min_size=10, - **self._default_task_kws) + task = HdfsRegexFileSensor( + task_id='should_match_the_regex_but_not_size', + pattern='/data/*', + regex=regex, + min_size=10, + **self._default_task_kws) self.assertFalse(task.poke(context={})) compiled_regex = re.compile("copying.*") - task = HdfsSensorRegex(task_id='should_match_the_regex_but_not_ext', - file_pattern='/data/*', - regex=compiled_regex, - ignore_exts=['_COPYING_'], - **self._default_task_kws) + task = HdfsRegexFileSensor( + task_id='should_match_the_regex_but_not_ext', + pattern='/data/*', + regex=compiled_regex, + ignore_exts=['_COPYING_'], + **self._default_task_kws) self.assertFalse(task.poke(context={})) + def test_calling_old_class(self): + """Tests call to old class.""" + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + + regex = re.compile("test[1-2]file") + task = HdfsSensorRegex( + task_id='should_match_the_regex_old', + pattern='/data', + regex=regex, + **self._default_task_kws) + self.assertTrue(task.poke(context={})) + class HdfsSensorFolderTests(unittest.TestCase): + """Tests for the (deprecated) HdfsSensorFolder class.""" def setUp(self): file_details = [ - { - 'kind': 'directory', - 'name': '/empty', - 'size': 0 - }, - { - 'kind': 'directory', - 'name': '/not_empty', - 'size': 0 - }, - { - 'kind': 'file', - 'name': '/not_empty/test1file', - 'size': 2000000 - } + {"kind": "directory", "name": "/empty", "size": 0}, + {"kind": "directory", "name": "/not_empty", "size": 0}, + {"kind": "file", "name": "/not_empty/a.tmp", "size": 2000000}, + {"kind": "directory", "name": "/nested", "size": 0}, + {"kind": "directory", "name": "/nested/a", "size": 0} ] - self._mock_client, self._mock_params = \ - MockHdfs3Client.from_file_details(file_details, test_instance=self) + self._mock_client, self._mock_params = MockHdfs3Client.from_file_details( + file_details, test_instance=self + ) self._default_task_kws = { - 'timeout': 1, - 'retry_delay': timedelta(seconds=1), - 'poke_interval': 1 + "timeout": 1, + "retry_delay": dt.timedelta(seconds=1), + "poke_interval": 1, } self._mock_client = MockHdfs3Client(file_details) - self._mock_params = models.Connection(conn_id='hdfs_default') - - def test_should_be_empty_directory(self): - """Tests example with empty directory.""" - - # Given - self.log.debug('#' * 10) - self.log.debug('Running %s', self._testMethodName) - self.log.debug('#' * 10) - task = HdfsSensorFolder(task_id='Should_be_empty_directory', - filepath='/datadirectory/empty_directory', - be_empty=True, - timeout=1, - retry_delay=timedelta(seconds=1), - poke_interval=1, - hook=self.hook) - - # When - task.execute(None) - - # Then - # Nothing happens, nothing is raised exec is ok - -# def test_should_be_empty_directory_fail(self): -# """ -# test the empty directory behaviour -# :return: -# """ -# # Given -# self.log.debug('#' * 10) -# self.log.debug('Running %s', self._testMethodName) -# self.log.debug('#' * 10) -# task = HdfsSensorFolder(task_id='Should_be_empty_directory_fail', -# filepath='/datadirectory/not_empty_directory', -# be_empty=True, -# timeout=1, -# retry_delay=timedelta(seconds=1), -# poke_interval=1, -# hook=self.hook) - -# # When -# # Then -# with self.assertRaises(AirflowSensorTimeout): -# task.execute(None) - -# def test_should_be_a_non_empty_directory(self): -# """ -# test the empty directory behaviour -# :return: -# """ -# # Given -# self.log.debug('#' * 10) -# self.log.debug('Running %s', self._testMethodName) -# self.log.debug('#' * 10) -# task = HdfsSensorFolder(task_id='Should_be_non_empty_directory', -# filepath='/datadirectory/not_empty_directory', -# timeout=1, -# retry_delay=timedelta(seconds=1), -# poke_interval=1, -# hook=self.hook) - -# # When -# task.execute(None) - -# # Then -# # Nothing happens, nothing is raised exec is ok - -# def test_should_be_non_empty_directory_fail(self): -# """ -# test the empty directory behaviour -# :return: -# """ -# # Given -# self.log.debug('#' * 10) -# self.log.debug('Running %s', self._testMethodName) -# self.log.debug('#' * 10) -# task = HdfsSensorFolder(task_id='Should_be_empty_directory_fail', -# filepath='/datadirectory/empty_directory', -# timeout=1, -# retry_delay=timedelta(seconds=1), -# poke_interval=1, -# hook=self.hook) - -# # When -# # Then -# with self.assertRaises(AirflowSensorTimeout): -# task.execute(None) + self._mock_params = models.Connection(conn_id="hdfs_default") + + def test_empty_directory(self): + """Tests example with an empty directory.""" + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + + task = HdfsSensorFolder( + task_id="test_empty_directory", + pattern="/empty") + self.assertTrue(task.poke(context={})) + + def test_non_empty_directory(self): + """Tests example with a non-empty directory.""" + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + + task = HdfsSensorFolder( + task_id="test_non_empty_directory", + pattern="/empty") + self.assertTrue(task.poke(context={})) if __name__ == '__main__': diff --git a/tests/sensors/test_hdfs_sensor.py b/tests/sensors/test_hdfs_sensor.py index f28c232f9c45b..822d8c37b23e7 100644 --- a/tests/sensors/test_hdfs_sensor.py +++ b/tests/sensors/test_hdfs_sensor.py @@ -19,25 +19,25 @@ from datetime import timedelta import fnmatch +import posixpath import unittest +import warnings import mock from airflow import models -from airflow.sensors.hdfs_sensor import HdfsSensor, HdfsHook +from airflow.sensors.hdfs_sensor import (HdfsSensor, HdfsFileSensor, + HdfsFolderSensor, HdfsHook) class MockHdfs3Client(object): """Mock hdfs3 client for testing purposes.""" def __init__(self, file_details): - self._file_details = { - entry['name']: entry for entry in file_details - } + self._file_details = {entry["name"]: entry for entry in file_details} @classmethod - def from_file_details(cls, file_details, test_instance, - conn_id='hdfs_default'): + def from_file_details(cls, file_details, test_instance, conn_id="hdfs_default"): """Builds mock hdsf3 client using file_details.""" mock_client = cls(file_details) @@ -45,13 +45,13 @@ def from_file_details(cls, file_details, test_instance, # Setup mock for get_connection. patcher = mock.patch.object( - HdfsHook, 'get_connection', return_value=mock_params) + HdfsHook, "get_connection", return_value=mock_params + ) test_instance.addCleanup(patcher.stop) patcher.start() # Setup mock for get_conn. - patcher = mock.patch.object( - HdfsHook, 'get_conn', return_value=mock_client) + patcher = mock.patch.object(HdfsHook, "get_conn", return_value=mock_client) test_instance.addCleanup(patcher.stop) patcher.start() @@ -59,126 +59,347 @@ def from_file_details(cls, file_details, test_instance, def glob(self, pattern): """Returns glob of files matching pattern.""" - return fnmatch.filter(self._file_details.keys(), pattern) - def info(self, file_path): + # Implements a non-recursive glob on file names. + pattern_dir = posixpath.dirname(pattern) + pattern_base = posixpath.basename(pattern) + + # Ensure pattern_dir ends with '/'. + pattern_dir = posixpath.join(pattern_dir, "") + + for file_path in self._file_details.keys(): + file_basename = posixpath.basename(file_path) + if file_path.startswith(pattern_dir) and \ + fnmatch.fnmatch(file_basename, pattern_base): + yield file_path + + def isdir(self, path): + """Returns true if path is a directory.""" + + try: + details = self._file_details[path] + except KeyError: + raise IOError() + + return details["kind"] == "directory" + + def info(self, path): """Returns info for given file path.""" try: - return self._file_details[file_path] + return self._file_details[path] except KeyError: raise IOError() -class HdfsSensorTests(unittest.TestCase): - """Tests for the HdfsSensor class.""" +class HdfsFileSensorTests(unittest.TestCase): + """Tests for the HdfsFileSensor class.""" def setUp(self): file_details = [ - { - 'kind': 'directory', - 'name': '/data/empty', - 'size': 0 - }, - { - 'kind': 'directory', - 'name': '/data/not_empty', - 'size': 0 - }, - { - 'kind': 'file', - 'name': '/data/not_empty/small.txt', - 'size': 10 - }, - { - 'kind': 'file', - 'name': '/data/not_empty/large.txt', - 'size': 10000000 - }, - { - 'kind': 'file', - 'name': '/data/not_empty/file.txt._COPYING_' - } + {"kind": "directory", "name": "/data/empty", "size": 0}, + {"kind": "directory", "name": "/data/not_empty", "size": 0}, + {"kind": "file", "name": "/data/not_empty/small.txt", "size": 10}, + {"kind": "file", "name": "/data/not_empty/large.txt", "size": 10000000}, + {"kind": "file", "name": "/data/not_empty/file.txt._COPYING_"}, ] - self._mock_client, self._mock_params = \ - MockHdfs3Client.from_file_details(file_details, test_instance=self) + self._mock_client, self._mock_params = MockHdfs3Client.from_file_details( + file_details, test_instance=self + ) self._default_task_kws = { - 'timeout': 1, - 'retry_delay': timedelta(seconds=1), - 'poke_interval': 1 + "timeout": 1, + "retry_delay": timedelta(seconds=1), + "poke_interval": 1, } def test_existing_file(self): """Tests poking for existing file.""" - task = HdfsSensor(task_id='existing_file', - file_path='/data/not_empty/small.txt', - **self._default_task_kws) + task = HdfsFileSensor( + task_id="existing_file", + pattern="/data/not_empty/small.txt", + **self._default_task_kws + ) self.assertTrue(task.poke(context={})) def test_existing_file_glob(self): """Tests poking for existing file with glob.""" - task = HdfsSensor(task_id='existing_file', - file_path='/data/not_empty/*.txt', - **self._default_task_kws) + task = HdfsFileSensor( + task_id="existing_file", + pattern="/data/not_empty/*.txt", + **self._default_task_kws + ) self.assertTrue(task.poke(context={})) def test_nonexisting_file(self): """Tests poking for non-existing file.""" - task = HdfsSensor(task_id='nonexisting_file', - file_path='/data/not_empty/random.txt', - **self._default_task_kws) + task = HdfsFileSensor( + task_id="nonexisting_file", + pattern="/data/not_empty/random.txt", + **self._default_task_kws + ) self.assertFalse(task.poke(context={})) def test_nonexisting_file_glob(self): """Tests poking for non-existing file with glob.""" - task = HdfsSensor(task_id='existing_file', - file_path='/data/not_empty/*.xml', - **self._default_task_kws) + task = HdfsFileSensor( + task_id="existing_file", + pattern="/data/not_empty/*.xml", + **self._default_task_kws + ) self.assertFalse(task.poke(context={})) def test_nonexisting_path(self): """Tests poking for non-existing path.""" - task = HdfsSensor(task_id='nonexisting_path', - file_path='/data/not_empty/small.txt', - **self._default_task_kws) + task = HdfsFileSensor( + task_id="nonexisting_path", + pattern="/data/not_empty/small.txt", + **self._default_task_kws + ) - with mock.patch.object(self._mock_client, 'glob', side_effect=IOError): + with mock.patch.object(self._mock_client, "glob", side_effect=IOError): self.assertFalse(task.poke(context={})) def test_file_filter_size_small(self): """Tests poking for file while filtering for file size (too small).""" - task = HdfsSensor(task_id='existing_file_too_small', - file_path='/data/not_empty/small.txt', - min_size=1, - **self._default_task_kws) + task = HdfsFileSensor( + task_id="existing_file_too_small", + pattern="/data/not_empty/small.txt", + min_size=1, + **self._default_task_kws + ) self.assertFalse(task.poke(context={})) def test_file_filter_size_large(self): """Tests poking for file while filtering for file size (large).""" - task = HdfsSensor(task_id='existing_file_large', - file_path='/data/not_empty/large.txt', - min_size=1, - **self._default_task_kws) + task = HdfsFileSensor( + task_id="existing_file_large", + pattern="/data/not_empty/large.txt", + min_size=1, + **self._default_task_kws + ) self.assertTrue(task.poke(context={})) def test_file_filter_ext(self): """Tests poking for file while filtering for extension.""" - task = HdfsSensor(task_id='existing_file_large', - file_path='/data/not_empty/f*', - ignored_ext=('_COPYING_', ), - **self._default_task_kws) + task = HdfsFileSensor( + task_id="existing_file_large", + pattern="/data/not_empty/f*", + ignore_exts=("_COPYING_",), + **self._default_task_kws + ) + self.assertFalse(task.poke(context={})) + + def test_file_filter_ext_old(self): + """Tests poking for file while filtering for extension + with deprecated ignored_ext argument.""" + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + + task = HdfsFileSensor( + task_id="existing_file_large", + pattern="/data/not_empty/f*", + ignored_ext=("_COPYING_",), + **self._default_task_kws + ) + self.assertFalse(task.poke(context={})) + + def test_old_class(self): + """Tests sensor with old class name.""" + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + + task = HdfsSensor( + task_id="existing_file", + pattern="/data/not_empty/small.txt", + **self._default_task_kws + ) + + self.assertTrue(task.poke(context={})) + + +class HdfsFolderSensorTests(unittest.TestCase): + def setUp(self): + file_details = [ + {"kind": "directory", "name": "/empty", "size": 0}, + {"kind": "directory", "name": "/not_empty", "size": 0}, + {"kind": "file", "name": "/not_empty/a.tmp", "size": 2000000}, + {"kind": "directory", "name": "/nested", "size": 0}, + {"kind": "directory", "name": "/nested/a", "size": 0} + ] + + self._mock_client, self._mock_params = MockHdfs3Client.from_file_details( + file_details, test_instance=self + ) + + self._default_task_kws = { + "timeout": 1, + "retry_delay": timedelta(seconds=1), + "poke_interval": 1, + } + + self._mock_client = MockHdfs3Client(file_details) + self._mock_params = models.Connection(conn_id="hdfs_default") + + def test_empty_directory(self): + """Tests example with an empty directory.""" + + task = HdfsFolderSensor(task_id="test_empty_directory", pattern="/empty") + self.assertTrue(task.poke(context={})) + + def test_non_empty_directory(self): + """Tests example with a non-empty directory.""" + + task = HdfsFolderSensor(task_id="test_non_empty_directory", pattern="/empty") + self.assertTrue(task.poke(context={})) + + def test_required_empty_directory(self): + """Tests example with an empty directory and require_empty = True.""" + + task = HdfsFolderSensor( + task_id="test_empty_directory", pattern="/empty", require_empty=True + ) + self.assertTrue(task.poke(context={})) + + def test_required_empty_directory_fail(self): + """Tests example with a non-empty directory and require_empty = True.""" + + task = HdfsFolderSensor( + task_id="test_non_empty_directory", pattern="/not_empty", require_empty=True + ) + self.assertFalse(task.poke(context={})) + + def test_required_not_empty_directory(self): + """Tests example with a non-empty directory and + require_not_empty = True. + """ + + task = HdfsFolderSensor( + task_id="test_required_not_empty_directory", + pattern="/not_empty", + require_not_empty=True, + ) + self.assertTrue(task.poke(context={})) + + def test_required_not_empty_directory_fail(self): + """Tests example with an empty directory and + require_not_empty = True. + """ + + task = HdfsFolderSensor( + task_id="test_required_not_empty_directory", + pattern="/empty", + require_not_empty=True, + ) + self.assertFalse(task.poke(context={})) + + def test_glob(self): + """Tests globbing for directories.""" + + task = HdfsFolderSensor(task_id="test_glob", pattern="/*") + self.assertTrue(task.poke(context={})) + + def test_glob_non_existing(self): + """Tests globbing for non-existing directories.""" + + task = HdfsFolderSensor( + task_id="test_glob_non_existing", pattern="/non-existing/*" + ) + self.assertFalse(task.poke(context={})) + + def test_glob_require_empty(self): + """Tests globbing with only empty dir and require_empty = True.""" + + task = HdfsFolderSensor( + task_id="test_glob_require_empty", pattern="/e*", require_empty=True + ) + self.assertTrue(task.poke(context={})) + + def test_glob_require_empty_fail(self): + """Tests globbing with non-empty dir and require_empty = True.""" + + task = HdfsFolderSensor( + task_id="test_glob_require_empty_fail", pattern="/*", require_empty=True + ) + self.assertFalse(task.poke(context={})) + + def test_glob_require_not_empty(self): + """Tests globbing with only non-empty dir and + require_not_empty = True. + """ + + task = HdfsFolderSensor( + task_id="test_glob_require_not_empty", pattern="/n*", require_not_empty=True + ) + self.assertTrue(task.poke(context={})) + + def test_glob_require_not_empty_fail(self): + """Tests globbing with empty dir and require_not_empty = True.""" + + task = HdfsFolderSensor( + task_id="test_glob_require_not_empty_fail", + pattern="/*", + require_not_empty=True, + ) + self.assertFalse(task.poke(context={})) + + def test_sub_pattern(self): + """Tests filtering directory with matching sub_pattern.""" + + task = HdfsFolderSensor( + task_id="test_sub_pattern", + pattern="/not_empty", + require_not_empty=True, + sub_pattern="*.tmp" + ) + self.assertTrue(task.poke(context={})) + + def test_sub_pattern_no_match(self): + """Tests filtering directory with sub_pattern that doesn't match.""" + + task = HdfsFolderSensor( + task_id="test_sub_pattern_no_match", + pattern="/not_empty", + require_not_empty=True, + sub_pattern="*.txt" + ) + self.assertFalse(task.poke(context={})) + + def test_sub_pattern_subdir(self): + """Tests filtering directory with sub_pattern that matches subdir.""" + + task = HdfsFolderSensor( + task_id="test_sub_pattern_subdir", + pattern="/nested", + require_not_empty=True, + sub_pattern="a" + ) + self.assertTrue(task.poke(context={})) + + def test_sub_pattern_subdir_no_match(self): + """Tests filtering directory with sub_pattern that doesn't + match subdir. + """ + + task = HdfsFolderSensor( + task_id="test_sub_pattern_subdir_no_match", + pattern="/nested", + require_not_empty=True, + sub_pattern="b" + ) self.assertFalse(task.poke(context={})) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main() From c09b1c95f45a6e3ea4412dc413a3c2491e4e30a8 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Mon, 23 Jul 2018 16:40:49 +0200 Subject: [PATCH 09/15] Apply kerberos even without connection. --- airflow/hooks/hdfs_hook.py | 43 ++++++++-------- airflow/sensors/hdfs_sensor.py | 3 +- tests/core.py | 3 ++ tests/hooks/test_hdfs_hook.py | 94 ++++++++++++++++++++++++++++++++-- 4 files changed, 117 insertions(+), 26 deletions(-) diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index c3d7e19e64865..267890e4748de 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -17,8 +17,6 @@ # specific language governing permissions and limitations # under the License. -import warnings - import hdfs3 from hdfs3.utils import MyNone @@ -43,12 +41,12 @@ class HdfsHook(BaseHook): "ha": { "host": "nameservice1", "conf": { - "dfs.nameservices": "nameservice1", - "dfs.ha.namenodes.nameservice1": "namenode113,namenode188", - "dfs.namenode.rpc-address.nameservice1.namenode113": "host1:8020", - "dfs.namenode.rpc-address.nameservice1.namenode188": "host2:8020", - "dfs.namenode.http-address.nameservice1.namenode113": "host1:50070", - "dfs.namenode.http-address.nameservice1.namenode188": "host2:50070" + "dfs.nameservices": "ns1", + "dfs.ha.namenodes.ns1": "nn1,nn2", + "dfs.namenode.rpc-address.ns1.nn1": "host1:8020", + "dfs.namenode.rpc-address.ns1.nn2": "host2:8020", + "dfs.namenode.http-address.ns1.nn1": "host1:50070", + "dfs.namenode.http-address.ns1.nn2": "host2:50070" } } } @@ -78,32 +76,35 @@ def __init__(self, hdfs_conn_id=None, autoconf=True): def get_conn(self): if self._conn is None: + hdfs_params = {} + + # Configure kerberos security if used by Airflow. + if configuration.conf.get("core", "security") == "kerberos": + hdfs_params["hadoop.security.authentication"] = "kerberos" + if self.hdfs_conn_id is None: - self._conn = hdfs3.HDFileSystem(autoconf=self._autoconf) + self._conn = hdfs3.HDFileSystem( + autoconf=self._autoconf, pars=hdfs_params) else: - params = self.get_connection(self.hdfs_conn_id) - extra_params = params.extra_dejson + conn_params = self.get_connection(self.hdfs_conn_id) + conn_extra_params = conn_params.extra_dejson # Extract hadoop parameters from extra. - hdfs_params = extra_params.get("pars", {}) - - # Configure kerberos security if used by Airflow. - if configuration.conf.get("core", "security") == "kerberos": - hdfs_params["hadoop.security.authentication"] = "kerberos" + hdfs_params.update(conn_extra_params.get("pars", {})) # Extract high-availability config if given. - ha_params = extra_params.get("ha", {}) + ha_params = conn_extra_params.get("ha", {}) hdfs_params.update(ha_params.get("conf", {})) # Collect extra parameters to pass to kwargs. extra_kws = {} - if params.login: - extra_kws["user"] = params.login + if conn_params.login: + extra_kws["user"] = conn_params.login # Build connection. self._conn = hdfs3.HDFileSystem( - host=ha_params.get("host") or params.host or MyNone, - port=params.port or MyNone, + host=ha_params.get("host") or conn_params.host or MyNone, + port=conn_params.port or MyNone, pars=hdfs_params, autoconf=self._autoconf, **extra_kws) diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index 7e54479e8ad23..491a2e9908a89 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -121,7 +121,8 @@ def poke(self, context): try: file_paths = [ - fp for fp in conn.glob(self._pattern) if not conn.isdir(fp) + fp for fp in conn.glob(self._pattern) + if not conn.isdir(fp) ] except IOError: # File path doesn't exist yet. diff --git a/tests/core.py b/tests/core.py index f7e3edb2cefea..62a5817001670 100644 --- a/tests/core.py +++ b/tests/core.py @@ -2749,6 +2749,7 @@ def test_init_proxy_user(self): self.assertEqual('someone', c.proxy_user) +<<<<<<< HEAD HDFSHook = None if six.PY2: from airflow.hooks.hdfs_hook import HDFSHook @@ -2798,6 +2799,8 @@ def test_get_ha_client(self, mock_get_connections): self.assertIsInstance(client, snakebite.client.HAClient) +======= +>>>>>>> Apply kerberos even without connection. send_email_test = mock.Mock() diff --git a/tests/hooks/test_hdfs_hook.py b/tests/hooks/test_hdfs_hook.py index e0371c85b4133..4ca70b498d28b 100644 --- a/tests/hooks/test_hdfs_hook.py +++ b/tests/hooks/test_hdfs_hook.py @@ -22,7 +22,7 @@ import mock from hdfs3.utils import MyNone -from airflow.hooks.hdfs_hook import HdfsHook, hdfs3 +from airflow.hooks.hdfs_hook import HdfsHook, hdfs3, configuration class TestHDFSHook(unittest.TestCase): @@ -48,7 +48,7 @@ def test_get_conn(self, conn_mock, hdfs3_mock): hook.get_conn() conn_mock.assert_not_called() - hdfs3_mock.assert_called_once_with(autoconf=True) + hdfs3_mock.assert_called_once_with(autoconf=True, pars={}) @mock.patch.object(hdfs3, 'HDFileSystem') @mock.patch.object(HdfsHook, 'get_connection') @@ -59,12 +59,12 @@ def test_get_conn_no_autoconf(self, conn_mock, hdfs3_mock): hook.get_conn() conn_mock.assert_not_called() - hdfs3_mock.assert_called_once_with(autoconf=False) + hdfs3_mock.assert_called_once_with(autoconf=False, pars={}) @mock.patch.object(hdfs3, 'HDFileSystem') @mock.patch.object(HdfsHook, 'get_connection') def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): - """Tests get_conn call with ID.""" + """Tests get_conn call with specified connection.""" conn_mock.return_value = mock.Mock( host='namenode', @@ -84,6 +84,92 @@ def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): user='hdfs_user', autoconf=True) + @mock.patch.object(hdfs3, 'HDFileSystem') + @mock.patch.object(HdfsHook, 'get_connection') + def test_get_conn_with_conn_ha(self, conn_mock, hdfs3_mock): + """Tests get_conn call with connection containing HA config.""" + + conn_mock.return_value = mock.Mock( + host='namenode', + login='hdfs_user', + port=8020, + extra_dejson={ + 'pars': { + 'dfs.namenode.logging.level': 'info' + }, + "ha": { + "host": "ns1", + "conf": { + "dfs.nameservices": "ns1", + "dfs.ha.namenodes.ns1": "nn1,nn2", + "dfs.namenode.rpc-address.ns1.nn1": "host1:8020", + "dfs.namenode.rpc-address.ns1.nn2": "host2:8020", + "dfs.namenode.http-address.ns1.nn1": "host1:50070", + "dfs.namenode.http-address.ns1.nn2": "host2:50070" + } + } + }) + + with HdfsHook(hdfs_conn_id='hdfs_default') as hook: + hook.get_conn() + + conn_mock.assert_called_once_with('hdfs_default') + + hdfs3_mock.assert_called_once_with( + host='ns1', + port=8020, + pars={ + 'dfs.namenode.logging.level': 'info', + 'dfs.nameservices': 'ns1', + 'dfs.ha.namenodes.ns1': 'nn1,nn2', + 'dfs.namenode.rpc-address.ns1.nn1': 'host1:8020', + 'dfs.namenode.rpc-address.ns1.nn2': 'host2:8020', + 'dfs.namenode.http-address.ns1.nn1': 'host1:50070', + 'dfs.namenode.http-address.ns1.nn2': 'host2:50070' + }, + user='hdfs_user', + autoconf=True) + + @mock.patch.object(configuration.conf, 'get') + @mock.patch.object(hdfs3, 'HDFileSystem') + @mock.patch.object(HdfsHook, 'get_connection') + def test_kerberos(self, conn_mock, hdfs3_mock, conf_mock): + """Tests setting kerberos auth from Airflow config.""" + + conn_mock.return_value = mock.Mock( + host='namenode', + login='hdfs_user', + port=8020, + extra_dejson={}) + + conf_mock.return_value = 'kerberos' + + with HdfsHook(hdfs_conn_id='hdfs_default') as hook: + hook.get_conn() + + hdfs3_mock.assert_called_once_with( + host='namenode', + port=8020, + user='hdfs_user', + pars={'hadoop.security.authentication': 'kerberos'}, + autoconf=True + ) + + @mock.patch.object(configuration.conf, 'get') + @mock.patch.object(hdfs3, 'HDFileSystem') + def test_kerberos_wo_conn(self, hdfs3_mock, conf_mock): + """Tests setting kerberos auth from Airflow config without conn.""" + + conf_mock.return_value = 'kerberos' + + with HdfsHook() as hook: + hook.get_conn() + + hdfs3_mock.assert_called_once_with( + pars={'hadoop.security.authentication': 'kerberos'}, + autoconf=True + ) + @mock.patch.object(hdfs3, 'HDFileSystem') @mock.patch.object(HdfsHook, 'get_connection') def test_get_conn_with_empty_conn(self, conn_mock, hdfs3_mock): From ce9f0915bc3a3f2b1eafa01346e1b88a6c4afcd8 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Tue, 24 Jul 2018 09:06:02 +0200 Subject: [PATCH 10/15] Fix Python 2 specific errors. --- airflow/hooks/hdfs_hook.py | 2 +- airflow/sensors/hdfs_sensor.py | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index 267890e4748de..9d1bfec6fcca8 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -67,7 +67,7 @@ class HdfsHook(BaseHook): """ def __init__(self, hdfs_conn_id=None, autoconf=True): - super().__init__(None) + super(HdfsHook, self).__init__(None) self.hdfs_conn_id = hdfs_conn_id self._autoconf = autoconf diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index 491a2e9908a89..30defb56a2566 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -80,23 +80,23 @@ def conn_id(self): # Deprecated properties that exist for backwards compatibility. - @deprecated(new_name="file_pattern") @property + @deprecated(new_name="file_pattern") def filepath(self): return self._pattern - @deprecated(new_name="conn_id") @property + @deprecated(new_name="conn_id") def hdfs_conn_id(self): return self._conn_id - @deprecated() @property + @deprecated() def min_size(self): return self._min_size - @deprecated() @property + @deprecated() def ignored_ext(self): return self._ignore_exts From a2e34012b61b198074c9e15dcddb6bbddf36b5b9 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Tue, 24 Jul 2018 11:33:31 +0200 Subject: [PATCH 11/15] Make extra_kws optional. --- airflow/hooks/hdfs_hook.py | 37 ++++++++++++++++++++++------------- tests/hooks/test_hdfs_hook.py | 25 ++++++++++------------- 2 files changed, 33 insertions(+), 29 deletions(-) diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index 9d1bfec6fcca8..e46aac8908af0 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -48,14 +48,20 @@ class HdfsHook(BaseHook): "dfs.namenode.http-address.ns1.nn1": "host1:50070", "dfs.namenode.http-address.ns1.nn2": "host2:50070" } - } + }, + "autoconf": True, + "token": "...", + "ticket_cache": "..." } Here `pars` can be used to supply configuration options with the same key names as typically contained in the XML config files, which will take precedence over any parameters loaded from files. The `ha` configuration section can be used to supply options for using hdfs3 in high-availability - mode. See the hdfs3 documentation for more details. + mode (see the hdfs3 documentation for more details). The `autoconf` option + controls whether hdfs3 uses the available hadoop config files for + its configuration, whilst the `token` and `ticket_cache` options are + used for configuring kerberos. Security modes can also be configured by defining appropriate value for the `hadoop.security.authentication` key in `pars`. Kerberos is used @@ -66,12 +72,10 @@ class HdfsHook(BaseHook): configuration options from the hdfs XML configuration files. """ - def __init__(self, hdfs_conn_id=None, autoconf=True): + def __init__(self, hdfs_conn_id=None): super(HdfsHook, self).__init__(None) self.hdfs_conn_id = hdfs_conn_id - self._autoconf = autoconf - self._conn = None def get_conn(self): @@ -83,8 +87,7 @@ def get_conn(self): hdfs_params["hadoop.security.authentication"] = "kerberos" if self.hdfs_conn_id is None: - self._conn = hdfs3.HDFileSystem( - autoconf=self._autoconf, pars=hdfs_params) + self._conn = hdfs3.HDFileSystem(pars=hdfs_params, autoconf=True) else: conn_params = self.get_connection(self.hdfs_conn_id) conn_extra_params = conn_params.extra_dejson @@ -96,18 +99,24 @@ def get_conn(self): ha_params = conn_extra_params.get("ha", {}) hdfs_params.update(ha_params.get("conf", {})) - # Collect extra parameters to pass to kwargs. - extra_kws = {} - if conn_params.login: - extra_kws["user"] = conn_params.login + # Collect extra parameters to pass to kwargs. Note that we + # avoid passing empty parameters, as hdfs3 seems to do + # funky things with defining its own None variable. + extra_kws = { + "user": conn_params.login or None, + "ticket_cache": conn_extra_params.get("ticket_cache"), + "token": conn_extra_params.get("token"), + } + extra_kws = {k: v for k, v in extra_kws.items() if v} # Build connection. self._conn = hdfs3.HDFileSystem( host=ha_params.get("host") or conn_params.host or MyNone, port=conn_params.port or MyNone, pars=hdfs_params, - autoconf=self._autoconf, - **extra_kws) + autoconf=conn_extra_params.get("autoconf", True), + **extra_kws + ) return self._conn @@ -125,4 +134,4 @@ def close(self): self._conn = None -HDFSHook = RenamedClass('HDFSHook', new_class=HdfsHook, old_module=__name__) +HDFSHook = RenamedClass("HDFSHook", new_class=HdfsHook, old_module=__name__) diff --git a/tests/hooks/test_hdfs_hook.py b/tests/hooks/test_hdfs_hook.py index 4ca70b498d28b..b7e9a22a7c025 100644 --- a/tests/hooks/test_hdfs_hook.py +++ b/tests/hooks/test_hdfs_hook.py @@ -50,17 +50,6 @@ def test_get_conn(self, conn_mock, hdfs3_mock): conn_mock.assert_not_called() hdfs3_mock.assert_called_once_with(autoconf=True, pars={}) - @mock.patch.object(hdfs3, 'HDFileSystem') - @mock.patch.object(HdfsHook, 'get_connection') - def test_get_conn_no_autoconf(self, conn_mock, hdfs3_mock): - """Tests get_conn call without ID and autoconf = False.""" - - with HdfsHook(autoconf=False) as hook: - hook.get_conn() - - conn_mock.assert_not_called() - hdfs3_mock.assert_called_once_with(autoconf=False, pars={}) - @mock.patch.object(hdfs3, 'HDFileSystem') @mock.patch.object(HdfsHook, 'get_connection') def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): @@ -70,7 +59,13 @@ def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): host='namenode', login='hdfs_user', port=8020, - extra_dejson={'pars': {'dfs.namenode.logging.level': 'info'}}) + extra_dejson={ + 'pars': { + 'dfs.namenode.logging.level': 'info' + }, + 'autoconf': False, + 'ticket_cache': '/path/to/cache' + }) with HdfsHook(hdfs_conn_id='hdfs_default') as hook: hook.get_conn() @@ -82,7 +77,8 @@ def test_get_conn_with_conn(self, conn_mock, hdfs3_mock): port=8020, pars={'dfs.namenode.logging.level': 'info'}, user='hdfs_user', - autoconf=True) + autoconf=False, + ticket_cache='/path/to/cache') @mock.patch.object(hdfs3, 'HDFileSystem') @mock.patch.object(HdfsHook, 'get_connection') @@ -152,8 +148,7 @@ def test_kerberos(self, conn_mock, hdfs3_mock, conf_mock): port=8020, user='hdfs_user', pars={'hadoop.security.authentication': 'kerberos'}, - autoconf=True - ) + autoconf=True) @mock.patch.object(configuration.conf, 'get') @mock.patch.object(hdfs3, 'HDFileSystem') From a21c3814994f0ee6910fa7a58bc461f94a479936 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Fri, 27 Jul 2018 14:10:09 +0200 Subject: [PATCH 12/15] Add args for backwards compatibility. --- airflow/contrib/sensors/hdfs_sensor.py | 3 +- airflow/sensors/hdfs_sensor.py | 44 ++++++----------------- tests/contrib/sensors/test_hdfs_sensor.py | 5 ++- 3 files changed, 14 insertions(+), 38 deletions(-) diff --git a/airflow/contrib/sensors/hdfs_sensor.py b/airflow/contrib/sensors/hdfs_sensor.py index 015a402b8429b..edd6e9cb6b2f9 100644 --- a/airflow/contrib/sensors/hdfs_sensor.py +++ b/airflow/contrib/sensors/hdfs_sensor.py @@ -21,12 +21,13 @@ import posixpath from airflow.sensors import hdfs_sensor -from airflow.utils.deprecation import RenamedClass +from airflow.utils.deprecation import RenamedClass, deprecated_args class HdfsRegexFileSensor(hdfs_sensor.HdfsFileSensor): """HdfsSensor subclass that filters using a specific regex.""" + @deprecated_args(renamed={"filepath": "pattern"}) def __init__(self, pattern, regex, **kwargs): if not self._is_pattern(pattern): # If file path is not a pattern, we assume it is a directory diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index 30defb56a2566..9947476de1bbb 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -24,7 +24,7 @@ from airflow.hooks.hdfs_hook import HdfsHook from airflow.sensors.base_sensor_operator import BaseSensorOperator from airflow.utils.decorators import apply_defaults -from airflow.utils.deprecation import deprecated_args, deprecated, RenamedClass +from airflow.utils.deprecation import deprecated_args, RenamedClass class HdfsFileSensor(BaseSensorOperator): @@ -35,7 +35,7 @@ class HdfsFileSensor(BaseSensorOperator): @deprecated_args( renamed={ - "filepath": "file_pattern", + "filepath": "pattern", "hdfs_conn_id": "conn_id", "file_size": "min_size", "ignored_ext": "ignore_exts", @@ -68,38 +68,6 @@ def __init__( self._min_size = min_size self._ignore_exts = set(ignore_exts) - @property - def pattern(self): - """File pattern (glob) that the sensor matches against.""" - return self._pattern - - @property - def conn_id(self): - """ID of connection used by the sensor.""" - return self._conn_id - - # Deprecated properties that exist for backwards compatibility. - - @property - @deprecated(new_name="file_pattern") - def filepath(self): - return self._pattern - - @property - @deprecated(new_name="conn_id") - def hdfs_conn_id(self): - return self._conn_id - - @property - @deprecated() - def min_size(self): - return self._min_size - - @property - @deprecated() - def ignored_ext(self): - return self._ignore_exts - @classmethod def _default_filters(cls, min_size=None, ignore_exts=None): filters = [] @@ -152,6 +120,14 @@ class HdfsFolderSensor(BaseSensorOperator): template_fields = ("_pattern",) ui_color = settings.WEB_COLORS["LIGHTBLUE"] + @deprecated_args( + renamed={ + "filepath": "pattern", + "hdfs_conn_id": "conn_id", + "be_empty": "require_empty" + }, + dropped=["ignored_ext", "min_size"] + ) def __init__( self, pattern, diff --git a/tests/contrib/sensors/test_hdfs_sensor.py b/tests/contrib/sensors/test_hdfs_sensor.py index 82bada8c044ce..f9afaa16ce003 100644 --- a/tests/contrib/sensors/test_hdfs_sensor.py +++ b/tests/contrib/sensors/test_hdfs_sensor.py @@ -166,9 +166,8 @@ def setUp(self): {"kind": "directory", "name": "/nested/a", "size": 0} ] - self._mock_client, self._mock_params = MockHdfs3Client.from_file_details( - file_details, test_instance=self - ) + self._mock_client, self._mock_params = \ + MockHdfs3Client.from_file_details(file_details, test_instance=self) self._default_task_kws = { "timeout": 1, From fb672aa50000c5025201aea110ff6b03dbd62782 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Mon, 30 Jul 2018 09:36:59 +0200 Subject: [PATCH 13/15] Fixed missing default args. --- airflow/utils/deprecation.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/airflow/utils/deprecation.py b/airflow/utils/deprecation.py index 8bce49cae5373..0e4aa9a7159a1 100644 --- a/airflow/utils/deprecation.py +++ b/airflow/utils/deprecation.py @@ -41,6 +41,9 @@ def deprecated_args(renamed=None, dropped=None): in the new function/method. """ + renamed = renamed or {} + dropped = dropped or set() + def decorator(function): @functools.wraps(function) def wrapper(*args, **kwargs): From 12c924f6707570ef4405ed239a8ec635284b3988 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Fri, 10 Aug 2018 16:27:05 +0200 Subject: [PATCH 14/15] Remove deprecation code. --- airflow/contrib/sensors/hdfs_sensor.py | 66 ---------- airflow/hooks/hdfs_hook.py | 4 - airflow/sensors/hdfs_sensor.py | 160 ++++++++++++++----------- tests/sensors/test_hdfs_sensor.py | 33 +---- 4 files changed, 93 insertions(+), 170 deletions(-) delete mode 100644 airflow/contrib/sensors/hdfs_sensor.py diff --git a/airflow/contrib/sensors/hdfs_sensor.py b/airflow/contrib/sensors/hdfs_sensor.py deleted file mode 100644 index edd6e9cb6b2f9..0000000000000 --- a/airflow/contrib/sensors/hdfs_sensor.py +++ /dev/null @@ -1,66 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -from functools import partial -import posixpath - -from airflow.sensors import hdfs_sensor -from airflow.utils.deprecation import RenamedClass, deprecated_args - - -class HdfsRegexFileSensor(hdfs_sensor.HdfsFileSensor): - """HdfsSensor subclass that filters using a specific regex.""" - - @deprecated_args(renamed={"filepath": "pattern"}) - def __init__(self, pattern, regex, **kwargs): - if not self._is_pattern(pattern): - # If file path is not a pattern, we assume it is a directory - # containing files that we want to match the regex against. - # This matches the legacy behaviour of the sensor. - pattern = posixpath.join(pattern, "*") - - super(HdfsRegexFileSensor, self).__init__( - pattern=pattern, - filters=[partial(filter_regex, regex=regex)], - **kwargs - ) - - @staticmethod - def _is_pattern(path_): - """Checks if given path contains any glob patterns.""" - return "*" in path_ or "[" in path_ - - -def filter_regex(_, file_paths, regex): - """Filters file paths for given regex.""" - - for file_path in file_paths: - if regex.match(posixpath.basename(file_path)): - yield file_path - - -HdfsSensorRegex = RenamedClass( - "HdfsSensorRegex", new_class=HdfsRegexFileSensor, old_module=__name__ -) - -HdfsSensorFolder = RenamedClass( - "HdfsSensorFolder", - new_class=hdfs_sensor.HdfsFolderSensor, - old_module=__name__ -) diff --git a/airflow/hooks/hdfs_hook.py b/airflow/hooks/hdfs_hook.py index e46aac8908af0..2ef51d873282a 100644 --- a/airflow/hooks/hdfs_hook.py +++ b/airflow/hooks/hdfs_hook.py @@ -22,7 +22,6 @@ from airflow import configuration from airflow.hooks.base_hook import BaseHook -from airflow.utils.deprecation import RenamedClass class HdfsHook(BaseHook): @@ -132,6 +131,3 @@ def close(self): if self._conn is not None: self._conn.disconnect() self._conn = None - - -HDFSHook = RenamedClass("HDFSHook", new_class=HdfsHook, old_module=__name__) diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py index 9947476de1bbb..155903ed3607a 100644 --- a/airflow/sensors/hdfs_sensor.py +++ b/airflow/sensors/hdfs_sensor.py @@ -17,31 +17,31 @@ # specific language governing permissions and limitations # under the License. -import functools import posixpath from airflow import settings from airflow.hooks.hdfs_hook import HdfsHook from airflow.sensors.base_sensor_operator import BaseSensorOperator from airflow.utils.decorators import apply_defaults -from airflow.utils.deprecation import deprecated_args, RenamedClass class HdfsFileSensor(BaseSensorOperator): - """Waits for file(s) to land in HDFS.""" + """Sensor that waits for files matching a specific (glob) pattern to land in HDFS. + + :param str file_pattern: Glob pattern to match. + :param str conn_id: Connection to use. + :param Iterable[FilePathFilter] filters: Optional list of filters that can be + used to apply further filtering to any file paths matching the glob pattern. + Any files that fail a filter are dropped from consideration. + :param int min_size: Minimum size (in MB) for files to be considered. Can be used + to filter any intermediate files that are below the expected file size. + :param Set[str] ignore_exts: File extensions to ignore. By default, files with + a '_COPYING_' extension are ignored, as these represent temporary files. + """ template_fields = ("_pattern",) ui_color = settings.WEB_COLORS["LIGHTBLUE"] - @deprecated_args( - renamed={ - "filepath": "pattern", - "hdfs_conn_id": "conn_id", - "file_size": "min_size", - "ignored_ext": "ignore_exts", - }, - dropped={"ignore_copying", "hook"}, - ) @apply_defaults def __init__( self, @@ -56,29 +56,17 @@ def __init__( # Min-size and ignore-ext filters are added via # arguments for backwards compatibility. - default_filters = self._default_filters( - min_size=min_size, ignore_exts=ignore_exts - ) - filters = default_filters + (filters or []) + filters = list(filters or []) - self._pattern = pattern - self._conn_id = conn_id - self._filters = filters - - self._min_size = min_size - self._ignore_exts = set(ignore_exts) - - @classmethod - def _default_filters(cls, min_size=None, ignore_exts=None): - filters = [] - - if min_size is not None: - filters.append(functools.partial(filter_by_size, min_size=min_size)) + if min_size: + filters.append(SizeFilter(min_size=min_size)) if ignore_exts: - filters.append(functools.partial(filter_for_exts, exts=ignore_exts)) + filters.append(ExtFilter(exts=ignore_exts)) - return filters + self._pattern = pattern + self._conn_id = conn_id + self._filters = filters def poke(self, context): with HdfsHook(self._conn_id) as hook: @@ -89,8 +77,7 @@ def poke(self, context): try: file_paths = [ - fp for fp in conn.glob(self._pattern) - if not conn.isdir(fp) + fp for fp in conn.glob(self._pattern) if not conn.isdir(fp) ] except IOError: # File path doesn't exist yet. @@ -99,8 +86,8 @@ def poke(self, context): self.log.info("Files matching pattern: %s", file_paths) # Filter using any provided filters. - for filter_func in self._filters: - file_paths = filter_func(hook, file_paths) + for filter_ in self._filters: + file_paths = filter_(file_paths, hook) file_paths = list(file_paths) self.log.info("Filters after filtering: %s", file_paths) @@ -108,26 +95,33 @@ def poke(self, context): return bool(file_paths) -HdfsSensor = RenamedClass( - "HdfsSensor", - new_class=HdfsFileSensor, - old_module=__name__) - - class HdfsFolderSensor(BaseSensorOperator): - """Waits for folders to lands in HDFS.""" + """Waits for folder(s) matching the given pattern to be created in HDFS. + + By default, the sensor does not check whether any matched folders contain + files. If folders are required to (not) be empty, this behaviour can be modified + using the `require_empty` and `require_not_empty` parameters. If `require_empty` + is True, the sensor fails if any matched folders contain files. Similarly, if + `require_not_empty` is True, the sensor fails if any matched folders do not contain + files. The list of file paths considered in these checks can be modified using the + `sub_pattern` and `sub_filters` parameters. + + :param str pattern: Glob pattern to match. + :param str conn_id: Connection to use. + :param bool require_empty: Whether folders are required to be empty. If true, + the sensor fails if any of the matched directories is not empty. + :param bool require_not_empty: Whether folders are required to be NOT empty. + If true, the sensor fails if any of the matched directories is empty. + :param str sub_pattern: Glob pattern to filter nested file paths on when + checking whether directories are (not) empty. + :param Iterable[FilePathFilter] sub_filters: File path filters that should be used + to filter nested file paths on when checking whether directories are (not) + empty. See `HdfsFileSensor` for more details. + """ template_fields = ("_pattern",) ui_color = settings.WEB_COLORS["LIGHTBLUE"] - @deprecated_args( - renamed={ - "filepath": "pattern", - "hdfs_conn_id": "conn_id", - "be_empty": "require_empty" - }, - dropped=["ignored_ext", "min_size"] - ) def __init__( self, pattern, @@ -153,7 +147,7 @@ def __init__( self._require_not_empty = require_not_empty self._sub_pattern = sub_pattern or "*" - self._sub_filters = sub_filters or [] + self._sub_filters = list(sub_filters or []) def poke(self, context): with HdfsHook(self._conn_id) as hook: @@ -174,7 +168,7 @@ def poke(self, context): if self._require_empty or self._require_not_empty: self.log.info( - "Checking for files or subdirectories " "matching pattern: %s", + "Checking for files or subdirectories matching pattern: %s", self._sub_pattern, ) @@ -184,16 +178,16 @@ def poke(self, context): sub_pattern = posixpath.join(dir_path, self._sub_pattern) sub_paths = conn.glob(sub_pattern) - for filter_func in self._sub_filters: - sub_paths = filter_func(hook, sub_paths) + for filter_ in self._sub_filters: + sub_paths = filter_(sub_paths, hook) sub_paths = list(sub_paths) - is_empty = not sub_paths self.log.info( "Sub-directories/files matching pattern: %s", sub_paths ) + is_empty = not sub_paths if (self._require_empty and not is_empty) or ( self._require_not_empty and is_empty ): @@ -202,22 +196,52 @@ def poke(self, context): return bool(dir_paths) -def filter_by_size(hook, file_paths, min_size): - """Filters any HDFS files below a minimum file size.""" +class FilePathFilter: + """Base file path filter class. + + Allows a given list of file paths to be filtered on a given criteria. + Examples include a size filter (which filters files for a given minimum file + size) and a extension filter (which filters files with a given extension). + + Filters are required to implement a single __call__ method, which takes the + file system hook and a set of file paths to filter. As a result, this method + should return the set of filtered file paths. + """ + + def __call__(self, file_paths, hook): + raise NotImplementedError() + + +class SizeFilter(FilePathFilter): + """Filter that drops any file paths below the given file size. + + :param int min_size: Minimum file size in megabytes. + """ + + def __init__(self, min_size): + self._min_size = min_size + + def __call__(self, file_paths, hook): + min_size_mb = self._min_size * settings.MEGABYTE + + conn = hook.get_conn() + for file_path in file_paths: + info = conn.info(file_path) - conn = hook.get_conn() - min_size_mb = min_size * settings.MEGABYTE + if info["kind"] == "file" and info["size"] > min_size_mb: + yield file_path - for file_path in file_paths: - info = conn.info(file_path) - if info["kind"] == "file" and info["size"] > min_size_mb: - yield file_path +class ExtFilter(FilePathFilter): + """Filter that drops any file paths with given extensions. + :param Set[str] exts: Set of extensions to filter for. + """ -def filter_for_exts(_, file_paths, exts): - """Filters any HDFS files with the given extensions.""" + def __init__(self, exts): + self._exts = set(exts) - for file_path in file_paths: - if posixpath.splitext(file_path)[1][1:] not in exts: - yield file_path + def __call__(self, file_paths, hook): + for file_path in file_paths: + if posixpath.splitext(file_path)[1][1:] not in self._exts: + yield file_path diff --git a/tests/sensors/test_hdfs_sensor.py b/tests/sensors/test_hdfs_sensor.py index 822d8c37b23e7..db2ac5092e571 100644 --- a/tests/sensors/test_hdfs_sensor.py +++ b/tests/sensors/test_hdfs_sensor.py @@ -21,13 +21,11 @@ import fnmatch import posixpath import unittest -import warnings import mock from airflow import models -from airflow.sensors.hdfs_sensor import (HdfsSensor, HdfsFileSensor, - HdfsFolderSensor, HdfsHook) +from airflow.sensors.hdfs_sensor import HdfsFileSensor, HdfsFolderSensor, HdfsHook class MockHdfs3Client(object): @@ -199,35 +197,6 @@ def test_file_filter_ext(self): ) self.assertFalse(task.poke(context={})) - def test_file_filter_ext_old(self): - """Tests poking for file while filtering for extension - with deprecated ignored_ext argument.""" - - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - - task = HdfsFileSensor( - task_id="existing_file_large", - pattern="/data/not_empty/f*", - ignored_ext=("_COPYING_",), - **self._default_task_kws - ) - self.assertFalse(task.poke(context={})) - - def test_old_class(self): - """Tests sensor with old class name.""" - - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - - task = HdfsSensor( - task_id="existing_file", - pattern="/data/not_empty/small.txt", - **self._default_task_kws - ) - - self.assertTrue(task.poke(context={})) - class HdfsFolderSensorTests(unittest.TestCase): def setUp(self): From 0cf55871a71d23579ae224ca17211261f9784b75 Mon Sep 17 00:00:00 2001 From: Julian de Ruiter Date: Fri, 10 Aug 2018 16:41:44 +0200 Subject: [PATCH 15/15] Remove deprecation helpers. --- airflow/utils/deprecation.py | 160 ----------------------------------- 1 file changed, 160 deletions(-) delete mode 100644 airflow/utils/deprecation.py diff --git a/airflow/utils/deprecation.py b/airflow/utils/deprecation.py deleted file mode 100644 index 0e4aa9a7159a1..0000000000000 --- a/airflow/utils/deprecation.py +++ /dev/null @@ -1,160 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -from __future__ import absolute_import -from __future__ import division -from __future__ import print_function -from __future__ import unicode_literals - -import functools -import warnings - - -def deprecated_args(renamed=None, dropped=None): - """Decorator for the deprecation of renamed/removed keyword arguments. - - Wraps functions with changed keyword arguments (either renamed to new - arguments or dropped entirely). Functions calls with deprecated arguments - raise appropriate warnings. For removed arguments, any given values are - ignored (outside of the warning). For renamed arguments, values are - transparently proxied to their new argument names. - - :param dict[str, str] renamed: Dict mapping old arguments to their - argument names in the new function/method. - :param list[str] dropped: List of arguments that have been removed - in the new function/method. - """ - - renamed = renamed or {} - dropped = dropped or set() - - def decorator(function): - @functools.wraps(function) - def wrapper(*args, **kwargs): - new_kwargs = {} - - for key in kwargs: - if key in dropped: - warnings.warn( - "Argument {!r} is no longer supported and will be" - "removed in a future version of Airflow.".format(key), - category=DeprecationWarning, - ) - elif key in renamed: - warnings.warn( - "Argument {!r} has been renamed to {!r}. The old name " - "will no longer be supported in a future version of " - "Airflow.".format(key, renamed[key]), - category=DeprecationWarning, - ) - - new_kwargs[renamed[key]] = kwargs[key] - else: - new_kwargs[key] = kwargs[key] - - return function(*args, **new_kwargs) - - return wrapper - - return decorator - - -def deprecated(new_name=None): - """This is a decorator which can be used to mark functions as deprecated. - - The decorator ensures a warning is emitted whenever the function is - called to warn the user of its deprecated status. The parameter new_name - can be used to indicate a replacing function, if the deprecated function - has been renamed or replaced. - - :param str new_name: Optional name of a replacing function, if applicable. - """ - - def decorator(function): - @functools.wraps(function) - def wrapper(*args, **kwargs): - if new_name: - message = ( - "{} has been deprecated and will be replaced by " - "{} in a future version of Airflow.".format( - function.__name__, new_name - ) - ) - else: - message = ( - "{} has been deprecated and will be removed " - "in a future version of Airflow.".format(function.__name__) - ) - warnings.warn(message, category=DeprecationWarning) - return function(*args, **kwargs) - - return wrapper - - return decorator - - -class RenamedClass(object): - """Helper class used for deprecating old classes that have new names. - - For example, we can use this class to rename the (old) class - `HDFSHook` to it's new class name `HdfsHook` as follows: - - class HdfsHook(object): - ... - - HDFSHook = RenamedClass('HDFSHook', new_class=HdfsHook) - - so that old code can still use the deprecated form: - - hook = HDFSHook(...) - - which will raise an appropriate warning when called. - - :param str old_name: Name of the old class. - :param class new_class: The replacing class. - :param str old_module: Name of the module containing the old class. - """ - - def __init__(self, old_name, new_class, old_module=None): - self._old_name = old_name - self._old_module = old_module - self._new_class = new_class - - def _warn(self): - old_name = self._old_name - - if self._old_module: - old_name = self._old_module + '.' + old_name - - message = ("Class {!r} has been renamed to {!r}. Support for the old " - "class name will be removed in future versions of Airflow." - .format(old_name, _full_class_name(self._new_class))) - warnings.warn(message, category=DeprecationWarning) - - def __call__(self, *args, **kwargs): - self._warn() - return self._new_class(*args, **kwargs) - - def __getattr__(self, attr): - self._warn() - return getattr(self._new_class, attr) - - -def _full_class_name(cls): - return cls.__module__ + "." + cls.__name__