spotify/luigi — pull request #517 by mikepii (2014-10-30 22:05:03+00:00)
Title: Hadoop JobTask: distcp and S3

This PR arose out of:
- the performance limitations of `hadoop fs -mv ...` for very large output,
- a workaround for a bug in Amazon's EMRFS, and
- a need for the ability to work with S3 inputs and outputs in Hadoop (we had a hacky way of doing it in Luigi before).

## distcp

Currently, HadoopJobRunner outputs to a temporary HDFS directory and then moves the output to the correct directory with `hadoop fs -mv ...` once the job completes. According to Amazon support, Apache distcp is more efficient for moving many files since it does it in a distributed fashion, whereas `hadoop fs -mv ...` moves files one-at-a-time.

This PR lets the user set these configuration values (or a `move_strategy` variable on the task class) to use distcp intsead of `hadoop fs -mv ...`. It defaults to the old way (or the user can specify `hadoop-fs`).
```ini
[hadoop]
move-strategy=distcp ; or distcp-jar to use the below jar instead of built-in distcp
distcp-jar=/path/to/distcp.jar
```

This change is in hadoop.py instead of in `move()` hdfs.py and s3.py. I think this makes sense because distcp is a Hadoop map/reduce job. Also, it would create a circular dependency to use the Hadoop job tracking facilities from hadoop.py if the change was in hdfs.py and s3.py.

## S3

Currently, `luigi.hadoop.JobTask` requires inputs and outputs to be instances of `luigi.hdfs.HdfsTarget`. We wanted to use `luigi.hadoop.s3.S3FlagTarget` as output from one task and Hadoop input for the next. We had an ugly workaround before. This PR lets the user supply an `S3Target` for Hadoop JobTask inputs and outputs.

Additionally, there is a bug in Amazon's EMRFS which hangs forever when copying S3 files larger than 5mb through `hadoop fs -mv`. Amazon support told us to use distcp instead. Amazon users can supply the distcp jar at `/home/hadoop/lib/emr-s3distcp-1.0.jar`.