Hi Imran,
Thank you for your email.
In examing the condition (t2 - t1) < (t_ser + t_deser + t_exec), I
have found it to be true, although I have not included the
t_{wait_for_read} in this, since it is---so far as I can tell---been
either zero or negligible compared to the task time.
Thanks,
Mike
On 6/8/15, Imran Rashid <iras...@cloudera.com> wrote:
> Hi Mike,
>
> all good questions, let me take a stab at answering them:
>
> 1. Event Logs + Stages:
>
> Its normal for stages to get skipped if they are shuffle map stages, which
> get read multiple times. Eg., here's a little example program I wrote
> earlier to demonstrate this: "d3" doesn't need to be re-shuffled since each
> time its read w/ the same partitioner. So skipping stages in this way is a
> good thing:
>
> val partitioner = new org.apache.spark.HashPartitioner(10)
> val d3 = sc.parallelize(1 to 100).map { x => (x % 10) ->
> x}.partitionBy(partitioner)
> (0 until 5).foreach { idx =>
> val otherData = sc.parallelize(1 to (idx * 100)).map{ x => (x % 10) ->
> x}.partitionBy(partitioner)
> println(idx + " ---> " + otherData.join(d3).count())
> }
>
> If you run this, f you look in the UI you'd see that all jobs except for
> the first one have one stage that is skipped. You will also see this in
> the log:
>
> 15/06/08 10:52:37 INFO DAGScheduler: Parents of final stage: List(Stage 12,
> Stage 13)
>
> 15/06/08 10:52:37 INFO DAGScheduler: Missing parents: List(Stage 13)
>
> Admittedly that is not very clear, but that is sort of indicating to you
> that the DAGScheduler first created stage 12 as a necessary step, and then
> later on changed its mind by realizing that everything it needed for stage
> 12 already existed, so there was nothing to do.
>
>
> 2. Extracting Event Log Information
>
> maybe you are interested in SparkListener ? Though unfortunately, I don't
> know of a good blog post describing it, hopefully the docs are clear ...
>
> 3. Time Metrics in Spark Event Log
>
> This is a great question. I *think* the only exception is that t_gc is
> really overlapped with t_exec. So I think you should really expect
>
> (t2 - t1) < (t_ser + t_deser + t_exec)
>
> I am not 100% sure about this, though. I'd be curious if that was
> constraint was ever violated.
>
>
> As for your question on shuffle read vs. shuffle write time -- I wouldn't
> necessarily expect the same stage to have times for both shuffle read &
> shuffle write -- in the simplest case, you'll have shuffle write times in
> one, and shuffle read times in the next one. But even taking that into
> account, there is a difference in the way they work & are measured.
> shuffle read operations are pipelined and the way we measure shuffle read,
> its just how much time is spent *waiting* for network transfer. It could
> be that there is no (measurable) wait time b/c the next blocks are fetched
> before they are needed. Shuffle writes occur in the normal task execution
> thread, though, so we (try to) measure all of it.
>
>
> On Sun, Jun 7, 2015 at 11:12 PM, Mike Hynes <91m...@gmail.com> wrote:
>
>> Hi Patrick and Akhil,
>>
>> Thank you both for your responses. This is a bit of an extended email,
>> but I'd like to:
>> 1. Answer your (Patrick) note about the "missing" stages since the IDs
>> do (briefly) appear in the event logs
>> 2. Ask for advice/experience with extracting information from the
>> event logs in a columnar, delimiter-separated format.
>> 3. Ask about the time metrics reported in the event logs; currently,
>> the elapsed time for a task does not equal the sum of the times for
>> its components
>>
>> 1. Event Logs + Stages:
>> =========================
>>
>> As I said before, In the spark logs (the log4j configurable ones from
>> the driver), I only see references to some stages, where the stage IDs
>> are not arithmetically increasing. In the event logs, however, I will
>> see reference to *every* stage, although not all stages will have
>> tasks associated with them.
>>
>> For instance, to examine the actual stages that have tasks, you can
>> see missing stages:
>> # grep -E '"Event":"SparkListenerTaskEnd"' app.log \
>> # | grep -Eo '"Stage ID":[[:digit:]]+' \
>> # | sort -n|uniq | head -n 5
>> "Stage ID":0
>> "Stage ID":1
>> "Stage ID":10
>> "Stage ID":11
>> "Stage ID":110
>>
>> However, these "missing" stages *do* appear in the event logs as Stage
>> IDs in the jobs submitted, i.e: for
>> # grep -E '"Event":"SparkListenerJobStart"' app.log | grep -Eo 'Stage
>> IDs":\[.*\]' | head -n 5
>> Stage IDs":[0,1,2]
>> Stage IDs":[5,3,4]
>> Stage IDs":[6,7,8]
>> Stage IDs":[9,10,11]
>> Stage IDs":[12,13,14]
>>
>> I do not know if this amounts to a bug, since I am not familiar with
>> the scheduler in detail. The stages have seemingly been created
>> somewhere in the DAG, but then have no associated tasks and never
>> appear again.
>>
>> 2. Extracting Event Log Information
>> ====================================
>> Currently we are running scalability tests, and are finding very poor
>> scalability for certain block matrix algorithms. I would like to have
>> finer detail about the communication time and bandwidth when data is
>> transferred between nodes.
>>
>> I would really just like to have a file with nothing but task info in
>> a format such as:
>> timestamp (ms), task ID, hostname, execution time (ms), GC time (ms), ...
>> 0010294, 1, slave-1, 503, 34, ...
>> 0010392, 2, slave-2, 543, 32, ...
>> and similarly for jobs/stages/rdd_memory/shuffle output/etc.
>>
>> I have extracted the relevant time fields from the spark event logs
>> with a sed script, but I wonder if there is an even more expedient
>> way. Unfortunately, I do not immediately see how to do this using the
>> $SPARK_HOME/conf/metrics.properties file and haven't come across a
>> blog/etc that describes this. Could anyone please comment on whether
>> or not a metrics configuation for this already exists?
>>
>> 3. Time Metrics in Spark Event Log
>> ==================================
>> I am confused about the times reported for tasks in the event log.
>> There are launch and finish timestamps given for each task (call them
>> t1 and t2, respectively), as well as GC time (t_gc), execution time
>> (t_exec), and serialization times (t_ser, t_deser). However the times
>> do not add up as I would have expected. I would imagine that the
>> elapsed time t2 - t1 would be slightly larger than the sum of the
>> component times. However, I can find many instances in the event logs
>> where:
>> (t2 - t1) < (t_gc + t_ser + t_deser + t_exec)
>> The difference can be 500 ms or more, which is not negligible for my
>> current execution times of ~5000 ms. I have attached a plot that
>> illustrates this.
>>
>> Regarding this, I'd like to ask:
>> 1. How exactly are these times are being measured?
>> 2. Should the sum of the component times equal the elapsed (clock)
>> time for the task?
>> 3. If not, which component(s) is(are) being excluded, and when do they
>> occur?
>> 4. There are occasionally reported measurements for Shuffle Write
>> time, but not shuffle read time. Is there a method to determine the
>> time required to shuffle data? Could this be done by look at delays
>> between the first task in a new stage and the last task in the
>> previous stage?
>>
>> Thank you very much for your time,
>> Mike
>>
>>
>> On 6/7/15, Patrick Wendell <pwend...@gmail.com> wrote:
>> > Hey Mike,
>> >
>> > Stage ID's are not guaranteed to be sequential because of the way the
>> > DAG scheduler works (only increasing). In some cases stage ID numbers
>> > are skipped when stages are generated.
>> >
>> > Any stage/ID that appears in the Spark UI is an actual stage, so if
>> > you see ID's in there, but they are not in the logs, then let us know
>> > (that would be a bug).
>> >
>> > - Patrick
>> >
>> > On Sun, Jun 7, 2015 at 9:06 AM, Akhil Das <ak...@sigmoidanalytics.com>
>> > wrote:
>> >> Are you seeing the same behavior on the driver UI? (that running on
>> >> port
>> >> 4040), If you click on the stage id header you can sort the stages
>> >> based
>> >> on
>> >> IDs.
>> >>
>> >> Thanks
>> >> Best Regards
>> >>
>> >> On Fri, Jun 5, 2015 at 10:21 PM, Mike Hynes <91m...@gmail.com> wrote:
>> >>>
>> >>> Hi folks,
>> >>>
>> >>> When I look at the output logs for an iterative Spark program, I see
>> >>> that the stage IDs are not arithmetically numbered---that is, there
>> >>> are gaps between stages and I might find log information about Stage
>> >>> 0, 1,2, 5, but not 3 or 4.
>> >>>
>> >>> As an example, the output from the Spark logs below shows what I
>> >>> mean:
>> >>>
>> >>> # grep -rE "Stage [[:digit:]]+" spark_stderr | grep finished
>> >>> 12048:INFO:DAGScheduler:Stage 0 (mapPartitions at
>> >>> blockMap.scala:1444)
>> >>> finished in 7.820 s:
>> >>> 15994:INFO:DAGScheduler:Stage 1 (map at blockMap.scala:1810) finished
>> >>> in 3.874 s:
>> >>> 18291:INFO:DAGScheduler:Stage 2 (count at blockMap.scala:1179)
>> >>> finished in 2.237 s:
>> >>> 20121:INFO:DAGScheduler:Stage 4 (map at blockMap.scala:1817) finished
>> >>> in 1.749 s:
>> >>> 21254:INFO:DAGScheduler:Stage 5 (count at blockMap.scala:1180)
>> >>> finished in 1.082 s:
>> >>> 23422:INFO:DAGScheduler:Stage 7 (map at blockMap.scala:1810) finished
>> >>> in 2.078 s:
>> >>> 24773:INFO:DAGScheduler:Stage 8 (count at blockMap.scala:1188)
>> >>> finished in 1.317 s:
>> >>> 26455:INFO:DAGScheduler:Stage 10 (map at blockMap.scala:1817)
>> >>> finished
>> >>> in 1.638 s:
>> >>> 27228:INFO:DAGScheduler:Stage 11 (count at blockMap.scala:1189)
>> >>> finished in 0.732 s:
>> >>> 27494:INFO:DAGScheduler:Stage 14 (foreach at blockMap.scala:1302)
>> >>> finished in 0.192 s:
>> >>> 27709:INFO:DAGScheduler:Stage 17 (foreach at blockMap.scala:1302)
>> >>> finished in 0.170 s:
>> >>> 28018:INFO:DAGScheduler:Stage 20 (count at blockMap.scala:1201)
>> >>> finished in 0.270 s:
>> >>> 28611:INFO:DAGScheduler:Stage 23 (map at blockMap.scala:1355)
>> >>> finished
>> >>> in 0.455 s:
>> >>> 29598:INFO:DAGScheduler:Stage 24 (count at blockMap.scala:274)
>> >>> finished in 0.928 s:
>> >>> 29954:INFO:DAGScheduler:Stage 27 (map at blockMap.scala:1355)
>> >>> finished
>> >>> in 0.305 s:
>> >>> 30390:INFO:DAGScheduler:Stage 28 (count at blockMap.scala:275)
>> >>> finished in 0.391 s:
>> >>> 30452:INFO:DAGScheduler:Stage 32 (first at
>> >>> MatrixFactorizationModel.scala:60) finished in 0.028 s:
>> >>> 30506:INFO:DAGScheduler:Stage 36 (first at
>> >>> MatrixFactorizationModel.scala:60) finished in 0.023 s:
>> >>>
>> >>> Can anyone comment on this being normal behavior? Is it indicative of
>> >>> faults causing stages to be resubmitted? I also cannot find the
>> >>> missing stages in any stage's parent List(Stage x, Stage y, ...)
>> >>>
>> >>> Thanks,
>> >>> Mike
>> >>>
>> >>>
>> >>> On 6/1/15, Reynold Xin <r...@databricks.com> wrote:
>> >>> > Thanks, René. I actually added a warning to the new JDBC
>> reader/writer
>> >>> > interface for 1.4.0.
>> >>> >
>> >>> > Even with that, I think we should support throttling JDBC;
>> >>> > otherwise
>> >>> > it's
>> >>> > too convenient for our users to DOS their production database
>> servers!
>> >>> >
>> >>> >
>> >>> > /**
>> >>> > * Construct a [[DataFrame]] representing the database table
>> >>> > accessible
>> >>> > via JDBC URL
>> >>> > * url named table. Partitions of the table will be retrieved in
>> >>> > parallel
>> >>> > based on the parameters
>> >>> > * passed to this function.
>> >>> > *
>> >>> > * * Don't create too many partitions in parallel on a large
>> cluster;
>> >>> > otherwise Spark might crash*
>> >>> > * * your external database systems.*
>> >>> > *
>> >>> > * @param url JDBC database url of the form
>> >>> > `jdbc:subprotocol:subname`
>> >>> > * @param table Name of the table in the external database.
>> >>> > * @param columnName the name of a column of integral type that
>> will
>> >>> > be
>> >>> > used for partitioning.
>> >>> > * @param lowerBound the minimum value of `columnName` used to
>> >>> > decide
>> >>> > partition stride
>> >>> > * @param upperBound the maximum value of `columnName` used to
>> >>> > decide
>> >>> > partition stride
>> >>> > * @param numPartitions the number of partitions. the range
>> >>> > `minValue`-`maxValue` will be split
>> >>> > * evenly into this many partitions
>> >>> > * @param connectionProperties JDBC database connection
>> >>> > arguments,
>> a
>> >>> > list
>> >>> > of arbitrary string
>> >>> > * tag/value. Normally at least a
>> "user"
>> >>> > and
>> >>> > "password" property
>> >>> > * should be included.
>> >>> > *
>> >>> > * @since 1.4.0
>> >>> > */
>> >>> >
>> >>> >
>> >>> > On Mon, Jun 1, 2015 at 1:54 AM, René Treffer <rtref...@gmail.com>
>> >>> > wrote:
>> >>> >
>> >>> >> Hi,
>> >>> >>
>> >>> >> I'm using sqlContext.jdbc(uri, table, where).map(_ =>
>> >>> >> 1).aggregate(0)(_+_,_+_) on an interactive shell (where "where" is
>> an
>> >>> >> Array[String] of 32 to 48 elements). (The code is tailored to
>> >>> >> your
>> >>> >> db,
>> >>> >> specifically through the where conditions, I'd have otherwise post
>> >>> >> it)
>> >>> >> That should be the DataFrame API, but I'm just trying to load
>> >>> >> everything
>> >>> >> and discard it as soon as possible :-)
>> >>> >>
>> >>> >> (1) Never do a silent drop of the values by default: it kills
>> >>> >> confidence.
>> >>> >> An option sounds reasonable. Some sort of insight / log would be
>> >>> >> great.
>> >>> >> (How many columns of what type were truncated? why?)
>> >>> >> Note that I could declare the field as string via JdbcDialects
>> (thank
>> >>> >> you
>> >>> >> guys for merging that :-) ).
>> >>> >> I have quite bad experiences with silent drops / truncates of
>> columns
>> >>> >> and
>> >>> >> thus _like_ the strict way of spark. It causes trouble but
>> >>> >> noticing
>> >>> >> later
>> >>> >> that your data was corrupted during conversion is even worse.
>> >>> >>
>> >>> >> (2) SPARK-8004 https://issues.apache.org/jira/browse/SPARK-8004
>> >>> >>
>> >>> >> (3) One option would be to make it safe to use, the other option
>> >>> >> would
>> >>> >> be
>> >>> >> to document the behavior (s.th. like "WARNING: this method tries
>> >>> >> to
>> >>> >> load
>> >>> >> as many partitions as possible, make sure your database can handle
>> >>> >> the
>> >>> >> load
>> >>> >> or load them in chunks and use union"). SPARK-8008
>> >>> >> https://issues.apache.org/jira/browse/SPARK-8008
>> >>> >>
>> >>> >> Regards,
>> >>> >> Rene Treffer
>> >>> >>
>> >>> >
>> >>>
>> >>>
>> >>> --
>> >>> Thanks,
>> >>> Mike
>> >>>
>> >>> ---------------------------------------------------------------------
>> >>> To unsubscribe, e-mail: dev-unsubscr...@spark.apache.org
>> >>> For additional commands, e-mail: dev-h...@spark.apache.org
>> >>>
>> >>
>> >
>>
>>
>> --
>> Thanks,
>> Mike
>>
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: dev-unsubscr...@spark.apache.org
>> For additional commands, e-mail: dev-h...@spark.apache.org
>>
>
--
Thanks,
Mike
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscr...@spark.apache.org
For additional commands, e-mail: dev-h...@spark.apache.org
