Added SparkSubmitTask and deprecated SparkJob, Spark1xJob and PySpark1xJob

[jthi3rry]

@Tarrasch, @berkerpeksag, this is my follow up PR from comments in #806.

I deprecated both Spark1xJob and PySpark1xJob and created a single SparkSubmitTask that follows the command usage and can handle java, scala and python apps using spark standalone, yarn or mesos. I'm wondering whether SparkJob should be deprecated too.

Happy to discuss and make any changes.

Thanks

[landscape-bot]

Repository health increased by 0.00% when pulling 0c8606b on jthi3rry:master into fed8dd5 on spotify:master.

• 3 new problems were found (including 0 errors and 1 code smell).
• 1 problem was fixed (including 0 errors and 0 code smells).

[landscape-bot]

Repository health increased by 0.00% when pulling 203ffd4 on jthi3rry:master into fed8dd5 on spotify:master.

• 3 new problems were found (including 0 errors and 1 code smell).
• 1 problem was fixed (including 0 errors and 0 code smells).

[berkerpeksag]

It would be better to just say something like "See spark-submit -h for more information.". Or just add documentation link of the spark-submit script.

[erikbern]

Looks great but on a more general note I think it would be awesome if we can support inline Python code like in luigi.hadoop.

[landscape-bot]

Repository health increased by 0.00% when pulling 1ed80a4 on jthi3rry:master into fed8dd5 on spotify:master.

• 5 new problems were found (including 0 errors and 1 code smell).
• 2 problems were fixed (including 0 errors and 0 code smells).

[jthi3rry]

Thanks for the review. Updated PR contains:

• Incorporation of @berkerpeksag's comments and more global configuration options for the class and in the docs.
• Deprecation warnings using version 1.1.0. @Tarrasch, let me know if that changes and I'll update.
• Deprecation of SparkJob too as it seemed very YARN specific. Does that sound alright?

@erikbern, that would be awesome as an additional feature. A simple support for spark submit is necessary though to be able to use existing python drivers and java or scala drivers.

That could be introduced as a subclass (PySparkTask?) with a main(sc) abstract method where sc is a given spark context (like when using the pyspark shell). I'll have a look at luigi.hadoop to see how it's done. That's a good one for another PR :-)

On an unrelated note. Is there any kind of support in luigi for streaming tasks? It could be good to also be able to incorporate spark streaming in pipelines.

[jthi3rry]

@erikbern, added PySparkTask to run spark jobs inline.

Usage looks like this:

class MyJob(PySparkTask):
    param = IntParameter(default=2)

    def output(self):
        # return a target accessible from the spark cluster (HDFS, S3, ...)

    def main(self, sc, *args):
        exponent = self.param
        output = self.output().path
        sc.parallelize([1, 2, 3]).map(lambda x: x ** exponent).saveAsTextFile(output)

[erikbern]

That's really cool – can you map over an input source as well?

[erikbern]

If you want to, feel free to leave out the last commit for a separate PR. That way we can merge what we have so far

[jthi3rry]

I removed the last commit and squashed the existing ones together.

Yes, the input path can be used as a source by doing something like sc.textFile(self.input().path)

[landscape-bot]

Repository health decreased by 0.01% when pulling 021fcba on jthi3rry:master into 10064c8 on spotify:master.

• 7 new problems were found (including 0 errors and 2 code smells).
• 2 problems were fixed (including 0 errors and 0 code smells).

[ghost]

If stdout and stderr will always be text, you can open the process with universal_newline=True. The process will accept unicode as stdin and return unicode.

Also, any reason for env=env. According to my understanding of the doc, it's the default behavior.

[jthi3rry]

Added universal_newline=True to #837.

#837 adds a get_environment() that alters the env before passing it to Popen.

[erikbern]

Something seems weird with the delta now. Can you rebase on master?

[jthi3rry]

Done @erikbern, does that fix it?

[erikbern]

Yes looks good. Is this ready to be merged? How backwards-incompatible is it? Happy to merge just want to make sure there's no risk

[jthi3rry]

It should be fully backwards compatible, all existing unit tests pass, SparkJob is unchanged, (Py)Spark1xJob's have more features now as they extend SparkSubmitTask but their behaviour is unchanged.