[AIRFLOW-2697] Drop snakebite in favour of hdfs3

[jrderuiter]

Make sure you have checked all steps below.

JIRA

• My PR addresses the following Airflow JIRA issues and references them in the PR title. For example, "[AIRFLOW-XXX] My Airflow PR"
  • https://issues.apache.org/jira/browse/AIRFLOW-2697

Description

• Here are some details about my PR, including screenshots of any UI changes:

This PR replaces the Snakebite connection used by HdfsHook with hdfs3 to add Python 3 support for the HdfsHook. Unfortunately, this also required some changes in dependent classes, such as the HdfsSensor, which directly use the underlying connection object and therefore needed to be updated for the library switch. This kind of issue could be avoided in the future by providing a clean interface which dependent classes rely on.

Beside this, the HDFSHook class has been renamed to HdfsHook for a more consistent naming convention. For the same reason, the filepath parameter has been renamed to file_path.

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

• tests/hooks/test_hdfs_hook.py

• tests/sensors/test_hdfs_sensor.py

Commits

• My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• In case of new functionality, my PR adds documentation that describes how to use it.
  • When adding new operators/hooks/sensors, the autoclass documentation generation needs to be added.

Code Quality

• Passes git diff upstream/master -u -- "*.py" | flake8 --diff

[jrderuiter]

The current changes break the HdfsSensorRegex and HdfsSensorFolder subclasses in airflow.contrib.sensors, which essentially add some extra filtering behaviour to the base HdfsSensor.

One thing that wasn't clear to me about the HdfsSensor itself, is whether the sensor checks for an exact file path (or directory), or some pattern. If it should check for the exact path, then the used approach of using ls and filtering doesn't make sense to me, as there can only be a single result. If it checks for a pattern (using glob for example), then I don't understand the need for the regex sensor, as most patterns should be checkable via globs. Similarly, the sensor folder could be replaced by a glob checking for files in directory, with an optional complement argument if we need an empty directory.

Besides this, the subclasses duplicate most of the code used in the base class for checking file existence. A good way to avoid this would be to allow users to supply an arbitrary function (or list of functions) for filtering down potential file paths. This may however break the current interface.

[jrderuiter]

New changes add a filters argument to the HdfsSensor, which allow passing of a set of filter functions to the HdfsSensor to narrow down the list of potential matches (similar to what the file_size, ignored_ext and regex parameters are doing now). This makes for a more flexible approach for filtering potential hits than the current subclassing approach. Filtering for size unfortunately requires the filter functions to have access to the hdfs_conn, which is not ideal.

Beside this, I added a deprecated_args decorator which supports handling renamed/removed arguments with the appropriate deprecation warnings. This may be useful for supporting argument renames in future code, if the changes in this PR are approved.

To be honest, I think I would drop the HdfsSensorRegex and HdfsSensorFolder classes from contrib. Most of the regex cases should now be covered by the HdfsHook by using globbing to match file path patterns. Similarly, the HdfsSensorFolder sensor can be simulated by either checking for the folder directly (case for be_empty = False) or checking for any files in the directory (using a glob pattern) and negating the result (functionality we could add with a complement parameter).

[gglanzani]

Hi @jrderuiter,

I don't have time to review it all now (I'll pick up the rest next week), but I've added a couple of comments in the hdfs_hook.py.

[gglanzani]

Maybe hfds_pars could be augmented like so

if configuration.conf.get("core", "security") == "kerberos":
    hdfs_pars["hadoop.security.authentication"] = "kerberos"

to make existing code not break?

[jrderuiter]

Good idea. Should the airflow configuration override whatever is set in hdfs_pars? (I.e., should we keep the value for hadoop.security.authentication if it is already set in pars or should we override it?)

[gglanzani]

How is the high availability case handled in the new version?

hdfs3 will read from the configuration files the HA settings (see docs), but if we're getting the configuration from the connection, we need to at least specify how to do so in the extra section.

I was thinking that we could specify a ha key in the extra section, so that extra looks like

{
  'ha': {
    host = "nameservice1"
    conf = {
        "dfs.nameservices": "nameservice1",
        "dfs.ha.namenodes.nameservice1": "namenode113,namenode188",
        "dfs.namenode.rpc-address.nameservice1.namenode113": "hostname_of_server1:8020",
        "dfs.namenode.rpc-address.nameservice1.namenode188": "hostname_of_server2:8020",
        "dfs.namenode.http-address.nameservice1.namenode188": "hostname_of_server1:50070",
        "dfs.namenode.http-address.nameservice1.namenode188": "hostname_of_server2:50070"
    }
} # number of braces might be wrong here :)

and then in the code we could do

ha = params.extra_dejson.get('ha', {})
if ha:
    pars.update(ha.get('conf'))

...
self._conn = hdfs3.HDFileSystem(
    host=ha.get('host') or params.host or MyNone,
...
)

What do you think?

[ashb]

The way other services support HA via connections is to have multiple rows in the Connections table with different host names but the same conn_id.

(Sorry, quick note, will try to expand on this later)

[Fokko]

Also known as the poor mans load balancing. This will not really work since there is no fallback, but it will pick a random connection, and then you need to pray that one is up.

[ashb]

True.

Could we add a new method to Connection base/hook to get all connections with the given ID, and use that here?

[Fokko]

Yes, that would work.

Apart from the HA stuff, I think we should also drop hdfs3 and go for PyArrow: https://arrow.apache.org/docs/python/filesystems.html

[Fokko]

Very nice work @jrderuiter! Since Snakebite is only Python 2.7 compatible, we need to do this anyway.

@feng-tao @vfoucault can you take a look?

[bolkedebruin]

I like it! But one reason for snakebit was support for Kerberos. libhdfs also supports it, but I am not aware if it needs special configuration to do so. If that is the case please make sure to add it and to test it.

[gglanzani]

@bolkedebruin it does, see my comment on the code above.

[Fokko]

@jrderuiter can you add high available support?

[jrderuiter]

I'll have a look this week!

[jrderuiter]

@Fokko @gglanzani HA + kerberos support has been added, could you check if this is what you had in mind?

@Fokko I still need to refactor the HdfsSensorFolder class. To help in refactoring the sensors I added a few deprecation functions (+ one class). Would it make sense to put these in airflow.utils.deprecation? Or does Airflow already have similar functionality elsewhere?

[Fokko]

One minor comment. Looks good to me. Could you also add a line in: https://github.com/apache/incubator-airflow/blob/master/UPDATING.md

Saying that we've moved from snakebite to hdfs3, and that signatures of the methods are changed and this breaks backward compatibility.

[Fokko]

This is file_pattern, right?

[gglanzani]

@jrderuiter I'm wondering how ticket cache is implemented now vs in hdfs3 (http://hdfs3.readthedocs.io/en/latest/_modules/hdfs3/core.html). Give me some time to look into it :)

[Fokko]

This won't get into 1.10 anymore. Maybe if we move this to 2.0 we can get rid of the deprecations right away. What are your thoughts on it @bolkedebruin ?

[gglanzani]

HA and Kerberos part LGTM besides a couple of details.

[gglanzani]

This part could also maybe go to the "UPDATING" doc.

[gglanzani]

Reading through hdfs3 docs, it seems that parameters such as user, ticket_cache, and token might also be useful for accessing a kerberized cluster (see here).

These parameters, however, might need to be dag specific (i.e. a dag impersonates user_a, another user_b). So we might put them in the hook's __init__. What do you think?

[jrderuiter]

I'm not sure, as I would like to avoid adding too many too specific arguments. We could add a single argument hdfs3_kwargs, which contains kwargs that are passed directly to hdfs3.HDFileSystem.

However, it might be better to keep these arguments in the connection, as this provides a more uniform interface for the hooks down the line.

[gglanzani]

So, if we provide them in the connection, it means we need a connection per different ticket_cache/user.

@Fokko Is this something usual in Airflow? Providing this in the dag code is more flexible.

@bolkedebruin How are you handling this at ING?

[jrderuiter]

Commit #52a2ff3 changes the HdfsSensor classes for the new HdfsHook. The biggest change is that it rewrites the HdfsSensor + HdfsSensorFolder (from contrib) sensors into a HdfsFileSensor and a HdfsFolderSensor class, which aim to retain the functionality of the previous classes. The main difference between the two sensors is that the former checks for files, whilst the latter checks for directories. In doing so, the HdfsFolderSensor also provides options for requiring directories to be (non-)empty, which is difficult to do if we were to keep a single HdfsSensor class. Note that this pulls the HdfsFolderSensor into airflow.hooks, effectively promoting the sensor from contrib.

Besides this, the sensor classes now also accept a list of filter functions, which can filter a list of file/directory paths before the final check. This allows users to filter for files with a given minimum size, for example.

I added a few deprecation utility classes/functions to help keep the re-written classes as backwards compatible as possible. However, if we move this to Airflow 2.0 I would consider removing these shims and simply breaking with the old interface, as adapting DAGs to these new sensors should be fairly straightforward. In that case, I would also argue for removing the HdfsRegexSensor class, which is largely superseded by the use of glob patterns in file/directory paths.

[codecov-io]

Codecov Report

Merging #3560 into master will increase coverage by 0.37%.
The diff coverage is 90.47%.

@@            Coverage Diff             @@
##           master    #3560      +/-   ##
==========================================
+ Coverage   77.16%   77.53%   +0.37%     
==========================================
  Files         206      206              
  Lines       15769    15837      +68     
==========================================
+ Hits        12168    12280     +112     
+ Misses       3601     3557      -44

Impacted Files	Coverage Δ
airflow/hooks/hdfs_hook.py	100% <100%> (+7.5%)	⬆️
airflow/utils/deprecation.py	76.47% <76.47%> (ø)
airflow/sensors/hdfs_sensor.py	95.6% <95.34%> (-4.4%)	⬇️
airflow/utils/sqlalchemy.py	73.91% <0%> (-15.22%)	⬇️
airflow/hooks/base_hook.py	88.23% <0%> (-3.93%)	⬇️
airflow/www/app.py	99.01% <0%> (-0.99%)	⬇️
airflow/www_rbac/views.py	72.85% <0%> (ø)	⬆️
airflow/www/views.py	68.88% <0%> (ø)	⬆️
airflow/minihivecluster.py
... and 3 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 49fd23a...31e2340. Read the comment docs.

[jrderuiter]

@Fokko Any thoughts on my previous comment?

[Fokko]

It is evident that a lot of the API's are changed, and this is expected when you switch the underlying API.
For me it would also be acceptable to accept these breaking changes in Airflow 2.0

[Fokko]

I added a few deprecation utility classes/functions to help keep the re-written classes as backwards compatible as possible.

This is not the case here: def __init__(self, regex, *args, **kwargs): -> def __init__(self, pattern, regex, **kwargs): will definitely break.

[Fokko]

I added a few deprecation utility classes/functions to help keep the re-written classes as backwards compatible as possible.

This is not the case here: __init__(self, hdfs_conn_id='hdfs_default', proxy_user=None, autoconfig=False) -> __init__(self, hdfs_conn_id=None)

[Fokko]

I added a few deprecation utility classes/functions to help keep the re-written classes as backwards compatible as possible.

This is not the case here: __init__(self, filepath, ..) -> __init__(self, pattern .. )

[Fokko]

And this is not consistent with the deprecation args: "filepath": "file_pattern",

[jrderuiter]

Thanks @Fokko , I'll fix the above cases.

I waited with checking everything because I wanted to hear from you guys if you're going to Airflow 2.0 (in which breaking changes would be ok) or Airflow 1.11 (in which they wouldn't).

In the case of a minor change, we could also choose to keep but deprecate the old HdfsSensor class and introduce the new HdfsFileSensor/HdfsFolderSensor classes next to the old class, after which we remove the old class in the next major version.

[bolkedebruin]

Can you rebase and squash your commits? The we can (most likely :-) ) merge.

[Fokko]

@bolkedebruin My preference would be to get rid of the deprecation warnings and target it for Apache Airflow 2.0

[bolkedebruin]

Of course! But the ball needs to get rolling too

[jrderuiter]

@Fokko As we discussed, I removed the deprecation code so we can merge this in preparation for Airflow 2.0. I want to do some final testing, afterwards we can merge if everything passes.

[Fokko]

@jrderuiter Cool, thanks! 👍 Don't forget to rebase onto master. Let me know when you're ready.

[Fokko]

Please move this sensor to a separate file.

[Fokko]

For the tests as well.

[Fokko]

as well :)

[XD-DENG]

Hi @jrderuiter @Fokko , I think it would be good to explicitly tell users that how ignore_exts should be like in the comment (which will be the documentation later).

For example, both {'.py', '.exe'} and {'py', 'exe'} seem valid, but only {'py', 'exe'} would work here.

[Fokko]

Good point @XD-DENG

We could also trim the prepended . from the extension to make both situations work.

[XD-DENG]

Then maybe also add a .lower() to make sure both {'py', 'exe'} and {'PY', 'EXE'} work?

[bolkedebruin]

I would love to have this moving forward! Can we get this rebased, so we can start getting this in 2.0?

[Fokko]

Maybe move to PyArrow directly then? https://arrow.apache.org/docs/python/filesystems.html

[gglanzani]

@jrderuiter Can you let me know if you have time/will to move this to PyArrow? Otherwise I will try to take it up on one of the next Fridays.

[zhongjiajie]

This PR is Great, I would love to see this moving forward

[kouzant]

Is there any update on that matter? Is Airflow going to use PyArrow/Pydoop/* for Python 3 support?

[Fokko]

@kouzant We want to move to PyArrow since hdfs3 is not maintained anymore.

[jrderuiter]

@Fokko Looking into it!

[zhongjiajie]

Here to check if you still work on this PR? @jrderuiter

[r-richmond]

If dropping snakebite is the goal of this PR shouldn't this be tackled as well? (snakebite in kerberos option in setup.py)

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.