Provide automatic URL tracking for Spark applications #2661

GoodDok · 2019-02-14T22:51:07Z

Description

A new string parameter tracking_url_pattern was added to ExternalProgramTask task in order to provide a mechanism for parsing the task output and update tracking URL in Luigi web UI on the run. Also the new parameter is set for SparkSubmitTask according to the log message structure: https://github.com/apache/spark/blob/0a4c03f7d084f1d2aa48673b99f3b9496893ce8d/core/src/main/scala/org/apache/spark/ui/WebUI.scala#L133

Motivation and Context

One of the most missing things to me in Luigi, while I'm working with Spark applications, is automatic detection of the link for Spark Web UI. It is very uncomfortable to look for Spark jobs e.g. in Yarn UI among thousands of other jobs, having only the name of the Luigi task plus it's parameters.
The existence of such feature in Luigi would hugely simplify the process of tracking Spark jobs that are running at the moment.

Have you tested this? If so, how?

I ran my jobs with this code and it works for me.

Tarrasch

This would be really useful indeed!

Can you address the comments I've written?

Tarrasch · 2019-02-19T20:58:44Z

luigi/contrib/external_program.py

@@ -60,6 +63,7 @@ class ExternalProgramTask(luigi.Task):
    """

    capture_output = luigi.BoolParameter(default=True, significant=False, positional=False)
+    tracking_url_pattern = luigi.Parameter(default="", significant=False, positional=False)


Can you document this somewhere?

Added description + documentation below

Tarrasch · 2019-02-19T21:03:59Z

luigi/contrib/external_program.py

@@ -132,6 +140,47 @@ def run(self):
                tmp_stdout.close()


+class TrackingUrlContext(object):


Can we make this private? _TrackingUrlContext - then you don't need to add documentation

Thanks, done

Tarrasch · 2019-02-19T21:04:45Z

luigi/contrib/external_program.py

+        self.set_tracking_url = set_tracking_url
+        self.track_proc = Process(target=self._track_url_by_pattern)
+
+    def __enter__(self):


Is it possible to use contextlib? This reminds me of #2652. :)

Thanks, done, now looks cleaner

Tarrasch · 2019-02-19T21:05:46Z

luigi/contrib/external_program.py

+        return self.main_proc
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        # need to wait a bit to let the subprocess read the last lines


Is this the right way to do it? Is there no signalling or waiting for termination way?

replaced sleep with track_proc.join(timeout), not sure if I have any other option, what do you think?

Tarrasch · 2019-02-19T21:08:36Z

test/contrib/external_program_test.py

+        temp_dir = tempfile.mkdtemp()
+
+        # since set_tracking_url is called in a separate process
+        # we need to do something like creating a temp file


Maybe you want https://docs.python.org/3/library/multiprocessing.html#sharing-state-between-processes ?

Thank you, changed, now looks much better

Tarrasch

I'm so happy that most of my comments were useful and led to better code. And thank you so much for the first turn around time!!

Anyway I'm fine with this being merged once my last comments are addressed. I really like this PR. Thanks!

Tarrasch · 2019-02-20T21:29:58Z

luigi/contrib/external_program.py

+    for the job in the web UI. Example: 'Job UI is here: (https?://.*)'.
+
+    Default value is an empty string, so URL tracking is not performed.
+    """


Does this docs render nice?

Hmm, how can I check it?

I recently wrote about it here: #2548 (comment)

Tarrasch · 2019-02-20T21:30:26Z

luigi/contrib/external_program.py

@@ -131,6 +148,43 @@ def run(self):
                tmp_stderr.close()
                tmp_stdout.close()

+    @contextmanager
+    def proc_with_tracking_url_context(self, proc_args, proc_kwargs):


You made it public again -.-

My bad, fixed

Tarrasch · 2019-02-20T21:31:45Z

luigi/contrib/external_program.py

+            track_proc.join(time_to_sleep * 2)
+            if track_proc.is_alive():
+                track_proc.terminate()
+            pipe_to_read.close()


Thanks for updating this part of the code. Since it's always easy to make a mistake, would you mind trying to run this for a week in production for you?

Yup, did it with the latest changes, I'll let you know how it goes

GoodDok · 2019-02-21T18:09:40Z

@Tarrasch thank you for being so responsive!
While deploying these changes for the working pipeline I figured out that by default Spark apps write all the logs into stderr (before I was refactoring the code on my local machine and it seems like my Spark config is somewhat unusual writing to stdout). Also, the format of output logs for spark depends on deploy-mode used in spark-submit.
So I introduced a new parameter track_url_in_stderr that is False by default + added a test for it. Please let me know if anything else should be added.
Now it's working with the real pipeline, let's take some time, I'll be back with the update.

Tarrasch

Thanks. Let me know when it's been used in prod reliably!

Tarrasch · 2019-02-21T19:26:40Z

luigi/contrib/external_program.py

+    for the job in the web UI. Example: 'Job UI is here: (https?://.*)'.
+
+    Default value is an empty string, so URL tracking is not performed.
+    """


I recently wrote about it here: #2548 (comment)

Tarrasch · 2019-02-21T19:29:07Z

luigi/contrib/external_program.py

+
+    Default value is an empty string, so URL tracking is not performed.
+    """
+    track_url_in_stderr = luigi.BoolParameter(


Maybe you want track_url_in_stderr = luigi.ChoiceParameter?

GoodDok · 2019-03-04T09:06:30Z

Hi @Tarrasch, just wanted to notify that the changes were applied to the real pipeline and no issues were met during the period of testing.
Also, I was thinking about the changed you suggested regarding luigi.parameter.ChoiceParameter and committed yet another version that seems to me a bit easier to understand.
How it was before:

tracking_url_pattern = luigi.Parameter(
        default="", significant=False, positional=False, visibility=ParameterVisibility.HIDDEN,
        description="Regex pattern used for searching URL in the logs of the external program")

track_url_in_stderr = luigi.BoolParameter(
        default=False, significant=False, positional=False, visibility=ParameterVisibility.HIDDEN,
        description="When set to True, 'tracking_url_pattern' is searched in stderr instead of stdout")

How it is now:

stream_for_searching_tracking_url = luigi.parameter.ChoiceParameter(
        var_type=str, choices=['none', 'stdout', 'stderr'], default='none',
        significant=False, positional=False, visibility=ParameterVisibility.HIDDEN,
        description="Stream for searching tracking URL")

tracking_url_pattern = luigi.OptionalParameter(
        default=None, significant=False, positional=False, visibility=ParameterVisibility.HIDDEN,
        description="Regex pattern used for searching URL in the logs of the external program")

How do you think, is it a better option?
The changes are contained in the last commit (they haven't been tested in prod), and the thing that surprises me a lot is a failing codecov/patch: the lines related to the context manager are marked as not covered although there are several tests covering this part...

Tarrasch · 2019-03-05T20:15:17Z

Feel free to ignore CodeCov.

I would maybe have preferred

tracking_url_pattern = luigi.Parameter(...)

But it's not worth blocking this PR more for that. :)

GoodDok · 2019-03-06T09:35:38Z

@Tarrasch thank you very much for a thorough review of my changes! 👍
One more thing, after fetching my last changes to the real pipeline I found out that I forgot to apply the changes to spark.py itself, there must be
stream_for_searching_tracking_url = 'stderr'
instead of
track_url_in_stderr = True
according to the latest changes.
Here is the commit with change: GoodDok@68eee5b
What is the most convenient way to apply it after PR has been merged?

Tarrasch · 2019-03-08T07:11:04Z

Please send a fix, also with a test if it makes sense to avoid future breakages.

GoodDok · 2019-03-11T20:47:04Z

@Tarrasch here it is: ae065b3
The changes contain the correctly named parameter + two tests for cluster and client modes.
I tested the changes locally, but I wanted to run the remote ones also. Should I create a new PR to test these changes remotely?

Tarrasch · 2019-03-11T21:54:37Z

I can only merge something (conveniently) if you send it as a Pull Request. That way Travis will also run.

GoodDok marked this pull request as ready for review February 14, 2019 22:53

GoodDok requested review from dlstadther, honnix and Tarrasch as code owners February 14, 2019 22:53

GoodDok force-pushed the spark-task-with-tracking-url branch 3 times, most recently from 587c4db to 8c67161 Compare February 15, 2019 07:42

New option with tracking url in output logs for Spark applications

def262b

GoodDok force-pushed the spark-task-with-tracking-url branch from 8c67161 to def262b Compare February 18, 2019 22:24

Tarrasch approved these changes Feb 19, 2019

View reviewed changes

GoodDok force-pushed the spark-task-with-tracking-url branch from 6d3e4f7 to 7ed7fe3 Compare February 20, 2019 11:34

Changes after the review

59114a7

GoodDok force-pushed the spark-task-with-tracking-url branch from 7ed7fe3 to 59114a7 Compare February 20, 2019 11:43

Changes after the review

3c292a1

Tarrasch approved these changes Feb 20, 2019

View reviewed changes

GoodDok force-pushed the spark-task-with-tracking-url branch from f2dc2ba to 7eefdaa Compare February 21, 2019 17:27

Changes after the review

fc02a8a

GoodDok force-pushed the spark-task-with-tracking-url branch from 7eefdaa to fc02a8a Compare February 21, 2019 17:57

Tarrasch approved these changes Feb 21, 2019

View reviewed changes

GoodDok force-pushed the spark-task-with-tracking-url branch 3 times, most recently from 024417e to b62ff46 Compare March 4, 2019 00:16

Changes after the review

2d718da

GoodDok force-pushed the spark-task-with-tracking-url branch from b62ff46 to 2d718da Compare March 4, 2019 00:54

Tarrasch merged commit 3701c68 into spotify:master Mar 5, 2019

GoodDok mentioned this pull request Mar 12, 2019

Enable automatic URL tracking for Spark applications #2669

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Provide automatic URL tracking for Spark applications #2661

Provide automatic URL tracking for Spark applications #2661

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

		@@ -132,6 +140,47 @@ def run(self):
		tmp_stdout.close()


		class TrackingUrlContext(object):

Provide automatic URL tracking for Spark applications #2661

Provide automatic URL tracking for Spark applications #2661

Uh oh!

Conversation

Description

Motivation and Context

Have you tested this? If so, how?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!