You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@spark.apache.org by Mike Hynes <91...@gmail.com> on 2015/06/08 06:12:29 UTC
Stages with non-arithmetic numbering & Timing metrics in event logs
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 <pw...@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 <91...@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 <rx...@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 <rt...@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-unsubscribe@spark.apache.org
>>> For additional commands, e-mail: dev-help@spark.apache.org
>>>
>>
>
--
Thanks,
Mike
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Kay Ousterhout <ke...@eecs.berkeley.edu>.
Good idea -- I've added this to the wiki:
https://cwiki.apache.org/confluence/display/SPARK/Shuffle+Internals. Happy
to stick it elsewhere if folks think there's a more convenient place.
On Thu, Jun 11, 2015 at 4:46 PM, Gerard Maas <ge...@gmail.com> wrote:
> Kay,
>
> Excellent write-up. This should be preserved for reference somewhere
> searchable.
>
> -Gerard.
>
>
>
> On Fri, Jun 12, 2015 at 1:19 AM, Kay Ousterhout <ke...@eecs.berkeley.edu>
> wrote:
>
>> Here’s how the shuffle works. This explains what happens for a single
>> task; this will happen in parallel for each task running on the machine,
>> and as Imran said, Spark runs up to “numCores” tasks concurrently on each
>> machine. There's also an answer to the original question about why CPU use
>> is low at the very bottom.
>>
>> The key data structure used in fetching shuffle data is the “results”
>> queue in ShuffleBlockFetcherIterator, which buffers data that we have in
>> serialized (and maybe compressed) form, but haven’t yet deserialized /
>> processed. The results queue is filled by many threads fetching data over
>> the network (the number of concurrent threads fetching data is equal to the
>> number of remote executors we’re currently fetching data from) [0], and is
>> consumed by a single thread that deserializes the data and computes some
>> function over it (e.g., if you’re doing rdd.count(), the thread
>> deserializes the data and counts the number of items). As we fetch data
>> over the network, we track bytesInFlight, which is data that has been
>> requested (and possibly received) from a remote executor, but that hasn’t
>> yet been deserialized / processed by the consumer thread. So, this
>> includes all of the data in the “results” queue, and possibly more data
>> that’s currently outstanding over the network. We always issue as many
>> requests as we can, with the constraint that bytesInFlight remains less
>> than a specified maximum [1].
>>
>> In a little more detail, here’s exactly what happens when a task begins
>> reading shuffled data:
>>
>> (1) Issue requests [1.5] to fetch up to maxBytesInFlight bytes of data
>> [1] over the network (this happens here
>> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L260>
>> ).
>>
>> These requests are all executed asynchronously using a ShuffleClient [2]
>> via the shuffleClient.fetchBlocks call
>> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L149>
>> [3]. We pass in a callback that, once a block has been successfully
>> fetched, sticks it on the “results” queue.
>>
>> (2) Begin processing the local data. One by one, we request the local
>> data from the local block manager (which memory maps the file) and then stick
>> the result onto the results queue
>> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L230>.
>> Because we memory map the files, which is speedy, the local data typically
>> all ends up on the results in front of the remote data.
>>
>> (3) One the async network requests have been issued (note — issued, but
>> not finished!) and we’ve “read” (memory-mapped) the local data (i.e., (1)
>> and (2) have happened), ShuffleBlockFetcherIterator returns an iterator
>> that gets wrapped too many times to count [4] and eventually gets unrolled
>> [5]. Each time next() is called on the iterator, it blocks waiting for an
>> item from the results queue. This may return right away, or if the queue
>> is empty, will block waiting on new data from the network [6]. Before
>> returning from next(), we update our accounting for the bytes in flight:
>> the chunk of data we return is no longer considered in-flight, because it’s
>> about to be processed, so we update the current bytesInFlight, and if it
>> won’t result in > maxBytesInFlight outstanding, send some more requests for
>> data.
>>
>> ————————————————
>>
>> Notes:
>>
>> [0] Note that these threads consume almost no CPU resources, because they
>> just receive data from the OS and then execute a callback that sticks the
>> data on the results queue.
>>
>> [1] We limit the data outstanding on the network to avoid using too much
>> memory to hold the data we’ve fetched over the network but haven’t yet
>> processed.
>>
>> [1.5] Each request may include multiple shuffle blocks, where is a
>> "block" is the data output for this reduce task by a particular map task.
>> All of the reduce tasks for a shuffle read a total of # map tasks * #
>> reduce tasks shuffle blocks; each reduce task reads # map tasks blocks. We
>> do some hacks
>> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L177>
>> to try to size these requests in a "good" way: we limit each request to
>> about maxBytesInFlight / 5, so that we can fetch from roughly 5 machines
>> concurrently without exceeding maxBytesInFlight. 5 is completely a magic
>> number here that was probably guessed by someone long long ago, and it
>> seems to work ok.
>>
>> [2] The default configuration uses NettyBlockTransferService as the
>> ShuffleClient implementation (note that this extends BlockTransferService,
>> which extends ShuffleClient).
>>
>> [3] If you’re curious how the shuffle client fetches data, the default
>> Spark configuration results in exactly one TCP connection from an executor
>> to each other executor. If executor A is getting shuffle data from
>> executor B, we start by sending an OpenBlocks message from A to B. The
>> OpenBlocks message includes the list of blocks that A wants to fetch, and
>> causes the remote executor, B, to start to pull the corresponding data into
>> memory from disk (we typically memory map the files, so this may not
>> actually result in the data being read yet), and also to store some state
>> associated with this “stream” of data. The remote executor, B, responds
>> with a stream ID that helps it to identify the connection. Next, A
>> requests blocks one at a time from B using an ChunkFetchRequest message
>> (this happens here
>> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/network/shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java#L101> in
>> OneForOneBlockFetcher, which calls this code
>> <https://github.com/apache/spark/blob/master/network/common/src/main/java/org/apache/spark/network/client/TransportClient.java#L97>
>> in TransportClient; currently, we have a one-to-one mapping from a chunk to
>> a particular block). It’s possible that there are many sets of shuffle
>> data being fetched concurrently between A and B (e.g., because many tasks
>> are run concurrently). These requests are serialized, so one block is sent
>> at a time from B, and they’re sent in the order that the requests were
>> issued on A.
>>
>> [4] In BlockStoreShuffleFetcher, which handles failures; then in
>> HashShuffleReader, which helps aggregate some of the data; etc.
>>
>> [5] This happens in BlockManager.putIterator, if the RDD is going to be
>> cached; in the function passed in to ResultTask, if this is the last stage
>> in a job; or via the writer.write() call in ShuffleMapTask, if this is a
>> stage that generates intermediate shuffle data.
>>
>> [6] We time how long we spend blocking on data from the network; this is
>> what’s shown as “fetch wait time” in Spark’s UI.
>>
>> ——————————————————
>>
>> To answer the original question about why CPU use is low, this means that
>> the main thread (that pulls data off of the results queue) is blocking on
>> I/O. It could be blocking to receive data over the network (which will be
>> included in the fetch wait time shown in the UI, described above), or it
>> could be blocking to read data from the local disk (because the shuffle
>> data that is read locally is memory-mapped, so the thread may block on disk
>> if the data hasn’t been read into memory yet). It could also be blocking
>> because it’s writing new shuffle output to disk (also shown in the UI, as
>> shuffle write time), or because it’s spilling intermediate data.
>> Everything except the spilling is shown in the handy-dandy new
>> visualization that Kousuke recently added
>> <https://github.com/apache/spark/commit/a5f7b3b9c7f05598a1cc8e582e5facee1029cd5e>to
>> the UI (available in 1.4 onwards): if you look at the stage view, and click
>> on “Event Timeline”, you’ll see a visualization of the tasks and how long
>> they spent blocked on various things (we should add the spilling to
>> this…there is a long-outstanding JIRA for this).
>>
>> Hope this helps!
>>
>> -Kay
>>
>> On Thu, Jun 11, 2015 at 7:42 AM, Imran Rashid <ir...@cloudera.com>
>> wrote:
>>
>>> That is not exactly correct -- that being said I'm not 100% on these
>>> details either so I'd appreciate you double checking and / or another dev
>>> confirming my description.
>>>
>>>
>>> Spark actually has more threads going then the "numCores" you specify.
>>> "numCores" is really used for how many threads are actively executing
>>> tasks. There are more threads for doing the fetching (the details of which
>>> I'm not that familiar with) -- that never cuts into the number of actively
>>> executing tasks for "numCores". There isn't a 1-to-1 correspondence
>>> between one shuffle block and one task -- a shuffle-read task most likely
>>> needs to fetch many shuffle blocks, with some local and some remote (in
>>> most cases). So, from your lingo above, c is numCores, but c_1 is just an
>>> independent pool of threads.
>>>
>>> this obvious follow up question is, if you've actually more than
>>> "numCores" threads going at the same time, how come cpu usage is low? You
>>> could still have your cpus stuck waiting on i/o, from disk or network, so
>>> they aren't getting fully utilized. And the cpu can also be idle waiting
>>> for memory, if there are a lot of cache misses (I'm not sure how that will
>>> show up in cpu monitoring). If that were the case, that could even be from
>>> too many threads, as a lot of time is spent context switching ... but I'm
>>> just guessing now.
>>>
>>> hope this helps,
>>> Imran
>>>
>>>
>>>
>>> On Wed, Jun 10, 2015 at 1:41 PM, Mike Hynes <91...@gmail.com> wrote:
>>>
>>>> Hi Imran,
>>>>
>>>> Thank you again for your email.
>>>>
>>>> I just want to ask one further question to clarify the implementation
>>>> of the shuffle block fetches. When you say that rather than sitting
>>>> idle, [the executor] will immediately start reading the local block, I
>>>> would guess that, in implementation, the executor is going to launch
>>>> concurrent threads to read both local and remote blocks, which it
>>>> seems to do in the initialize() method of
>>>> core/.../storage/ShuffleBlockFetcherIterator.scala. Is that the case
>>>> or would the Executor run all local fetch threads first?
>>>>
>>>> The reason I ask is that if the slave machine on which the Executor is
>>>> running has some number of cores, c, then I would have thought that
>>>> some of the threads launched would occupy some number, c_1, of the
>>>> cores and conduct the local reads (where c_1 <= c). The other threads
>>>> would occupy the other (c - c_1) cores' cycles until *all* necessary
>>>> blocks have been read, and depending on c and the number of blocks to
>>>> fetch so that none of the cores are idle if there are many blocks to
>>>> fetch. (I monitor the CPU utilization of our nodes throughout a job,
>>>> and generally find them under-utilized statistically speaking; that
>>>> is, their usage over the whole job is lower than expected, with short
>>>> burst of high usage, so I ask this question in a specific way for this
>>>> reason, since I can see trends in the probability density functions of
>>>> CPU utilization as the #partitions of our RDDs are increased).
>>>>
>>>> ShuffleBlockFetcherIterator.scala:
>>>>
>>>> private[this] def initialize(): Unit = {
>>>> ...
>>>> // Send out initial requests for blocks, up to our maxBytesInFlight
>>>> while (fetchRequests.nonEmpty &&
>>>> (bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size
>>>> <= maxBytesInFlight)) {
>>>> sendRequest(fetchRequests.dequeue())
>>>> }
>>>> val numFetches = remoteRequests.size - fetchRequests.size
>>>> logInfo("Started " + numFetches + " remote fetches in" +
>>>> Utils.getUsedTimeMs(startTime))
>>>>
>>>> // Get Local Blocks
>>>> fetchLocalBlocks()
>>>> logDebug("Got local blocks in " + Utils.getUsedTimeMs(startTime))
>>>> }
>>>> private[this] def fetchLocalBlocks() {
>>>> val iter = localBlocks.iterator
>>>> while (iter.hasNext) {
>>>> val blockId = iter.next()
>>>> try {
>>>> val buf = blockManager.getBlockData(blockId)
>>>> shuffleMetrics.incLocalBlocksFetched(1)
>>>> shuffleMetrics.incLocalBytesRead(buf.size)
>>>> buf.retain()
>>>> results.put(new SuccessFetchResult(blockId, 0, buf))
>>>> } catch {
>>>> ...
>>>> }
>>>> }
>>>> }
>>>>
>>>> Obviously, I will have to sit down with core/.../network/nio/* and
>>>> core/.../shuffle/* and do my own homework on this, but from what I can
>>>> tell, the BlockDataManager relies on either
>>>> NioBlockTransferService.scala or the NettyBlockTransferService.scala
>>>> (which are set in SparkEnv.scala), both of which do the grunt work of
>>>> actually buffering and transferring the blocks' bytes. Finally, the
>>>> tasks in new stage for which the shuffle outputs have been fetched
>>>> will not commence until all of the block fetching threads (both local
>>>> and remote) have terminated.
>>>>
>>>> Does the above paint an accurate picture? I would really appreciate
>>>> clarification on the concurrency, since I would like to determine why
>>>> our jobs have under-utilization and poor weak scaling efficiency.
>>>>
>>>> I will cc this thread over to the dev list. I did not cc them in case
>>>> my previous question was trivial---I didn't want to spam the list
>>>> unnecessarily, since I do not see these kinds of questions posed there
>>>> frequently.
>>>>
>>>> Thanks a bunch,
>>>> Mike
>>>>
>>>>
>>>> On 6/10/15, Imran Rashid <ir...@cloudera.com> wrote:
>>>> > Hi Mike,
>>>> >
>>>> > no, this is a good question, I can see how my response could be
>>>> interpreted
>>>> > both ways.
>>>> >
>>>> > To be more precise:
>>>> > *nothing* is fetched until the shuffle-read stage starts. So it is
>>>> normal
>>>> > to see a spike in cluster bandwidth when that stage starts. There is
>>>> a
>>>> > hard-boundary between stages -- that is, spark never starts any tasks
>>>> in
>>>> > one stage until *all* tasks in the dependent stages have been
>>>> completed.
>>>> > (There has been on-and-off discussion about relaxing this, but IMO
>>>> this is
>>>> > unlikely to change in the near future.) So spark will wait for all
>>>> of the
>>>> > tasks in the previous shuffle-write stage to finish, and then kick
>>>> off a
>>>> > bunch of shuffle-read tasks in the next stage, leading to the spike
>>>> you
>>>> > see.
>>>> >
>>>> > I was referring to the way blocks are fetched within one of those
>>>> > shuffle-read tasks. One of those tasks is will probably going to
>>>> need a
>>>> > bunch of different blocks, from many executors. But some of the
>>>> blocks it
>>>> > needs will probably exist locally. So the task first sends out a
>>>> request
>>>> > to fetch blocks remotely (leading to the spike), but rather than
>>>> sitting
>>>> > idle, it will immediately start reading the local blocks. Ideally,
>>>> by the
>>>> > time its done reading the local blocks, some of the remote blocks have
>>>> > already been fetched, so no time is spent *waiting* for the remote
>>>> reads.
>>>> > As the remote blocks get read, spark sends out more requests, trying
>>>> to
>>>> > balance how much data needs to be buffered vs. preventing any waiting
>>>> on
>>>> > remote reads (which can be controlled by
>>>> spark.reducer.maxSizeInFlight).
>>>> >
>>>> > Hope that clarifies things!
>>>> >
>>>> > btw, you sent this last question to just me -- I think its a good
>>>> question,
>>>> > do you mind sending it to the list? I figured that was accidental but
>>>> > wanted to check.
>>>> >
>>>> > Imran
>>>> >
>>>> > On Wed, Jun 10, 2015 at 12:20 AM, Mike Hynes <91...@gmail.com>
>>>> wrote:
>>>> >
>>>> >> Hi Imran,
>>>> >> One additional quick question---I just want to confirm that I fully
>>>> >> understand your comment that "blocks are fetched before they are
>>>> >> needed." Typically on our system, we see spikes in cluster bandwidth
>>>> >> (with ganglia) at stage boundaries, so I previously assumed that all
>>>> >> shuffle read occurred there. Do you mean that the blocks are fetched
>>>> >> by the shuffle read iterator, and hence when tasks occur afterwards
>>>> >> the necessary blocks have already been fetched?
>>>> >> Thanks---I am sorry if this is an obvious question, but I'd like to
>>>> >> understand this as precisely as possible.
>>>> >> Mike
>>>> >>
>>>> >> On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>>>> >> > Ahhh---forgive my typo: what I mean is,
>>>> >> > (t2 - t1) >= (t_ser + t_deser + t_exec)
>>>> >> > is satisfied, empirically.
>>>> >> >
>>>> >> > On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>>>> >> >> 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 <ir...@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 <91...@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 <pw...@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 <
>>>> 91mbbh@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 <rx...@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 <
>>>> >> rtreffer@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-unsubscribe@spark.apache.org
>>>> >> >>>> >>> For additional commands, e-mail: dev-help@spark.apache.org
>>>> >> >>>> >>>
>>>> >> >>>> >>
>>>> >> >>>> >
>>>> >> >>>>
>>>> >> >>>>
>>>> >> >>>> --
>>>> >> >>>> Thanks,
>>>> >> >>>> Mike
>>>> >> >>>>
>>>> >> >>>>
>>>> >> >>>>
>>>> ---------------------------------------------------------------------
>>>> >> >>>> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
>>>> >> >>>> For additional commands, e-mail: dev-help@spark.apache.org
>>>> >> >>>>
>>>> >> >>>
>>>> >> >>
>>>> >> >>
>>>> >> >> --
>>>> >> >> Thanks,
>>>> >> >> Mike
>>>> >> >>
>>>> >> >
>>>> >> >
>>>> >> > --
>>>> >> > Thanks,
>>>> >> > Mike
>>>> >> >
>>>> >>
>>>> >>
>>>> >> --
>>>> >> Thanks,
>>>> >> Mike
>>>> >>
>>>> >
>>>>
>>>>
>>>> --
>>>> Thanks,
>>>> Mike
>>>>
>>>
>>>
>>
>
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Gerard Maas <ge...@gmail.com>.
Kay,
Excellent write-up. This should be preserved for reference somewhere
searchable.
-Gerard.
On Fri, Jun 12, 2015 at 1:19 AM, Kay Ousterhout <ke...@eecs.berkeley.edu>
wrote:
> Here’s how the shuffle works. This explains what happens for a single
> task; this will happen in parallel for each task running on the machine,
> and as Imran said, Spark runs up to “numCores” tasks concurrently on each
> machine. There's also an answer to the original question about why CPU use
> is low at the very bottom.
>
> The key data structure used in fetching shuffle data is the “results”
> queue in ShuffleBlockFetcherIterator, which buffers data that we have in
> serialized (and maybe compressed) form, but haven’t yet deserialized /
> processed. The results queue is filled by many threads fetching data over
> the network (the number of concurrent threads fetching data is equal to the
> number of remote executors we’re currently fetching data from) [0], and is
> consumed by a single thread that deserializes the data and computes some
> function over it (e.g., if you’re doing rdd.count(), the thread
> deserializes the data and counts the number of items). As we fetch data
> over the network, we track bytesInFlight, which is data that has been
> requested (and possibly received) from a remote executor, but that hasn’t
> yet been deserialized / processed by the consumer thread. So, this
> includes all of the data in the “results” queue, and possibly more data
> that’s currently outstanding over the network. We always issue as many
> requests as we can, with the constraint that bytesInFlight remains less
> than a specified maximum [1].
>
> In a little more detail, here’s exactly what happens when a task begins
> reading shuffled data:
>
> (1) Issue requests [1.5] to fetch up to maxBytesInFlight bytes of data [1]
> over the network (this happens here
> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L260>
> ).
>
> These requests are all executed asynchronously using a ShuffleClient [2]
> via the shuffleClient.fetchBlocks call
> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L149>
> [3]. We pass in a callback that, once a block has been successfully
> fetched, sticks it on the “results” queue.
>
> (2) Begin processing the local data. One by one, we request the local
> data from the local block manager (which memory maps the file) and then stick
> the result onto the results queue
> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L230>.
> Because we memory map the files, which is speedy, the local data typically
> all ends up on the results in front of the remote data.
>
> (3) One the async network requests have been issued (note — issued, but
> not finished!) and we’ve “read” (memory-mapped) the local data (i.e., (1)
> and (2) have happened), ShuffleBlockFetcherIterator returns an iterator
> that gets wrapped too many times to count [4] and eventually gets unrolled
> [5]. Each time next() is called on the iterator, it blocks waiting for an
> item from the results queue. This may return right away, or if the queue
> is empty, will block waiting on new data from the network [6]. Before
> returning from next(), we update our accounting for the bytes in flight:
> the chunk of data we return is no longer considered in-flight, because it’s
> about to be processed, so we update the current bytesInFlight, and if it
> won’t result in > maxBytesInFlight outstanding, send some more requests for
> data.
>
> ————————————————
>
> Notes:
>
> [0] Note that these threads consume almost no CPU resources, because they
> just receive data from the OS and then execute a callback that sticks the
> data on the results queue.
>
> [1] We limit the data outstanding on the network to avoid using too much
> memory to hold the data we’ve fetched over the network but haven’t yet
> processed.
>
> [1.5] Each request may include multiple shuffle blocks, where is a "block"
> is the data output for this reduce task by a particular map task. All of
> the reduce tasks for a shuffle read a total of # map tasks * # reduce tasks
> shuffle blocks; each reduce task reads # map tasks blocks. We do some
> hacks
> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L177>
> to try to size these requests in a "good" way: we limit each request to
> about maxBytesInFlight / 5, so that we can fetch from roughly 5 machines
> concurrently without exceeding maxBytesInFlight. 5 is completely a magic
> number here that was probably guessed by someone long long ago, and it
> seems to work ok.
>
> [2] The default configuration uses NettyBlockTransferService as the
> ShuffleClient implementation (note that this extends BlockTransferService,
> which extends ShuffleClient).
>
> [3] If you’re curious how the shuffle client fetches data, the default
> Spark configuration results in exactly one TCP connection from an executor
> to each other executor. If executor A is getting shuffle data from
> executor B, we start by sending an OpenBlocks message from A to B. The
> OpenBlocks message includes the list of blocks that A wants to fetch, and
> causes the remote executor, B, to start to pull the corresponding data into
> memory from disk (we typically memory map the files, so this may not
> actually result in the data being read yet), and also to store some state
> associated with this “stream” of data. The remote executor, B, responds
> with a stream ID that helps it to identify the connection. Next, A
> requests blocks one at a time from B using an ChunkFetchRequest message
> (this happens here
> <https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/network/shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java#L101> in
> OneForOneBlockFetcher, which calls this code
> <https://github.com/apache/spark/blob/master/network/common/src/main/java/org/apache/spark/network/client/TransportClient.java#L97>
> in TransportClient; currently, we have a one-to-one mapping from a chunk to
> a particular block). It’s possible that there are many sets of shuffle
> data being fetched concurrently between A and B (e.g., because many tasks
> are run concurrently). These requests are serialized, so one block is sent
> at a time from B, and they’re sent in the order that the requests were
> issued on A.
>
> [4] In BlockStoreShuffleFetcher, which handles failures; then in
> HashShuffleReader, which helps aggregate some of the data; etc.
>
> [5] This happens in BlockManager.putIterator, if the RDD is going to be
> cached; in the function passed in to ResultTask, if this is the last stage
> in a job; or via the writer.write() call in ShuffleMapTask, if this is a
> stage that generates intermediate shuffle data.
>
> [6] We time how long we spend blocking on data from the network; this is
> what’s shown as “fetch wait time” in Spark’s UI.
>
> ——————————————————
>
> To answer the original question about why CPU use is low, this means that
> the main thread (that pulls data off of the results queue) is blocking on
> I/O. It could be blocking to receive data over the network (which will be
> included in the fetch wait time shown in the UI, described above), or it
> could be blocking to read data from the local disk (because the shuffle
> data that is read locally is memory-mapped, so the thread may block on disk
> if the data hasn’t been read into memory yet). It could also be blocking
> because it’s writing new shuffle output to disk (also shown in the UI, as
> shuffle write time), or because it’s spilling intermediate data.
> Everything except the spilling is shown in the handy-dandy new
> visualization that Kousuke recently added
> <https://github.com/apache/spark/commit/a5f7b3b9c7f05598a1cc8e582e5facee1029cd5e>to
> the UI (available in 1.4 onwards): if you look at the stage view, and click
> on “Event Timeline”, you’ll see a visualization of the tasks and how long
> they spent blocked on various things (we should add the spilling to
> this…there is a long-outstanding JIRA for this).
>
> Hope this helps!
>
> -Kay
>
> On Thu, Jun 11, 2015 at 7:42 AM, Imran Rashid <ir...@cloudera.com>
> wrote:
>
>> That is not exactly correct -- that being said I'm not 100% on these
>> details either so I'd appreciate you double checking and / or another dev
>> confirming my description.
>>
>>
>> Spark actually has more threads going then the "numCores" you specify.
>> "numCores" is really used for how many threads are actively executing
>> tasks. There are more threads for doing the fetching (the details of which
>> I'm not that familiar with) -- that never cuts into the number of actively
>> executing tasks for "numCores". There isn't a 1-to-1 correspondence
>> between one shuffle block and one task -- a shuffle-read task most likely
>> needs to fetch many shuffle blocks, with some local and some remote (in
>> most cases). So, from your lingo above, c is numCores, but c_1 is just an
>> independent pool of threads.
>>
>> this obvious follow up question is, if you've actually more than
>> "numCores" threads going at the same time, how come cpu usage is low? You
>> could still have your cpus stuck waiting on i/o, from disk or network, so
>> they aren't getting fully utilized. And the cpu can also be idle waiting
>> for memory, if there are a lot of cache misses (I'm not sure how that will
>> show up in cpu monitoring). If that were the case, that could even be from
>> too many threads, as a lot of time is spent context switching ... but I'm
>> just guessing now.
>>
>> hope this helps,
>> Imran
>>
>>
>>
>> On Wed, Jun 10, 2015 at 1:41 PM, Mike Hynes <91...@gmail.com> wrote:
>>
>>> Hi Imran,
>>>
>>> Thank you again for your email.
>>>
>>> I just want to ask one further question to clarify the implementation
>>> of the shuffle block fetches. When you say that rather than sitting
>>> idle, [the executor] will immediately start reading the local block, I
>>> would guess that, in implementation, the executor is going to launch
>>> concurrent threads to read both local and remote blocks, which it
>>> seems to do in the initialize() method of
>>> core/.../storage/ShuffleBlockFetcherIterator.scala. Is that the case
>>> or would the Executor run all local fetch threads first?
>>>
>>> The reason I ask is that if the slave machine on which the Executor is
>>> running has some number of cores, c, then I would have thought that
>>> some of the threads launched would occupy some number, c_1, of the
>>> cores and conduct the local reads (where c_1 <= c). The other threads
>>> would occupy the other (c - c_1) cores' cycles until *all* necessary
>>> blocks have been read, and depending on c and the number of blocks to
>>> fetch so that none of the cores are idle if there are many blocks to
>>> fetch. (I monitor the CPU utilization of our nodes throughout a job,
>>> and generally find them under-utilized statistically speaking; that
>>> is, their usage over the whole job is lower than expected, with short
>>> burst of high usage, so I ask this question in a specific way for this
>>> reason, since I can see trends in the probability density functions of
>>> CPU utilization as the #partitions of our RDDs are increased).
>>>
>>> ShuffleBlockFetcherIterator.scala:
>>>
>>> private[this] def initialize(): Unit = {
>>> ...
>>> // Send out initial requests for blocks, up to our maxBytesInFlight
>>> while (fetchRequests.nonEmpty &&
>>> (bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size
>>> <= maxBytesInFlight)) {
>>> sendRequest(fetchRequests.dequeue())
>>> }
>>> val numFetches = remoteRequests.size - fetchRequests.size
>>> logInfo("Started " + numFetches + " remote fetches in" +
>>> Utils.getUsedTimeMs(startTime))
>>>
>>> // Get Local Blocks
>>> fetchLocalBlocks()
>>> logDebug("Got local blocks in " + Utils.getUsedTimeMs(startTime))
>>> }
>>> private[this] def fetchLocalBlocks() {
>>> val iter = localBlocks.iterator
>>> while (iter.hasNext) {
>>> val blockId = iter.next()
>>> try {
>>> val buf = blockManager.getBlockData(blockId)
>>> shuffleMetrics.incLocalBlocksFetched(1)
>>> shuffleMetrics.incLocalBytesRead(buf.size)
>>> buf.retain()
>>> results.put(new SuccessFetchResult(blockId, 0, buf))
>>> } catch {
>>> ...
>>> }
>>> }
>>> }
>>>
>>> Obviously, I will have to sit down with core/.../network/nio/* and
>>> core/.../shuffle/* and do my own homework on this, but from what I can
>>> tell, the BlockDataManager relies on either
>>> NioBlockTransferService.scala or the NettyBlockTransferService.scala
>>> (which are set in SparkEnv.scala), both of which do the grunt work of
>>> actually buffering and transferring the blocks' bytes. Finally, the
>>> tasks in new stage for which the shuffle outputs have been fetched
>>> will not commence until all of the block fetching threads (both local
>>> and remote) have terminated.
>>>
>>> Does the above paint an accurate picture? I would really appreciate
>>> clarification on the concurrency, since I would like to determine why
>>> our jobs have under-utilization and poor weak scaling efficiency.
>>>
>>> I will cc this thread over to the dev list. I did not cc them in case
>>> my previous question was trivial---I didn't want to spam the list
>>> unnecessarily, since I do not see these kinds of questions posed there
>>> frequently.
>>>
>>> Thanks a bunch,
>>> Mike
>>>
>>>
>>> On 6/10/15, Imran Rashid <ir...@cloudera.com> wrote:
>>> > Hi Mike,
>>> >
>>> > no, this is a good question, I can see how my response could be
>>> interpreted
>>> > both ways.
>>> >
>>> > To be more precise:
>>> > *nothing* is fetched until the shuffle-read stage starts. So it is
>>> normal
>>> > to see a spike in cluster bandwidth when that stage starts. There is a
>>> > hard-boundary between stages -- that is, spark never starts any tasks
>>> in
>>> > one stage until *all* tasks in the dependent stages have been
>>> completed.
>>> > (There has been on-and-off discussion about relaxing this, but IMO
>>> this is
>>> > unlikely to change in the near future.) So spark will wait for all of
>>> the
>>> > tasks in the previous shuffle-write stage to finish, and then kick off
>>> a
>>> > bunch of shuffle-read tasks in the next stage, leading to the spike you
>>> > see.
>>> >
>>> > I was referring to the way blocks are fetched within one of those
>>> > shuffle-read tasks. One of those tasks is will probably going to need
>>> a
>>> > bunch of different blocks, from many executors. But some of the
>>> blocks it
>>> > needs will probably exist locally. So the task first sends out a
>>> request
>>> > to fetch blocks remotely (leading to the spike), but rather than
>>> sitting
>>> > idle, it will immediately start reading the local blocks. Ideally, by
>>> the
>>> > time its done reading the local blocks, some of the remote blocks have
>>> > already been fetched, so no time is spent *waiting* for the remote
>>> reads.
>>> > As the remote blocks get read, spark sends out more requests, trying to
>>> > balance how much data needs to be buffered vs. preventing any waiting
>>> on
>>> > remote reads (which can be controlled by
>>> spark.reducer.maxSizeInFlight).
>>> >
>>> > Hope that clarifies things!
>>> >
>>> > btw, you sent this last question to just me -- I think its a good
>>> question,
>>> > do you mind sending it to the list? I figured that was accidental but
>>> > wanted to check.
>>> >
>>> > Imran
>>> >
>>> > On Wed, Jun 10, 2015 at 12:20 AM, Mike Hynes <91...@gmail.com> wrote:
>>> >
>>> >> Hi Imran,
>>> >> One additional quick question---I just want to confirm that I fully
>>> >> understand your comment that "blocks are fetched before they are
>>> >> needed." Typically on our system, we see spikes in cluster bandwidth
>>> >> (with ganglia) at stage boundaries, so I previously assumed that all
>>> >> shuffle read occurred there. Do you mean that the blocks are fetched
>>> >> by the shuffle read iterator, and hence when tasks occur afterwards
>>> >> the necessary blocks have already been fetched?
>>> >> Thanks---I am sorry if this is an obvious question, but I'd like to
>>> >> understand this as precisely as possible.
>>> >> Mike
>>> >>
>>> >> On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>>> >> > Ahhh---forgive my typo: what I mean is,
>>> >> > (t2 - t1) >= (t_ser + t_deser + t_exec)
>>> >> > is satisfied, empirically.
>>> >> >
>>> >> > On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>>> >> >> 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 <ir...@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 <91...@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 <pw...@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 <91mbbh@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 <rx...@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 <
>>> >> rtreffer@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-unsubscribe@spark.apache.org
>>> >> >>>> >>> For additional commands, e-mail: dev-help@spark.apache.org
>>> >> >>>> >>>
>>> >> >>>> >>
>>> >> >>>> >
>>> >> >>>>
>>> >> >>>>
>>> >> >>>> --
>>> >> >>>> Thanks,
>>> >> >>>> Mike
>>> >> >>>>
>>> >> >>>>
>>> >> >>>>
>>> ---------------------------------------------------------------------
>>> >> >>>> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
>>> >> >>>> For additional commands, e-mail: dev-help@spark.apache.org
>>> >> >>>>
>>> >> >>>
>>> >> >>
>>> >> >>
>>> >> >> --
>>> >> >> Thanks,
>>> >> >> Mike
>>> >> >>
>>> >> >
>>> >> >
>>> >> > --
>>> >> > Thanks,
>>> >> > Mike
>>> >> >
>>> >>
>>> >>
>>> >> --
>>> >> Thanks,
>>> >> Mike
>>> >>
>>> >
>>>
>>>
>>> --
>>> Thanks,
>>> Mike
>>>
>>
>>
>
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Kay Ousterhout <ke...@eecs.berkeley.edu>.
Here’s how the shuffle works. This explains what happens for a single
task; this will happen in parallel for each task running on the machine,
and as Imran said, Spark runs up to “numCores” tasks concurrently on each
machine. There's also an answer to the original question about why CPU use
is low at the very bottom.
The key data structure used in fetching shuffle data is the “results” queue
in ShuffleBlockFetcherIterator, which buffers data that we have in
serialized (and maybe compressed) form, but haven’t yet deserialized /
processed. The results queue is filled by many threads fetching data over
the network (the number of concurrent threads fetching data is equal to the
number of remote executors we’re currently fetching data from) [0], and is
consumed by a single thread that deserializes the data and computes some
function over it (e.g., if you’re doing rdd.count(), the thread
deserializes the data and counts the number of items). As we fetch data
over the network, we track bytesInFlight, which is data that has been
requested (and possibly received) from a remote executor, but that hasn’t
yet been deserialized / processed by the consumer thread. So, this
includes all of the data in the “results” queue, and possibly more data
that’s currently outstanding over the network. We always issue as many
requests as we can, with the constraint that bytesInFlight remains less
than a specified maximum [1].
In a little more detail, here’s exactly what happens when a task begins
reading shuffled data:
(1) Issue requests [1.5] to fetch up to maxBytesInFlight bytes of data [1]
over the network (this happens here
<https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L260>
).
These requests are all executed asynchronously using a ShuffleClient [2]
via the shuffleClient.fetchBlocks call
<https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L149>
[3]. We pass in a callback that, once a block has been successfully
fetched, sticks it on the “results” queue.
(2) Begin processing the local data. One by one, we request the local data
from the local block manager (which memory maps the file) and then stick
the result onto the results queue
<https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L230>.
Because we memory map the files, which is speedy, the local data typically
all ends up on the results in front of the remote data.
(3) One the async network requests have been issued (note — issued, but not
finished!) and we’ve “read” (memory-mapped) the local data (i.e., (1) and
(2) have happened), ShuffleBlockFetcherIterator returns an iterator that
gets wrapped too many times to count [4] and eventually gets unrolled [5].
Each time next() is called on the iterator, it blocks waiting for an item
from the results queue. This may return right away, or if the queue is
empty, will block waiting on new data from the network [6]. Before
returning from next(), we update our accounting for the bytes in flight:
the chunk of data we return is no longer considered in-flight, because it’s
about to be processed, so we update the current bytesInFlight, and if it
won’t result in > maxBytesInFlight outstanding, send some more requests for
data.
————————————————
Notes:
[0] Note that these threads consume almost no CPU resources, because they
just receive data from the OS and then execute a callback that sticks the
data on the results queue.
[1] We limit the data outstanding on the network to avoid using too much
memory to hold the data we’ve fetched over the network but haven’t yet
processed.
[1.5] Each request may include multiple shuffle blocks, where is a "block"
is the data output for this reduce task by a particular map task. All of
the reduce tasks for a shuffle read a total of # map tasks * # reduce tasks
shuffle blocks; each reduce task reads # map tasks blocks. We do some hacks
<https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L177>
to try to size these requests in a "good" way: we limit each request to
about maxBytesInFlight / 5, so that we can fetch from roughly 5 machines
concurrently without exceeding maxBytesInFlight. 5 is completely a magic
number here that was probably guessed by someone long long ago, and it
seems to work ok.
[2] The default configuration uses NettyBlockTransferService as the
ShuffleClient implementation (note that this extends BlockTransferService,
which extends ShuffleClient).
[3] If you’re curious how the shuffle client fetches data, the default
Spark configuration results in exactly one TCP connection from an executor
to each other executor. If executor A is getting shuffle data from
executor B, we start by sending an OpenBlocks message from A to B. The
OpenBlocks message includes the list of blocks that A wants to fetch, and
causes the remote executor, B, to start to pull the corresponding data into
memory from disk (we typically memory map the files, so this may not
actually result in the data being read yet), and also to store some state
associated with this “stream” of data. The remote executor, B, responds
with a stream ID that helps it to identify the connection. Next, A
requests blocks one at a time from B using an ChunkFetchRequest message
(this happens here
<https://github.com/apache/spark/blob/95690a17d328f205c3398b9b477b4072b6fe908f/network/shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java#L101>
in
OneForOneBlockFetcher, which calls this code
<https://github.com/apache/spark/blob/master/network/common/src/main/java/org/apache/spark/network/client/TransportClient.java#L97>
in TransportClient; currently, we have a one-to-one mapping from a chunk to
a particular block). It’s possible that there are many sets of shuffle
data being fetched concurrently between A and B (e.g., because many tasks
are run concurrently). These requests are serialized, so one block is sent
at a time from B, and they’re sent in the order that the requests were
issued on A.
[4] In BlockStoreShuffleFetcher, which handles failures; then in
HashShuffleReader, which helps aggregate some of the data; etc.
[5] This happens in BlockManager.putIterator, if the RDD is going to be
cached; in the function passed in to ResultTask, if this is the last stage
in a job; or via the writer.write() call in ShuffleMapTask, if this is a
stage that generates intermediate shuffle data.
[6] We time how long we spend blocking on data from the network; this is
what’s shown as “fetch wait time” in Spark’s UI.
——————————————————
To answer the original question about why CPU use is low, this means that
the main thread (that pulls data off of the results queue) is blocking on
I/O. It could be blocking to receive data over the network (which will be
included in the fetch wait time shown in the UI, described above), or it
could be blocking to read data from the local disk (because the shuffle
data that is read locally is memory-mapped, so the thread may block on disk
if the data hasn’t been read into memory yet). It could also be blocking
because it’s writing new shuffle output to disk (also shown in the UI, as
shuffle write time), or because it’s spilling intermediate data.
Everything except the spilling is shown in the handy-dandy new
visualization that Kousuke recently added
<https://github.com/apache/spark/commit/a5f7b3b9c7f05598a1cc8e582e5facee1029cd5e>to
the UI (available in 1.4 onwards): if you look at the stage view, and click
on “Event Timeline”, you’ll see a visualization of the tasks and how long
they spent blocked on various things (we should add the spilling to
this…there is a long-outstanding JIRA for this).
Hope this helps!
-Kay
On Thu, Jun 11, 2015 at 7:42 AM, Imran Rashid <ir...@cloudera.com> wrote:
> That is not exactly correct -- that being said I'm not 100% on these
> details either so I'd appreciate you double checking and / or another dev
> confirming my description.
>
>
> Spark actually has more threads going then the "numCores" you specify.
> "numCores" is really used for how many threads are actively executing
> tasks. There are more threads for doing the fetching (the details of which
> I'm not that familiar with) -- that never cuts into the number of actively
> executing tasks for "numCores". There isn't a 1-to-1 correspondence
> between one shuffle block and one task -- a shuffle-read task most likely
> needs to fetch many shuffle blocks, with some local and some remote (in
> most cases). So, from your lingo above, c is numCores, but c_1 is just an
> independent pool of threads.
>
> this obvious follow up question is, if you've actually more than
> "numCores" threads going at the same time, how come cpu usage is low? You
> could still have your cpus stuck waiting on i/o, from disk or network, so
> they aren't getting fully utilized. And the cpu can also be idle waiting
> for memory, if there are a lot of cache misses (I'm not sure how that will
> show up in cpu monitoring). If that were the case, that could even be from
> too many threads, as a lot of time is spent context switching ... but I'm
> just guessing now.
>
> hope this helps,
> Imran
>
>
>
> On Wed, Jun 10, 2015 at 1:41 PM, Mike Hynes <91...@gmail.com> wrote:
>
>> Hi Imran,
>>
>> Thank you again for your email.
>>
>> I just want to ask one further question to clarify the implementation
>> of the shuffle block fetches. When you say that rather than sitting
>> idle, [the executor] will immediately start reading the local block, I
>> would guess that, in implementation, the executor is going to launch
>> concurrent threads to read both local and remote blocks, which it
>> seems to do in the initialize() method of
>> core/.../storage/ShuffleBlockFetcherIterator.scala. Is that the case
>> or would the Executor run all local fetch threads first?
>>
>> The reason I ask is that if the slave machine on which the Executor is
>> running has some number of cores, c, then I would have thought that
>> some of the threads launched would occupy some number, c_1, of the
>> cores and conduct the local reads (where c_1 <= c). The other threads
>> would occupy the other (c - c_1) cores' cycles until *all* necessary
>> blocks have been read, and depending on c and the number of blocks to
>> fetch so that none of the cores are idle if there are many blocks to
>> fetch. (I monitor the CPU utilization of our nodes throughout a job,
>> and generally find them under-utilized statistically speaking; that
>> is, their usage over the whole job is lower than expected, with short
>> burst of high usage, so I ask this question in a specific way for this
>> reason, since I can see trends in the probability density functions of
>> CPU utilization as the #partitions of our RDDs are increased).
>>
>> ShuffleBlockFetcherIterator.scala:
>>
>> private[this] def initialize(): Unit = {
>> ...
>> // Send out initial requests for blocks, up to our maxBytesInFlight
>> while (fetchRequests.nonEmpty &&
>> (bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size
>> <= maxBytesInFlight)) {
>> sendRequest(fetchRequests.dequeue())
>> }
>> val numFetches = remoteRequests.size - fetchRequests.size
>> logInfo("Started " + numFetches + " remote fetches in" +
>> Utils.getUsedTimeMs(startTime))
>>
>> // Get Local Blocks
>> fetchLocalBlocks()
>> logDebug("Got local blocks in " + Utils.getUsedTimeMs(startTime))
>> }
>> private[this] def fetchLocalBlocks() {
>> val iter = localBlocks.iterator
>> while (iter.hasNext) {
>> val blockId = iter.next()
>> try {
>> val buf = blockManager.getBlockData(blockId)
>> shuffleMetrics.incLocalBlocksFetched(1)
>> shuffleMetrics.incLocalBytesRead(buf.size)
>> buf.retain()
>> results.put(new SuccessFetchResult(blockId, 0, buf))
>> } catch {
>> ...
>> }
>> }
>> }
>>
>> Obviously, I will have to sit down with core/.../network/nio/* and
>> core/.../shuffle/* and do my own homework on this, but from what I can
>> tell, the BlockDataManager relies on either
>> NioBlockTransferService.scala or the NettyBlockTransferService.scala
>> (which are set in SparkEnv.scala), both of which do the grunt work of
>> actually buffering and transferring the blocks' bytes. Finally, the
>> tasks in new stage for which the shuffle outputs have been fetched
>> will not commence until all of the block fetching threads (both local
>> and remote) have terminated.
>>
>> Does the above paint an accurate picture? I would really appreciate
>> clarification on the concurrency, since I would like to determine why
>> our jobs have under-utilization and poor weak scaling efficiency.
>>
>> I will cc this thread over to the dev list. I did not cc them in case
>> my previous question was trivial---I didn't want to spam the list
>> unnecessarily, since I do not see these kinds of questions posed there
>> frequently.
>>
>> Thanks a bunch,
>> Mike
>>
>>
>> On 6/10/15, Imran Rashid <ir...@cloudera.com> wrote:
>> > Hi Mike,
>> >
>> > no, this is a good question, I can see how my response could be
>> interpreted
>> > both ways.
>> >
>> > To be more precise:
>> > *nothing* is fetched until the shuffle-read stage starts. So it is
>> normal
>> > to see a spike in cluster bandwidth when that stage starts. There is a
>> > hard-boundary between stages -- that is, spark never starts any tasks in
>> > one stage until *all* tasks in the dependent stages have been completed.
>> > (There has been on-and-off discussion about relaxing this, but IMO this
>> is
>> > unlikely to change in the near future.) So spark will wait for all of
>> the
>> > tasks in the previous shuffle-write stage to finish, and then kick off a
>> > bunch of shuffle-read tasks in the next stage, leading to the spike you
>> > see.
>> >
>> > I was referring to the way blocks are fetched within one of those
>> > shuffle-read tasks. One of those tasks is will probably going to need a
>> > bunch of different blocks, from many executors. But some of the blocks
>> it
>> > needs will probably exist locally. So the task first sends out a
>> request
>> > to fetch blocks remotely (leading to the spike), but rather than sitting
>> > idle, it will immediately start reading the local blocks. Ideally, by
>> the
>> > time its done reading the local blocks, some of the remote blocks have
>> > already been fetched, so no time is spent *waiting* for the remote
>> reads.
>> > As the remote blocks get read, spark sends out more requests, trying to
>> > balance how much data needs to be buffered vs. preventing any waiting on
>> > remote reads (which can be controlled by
>> spark.reducer.maxSizeInFlight).
>> >
>> > Hope that clarifies things!
>> >
>> > btw, you sent this last question to just me -- I think its a good
>> question,
>> > do you mind sending it to the list? I figured that was accidental but
>> > wanted to check.
>> >
>> > Imran
>> >
>> > On Wed, Jun 10, 2015 at 12:20 AM, Mike Hynes <91...@gmail.com> wrote:
>> >
>> >> Hi Imran,
>> >> One additional quick question---I just want to confirm that I fully
>> >> understand your comment that "blocks are fetched before they are
>> >> needed." Typically on our system, we see spikes in cluster bandwidth
>> >> (with ganglia) at stage boundaries, so I previously assumed that all
>> >> shuffle read occurred there. Do you mean that the blocks are fetched
>> >> by the shuffle read iterator, and hence when tasks occur afterwards
>> >> the necessary blocks have already been fetched?
>> >> Thanks---I am sorry if this is an obvious question, but I'd like to
>> >> understand this as precisely as possible.
>> >> Mike
>> >>
>> >> On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>> >> > Ahhh---forgive my typo: what I mean is,
>> >> > (t2 - t1) >= (t_ser + t_deser + t_exec)
>> >> > is satisfied, empirically.
>> >> >
>> >> > On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>> >> >> 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 <ir...@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 <91...@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 <pw...@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 <91...@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 <rx...@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 <
>> >> rtreffer@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-unsubscribe@spark.apache.org
>> >> >>>> >>> For additional commands, e-mail: dev-help@spark.apache.org
>> >> >>>> >>>
>> >> >>>> >>
>> >> >>>> >
>> >> >>>>
>> >> >>>>
>> >> >>>> --
>> >> >>>> Thanks,
>> >> >>>> Mike
>> >> >>>>
>> >> >>>>
>> >> >>>>
>> ---------------------------------------------------------------------
>> >> >>>> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
>> >> >>>> For additional commands, e-mail: dev-help@spark.apache.org
>> >> >>>>
>> >> >>>
>> >> >>
>> >> >>
>> >> >> --
>> >> >> Thanks,
>> >> >> Mike
>> >> >>
>> >> >
>> >> >
>> >> > --
>> >> > Thanks,
>> >> > Mike
>> >> >
>> >>
>> >>
>> >> --
>> >> Thanks,
>> >> Mike
>> >>
>> >
>>
>>
>> --
>> Thanks,
>> Mike
>>
>
>
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Imran Rashid <ir...@cloudera.com>.
That is not exactly correct -- that being said I'm not 100% on these
details either so I'd appreciate you double checking and / or another dev
confirming my description.
Spark actually has more threads going then the "numCores" you specify.
"numCores" is really used for how many threads are actively executing
tasks. There are more threads for doing the fetching (the details of which
I'm not that familiar with) -- that never cuts into the number of actively
executing tasks for "numCores". There isn't a 1-to-1 correspondence
between one shuffle block and one task -- a shuffle-read task most likely
needs to fetch many shuffle blocks, with some local and some remote (in
most cases). So, from your lingo above, c is numCores, but c_1 is just an
independent pool of threads.
this obvious follow up question is, if you've actually more than "numCores"
threads going at the same time, how come cpu usage is low? You could still
have your cpus stuck waiting on i/o, from disk or network, so they aren't
getting fully utilized. And the cpu can also be idle waiting for memory,
if there are a lot of cache misses (I'm not sure how that will show up in
cpu monitoring). If that were the case, that could even be from too many
threads, as a lot of time is spent context switching ... but I'm just
guessing now.
hope this helps,
Imran
On Wed, Jun 10, 2015 at 1:41 PM, Mike Hynes <91...@gmail.com> wrote:
> Hi Imran,
>
> Thank you again for your email.
>
> I just want to ask one further question to clarify the implementation
> of the shuffle block fetches. When you say that rather than sitting
> idle, [the executor] will immediately start reading the local block, I
> would guess that, in implementation, the executor is going to launch
> concurrent threads to read both local and remote blocks, which it
> seems to do in the initialize() method of
> core/.../storage/ShuffleBlockFetcherIterator.scala. Is that the case
> or would the Executor run all local fetch threads first?
>
> The reason I ask is that if the slave machine on which the Executor is
> running has some number of cores, c, then I would have thought that
> some of the threads launched would occupy some number, c_1, of the
> cores and conduct the local reads (where c_1 <= c). The other threads
> would occupy the other (c - c_1) cores' cycles until *all* necessary
> blocks have been read, and depending on c and the number of blocks to
> fetch so that none of the cores are idle if there are many blocks to
> fetch. (I monitor the CPU utilization of our nodes throughout a job,
> and generally find them under-utilized statistically speaking; that
> is, their usage over the whole job is lower than expected, with short
> burst of high usage, so I ask this question in a specific way for this
> reason, since I can see trends in the probability density functions of
> CPU utilization as the #partitions of our RDDs are increased).
>
> ShuffleBlockFetcherIterator.scala:
>
> private[this] def initialize(): Unit = {
> ...
> // Send out initial requests for blocks, up to our maxBytesInFlight
> while (fetchRequests.nonEmpty &&
> (bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size
> <= maxBytesInFlight)) {
> sendRequest(fetchRequests.dequeue())
> }
> val numFetches = remoteRequests.size - fetchRequests.size
> logInfo("Started " + numFetches + " remote fetches in" +
> Utils.getUsedTimeMs(startTime))
>
> // Get Local Blocks
> fetchLocalBlocks()
> logDebug("Got local blocks in " + Utils.getUsedTimeMs(startTime))
> }
> private[this] def fetchLocalBlocks() {
> val iter = localBlocks.iterator
> while (iter.hasNext) {
> val blockId = iter.next()
> try {
> val buf = blockManager.getBlockData(blockId)
> shuffleMetrics.incLocalBlocksFetched(1)
> shuffleMetrics.incLocalBytesRead(buf.size)
> buf.retain()
> results.put(new SuccessFetchResult(blockId, 0, buf))
> } catch {
> ...
> }
> }
> }
>
> Obviously, I will have to sit down with core/.../network/nio/* and
> core/.../shuffle/* and do my own homework on this, but from what I can
> tell, the BlockDataManager relies on either
> NioBlockTransferService.scala or the NettyBlockTransferService.scala
> (which are set in SparkEnv.scala), both of which do the grunt work of
> actually buffering and transferring the blocks' bytes. Finally, the
> tasks in new stage for which the shuffle outputs have been fetched
> will not commence until all of the block fetching threads (both local
> and remote) have terminated.
>
> Does the above paint an accurate picture? I would really appreciate
> clarification on the concurrency, since I would like to determine why
> our jobs have under-utilization and poor weak scaling efficiency.
>
> I will cc this thread over to the dev list. I did not cc them in case
> my previous question was trivial---I didn't want to spam the list
> unnecessarily, since I do not see these kinds of questions posed there
> frequently.
>
> Thanks a bunch,
> Mike
>
>
> On 6/10/15, Imran Rashid <ir...@cloudera.com> wrote:
> > Hi Mike,
> >
> > no, this is a good question, I can see how my response could be
> interpreted
> > both ways.
> >
> > To be more precise:
> > *nothing* is fetched until the shuffle-read stage starts. So it is
> normal
> > to see a spike in cluster bandwidth when that stage starts. There is a
> > hard-boundary between stages -- that is, spark never starts any tasks in
> > one stage until *all* tasks in the dependent stages have been completed.
> > (There has been on-and-off discussion about relaxing this, but IMO this
> is
> > unlikely to change in the near future.) So spark will wait for all of
> the
> > tasks in the previous shuffle-write stage to finish, and then kick off a
> > bunch of shuffle-read tasks in the next stage, leading to the spike you
> > see.
> >
> > I was referring to the way blocks are fetched within one of those
> > shuffle-read tasks. One of those tasks is will probably going to need a
> > bunch of different blocks, from many executors. But some of the blocks
> it
> > needs will probably exist locally. So the task first sends out a request
> > to fetch blocks remotely (leading to the spike), but rather than sitting
> > idle, it will immediately start reading the local blocks. Ideally, by
> the
> > time its done reading the local blocks, some of the remote blocks have
> > already been fetched, so no time is spent *waiting* for the remote reads.
> > As the remote blocks get read, spark sends out more requests, trying to
> > balance how much data needs to be buffered vs. preventing any waiting on
> > remote reads (which can be controlled by spark.reducer.maxSizeInFlight).
> >
> > Hope that clarifies things!
> >
> > btw, you sent this last question to just me -- I think its a good
> question,
> > do you mind sending it to the list? I figured that was accidental but
> > wanted to check.
> >
> > Imran
> >
> > On Wed, Jun 10, 2015 at 12:20 AM, Mike Hynes <91...@gmail.com> wrote:
> >
> >> Hi Imran,
> >> One additional quick question---I just want to confirm that I fully
> >> understand your comment that "blocks are fetched before they are
> >> needed." Typically on our system, we see spikes in cluster bandwidth
> >> (with ganglia) at stage boundaries, so I previously assumed that all
> >> shuffle read occurred there. Do you mean that the blocks are fetched
> >> by the shuffle read iterator, and hence when tasks occur afterwards
> >> the necessary blocks have already been fetched?
> >> Thanks---I am sorry if this is an obvious question, but I'd like to
> >> understand this as precisely as possible.
> >> Mike
> >>
> >> On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
> >> > Ahhh---forgive my typo: what I mean is,
> >> > (t2 - t1) >= (t_ser + t_deser + t_exec)
> >> > is satisfied, empirically.
> >> >
> >> > On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
> >> >> 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 <ir...@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 <91...@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 <pw...@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 <91...@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 <rx...@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 <
> >> rtreffer@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-unsubscribe@spark.apache.org
> >> >>>> >>> For additional commands, e-mail: dev-help@spark.apache.org
> >> >>>> >>>
> >> >>>> >>
> >> >>>> >
> >> >>>>
> >> >>>>
> >> >>>> --
> >> >>>> Thanks,
> >> >>>> Mike
> >> >>>>
> >> >>>>
> >> >>>>
> ---------------------------------------------------------------------
> >> >>>> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
> >> >>>> For additional commands, e-mail: dev-help@spark.apache.org
> >> >>>>
> >> >>>
> >> >>
> >> >>
> >> >> --
> >> >> Thanks,
> >> >> Mike
> >> >>
> >> >
> >> >
> >> > --
> >> > Thanks,
> >> > Mike
> >> >
> >>
> >>
> >> --
> >> Thanks,
> >> Mike
> >>
> >
>
>
> --
> Thanks,
> Mike
>
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Mike Hynes <91...@gmail.com>.
Hi Imran,
Thank you again for your email.
I just want to ask one further question to clarify the implementation
of the shuffle block fetches. When you say that rather than sitting
idle, [the executor] will immediately start reading the local block, I
would guess that, in implementation, the executor is going to launch
concurrent threads to read both local and remote blocks, which it
seems to do in the initialize() method of
core/.../storage/ShuffleBlockFetcherIterator.scala. Is that the case
or would the Executor run all local fetch threads first?
The reason I ask is that if the slave machine on which the Executor is
running has some number of cores, c, then I would have thought that
some of the threads launched would occupy some number, c_1, of the
cores and conduct the local reads (where c_1 <= c). The other threads
would occupy the other (c - c_1) cores' cycles until *all* necessary
blocks have been read, and depending on c and the number of blocks to
fetch so that none of the cores are idle if there are many blocks to
fetch. (I monitor the CPU utilization of our nodes throughout a job,
and generally find them under-utilized statistically speaking; that
is, their usage over the whole job is lower than expected, with short
burst of high usage, so I ask this question in a specific way for this
reason, since I can see trends in the probability density functions of
CPU utilization as the #partitions of our RDDs are increased).
ShuffleBlockFetcherIterator.scala:
private[this] def initialize(): Unit = {
...
// Send out initial requests for blocks, up to our maxBytesInFlight
while (fetchRequests.nonEmpty &&
(bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size
<= maxBytesInFlight)) {
sendRequest(fetchRequests.dequeue())
}
val numFetches = remoteRequests.size - fetchRequests.size
logInfo("Started " + numFetches + " remote fetches in" +
Utils.getUsedTimeMs(startTime))
// Get Local Blocks
fetchLocalBlocks()
logDebug("Got local blocks in " + Utils.getUsedTimeMs(startTime))
}
private[this] def fetchLocalBlocks() {
val iter = localBlocks.iterator
while (iter.hasNext) {
val blockId = iter.next()
try {
val buf = blockManager.getBlockData(blockId)
shuffleMetrics.incLocalBlocksFetched(1)
shuffleMetrics.incLocalBytesRead(buf.size)
buf.retain()
results.put(new SuccessFetchResult(blockId, 0, buf))
} catch {
...
}
}
}
Obviously, I will have to sit down with core/.../network/nio/* and
core/.../shuffle/* and do my own homework on this, but from what I can
tell, the BlockDataManager relies on either
NioBlockTransferService.scala or the NettyBlockTransferService.scala
(which are set in SparkEnv.scala), both of which do the grunt work of
actually buffering and transferring the blocks' bytes. Finally, the
tasks in new stage for which the shuffle outputs have been fetched
will not commence until all of the block fetching threads (both local
and remote) have terminated.
Does the above paint an accurate picture? I would really appreciate
clarification on the concurrency, since I would like to determine why
our jobs have under-utilization and poor weak scaling efficiency.
I will cc this thread over to the dev list. I did not cc them in case
my previous question was trivial---I didn't want to spam the list
unnecessarily, since I do not see these kinds of questions posed there
frequently.
Thanks a bunch,
Mike
On 6/10/15, Imran Rashid <ir...@cloudera.com> wrote:
> Hi Mike,
>
> no, this is a good question, I can see how my response could be interpreted
> both ways.
>
> To be more precise:
> *nothing* is fetched until the shuffle-read stage starts. So it is normal
> to see a spike in cluster bandwidth when that stage starts. There is a
> hard-boundary between stages -- that is, spark never starts any tasks in
> one stage until *all* tasks in the dependent stages have been completed.
> (There has been on-and-off discussion about relaxing this, but IMO this is
> unlikely to change in the near future.) So spark will wait for all of the
> tasks in the previous shuffle-write stage to finish, and then kick off a
> bunch of shuffle-read tasks in the next stage, leading to the spike you
> see.
>
> I was referring to the way blocks are fetched within one of those
> shuffle-read tasks. One of those tasks is will probably going to need a
> bunch of different blocks, from many executors. But some of the blocks it
> needs will probably exist locally. So the task first sends out a request
> to fetch blocks remotely (leading to the spike), but rather than sitting
> idle, it will immediately start reading the local blocks. Ideally, by the
> time its done reading the local blocks, some of the remote blocks have
> already been fetched, so no time is spent *waiting* for the remote reads.
> As the remote blocks get read, spark sends out more requests, trying to
> balance how much data needs to be buffered vs. preventing any waiting on
> remote reads (which can be controlled by spark.reducer.maxSizeInFlight).
>
> Hope that clarifies things!
>
> btw, you sent this last question to just me -- I think its a good question,
> do you mind sending it to the list? I figured that was accidental but
> wanted to check.
>
> Imran
>
> On Wed, Jun 10, 2015 at 12:20 AM, Mike Hynes <91...@gmail.com> wrote:
>
>> Hi Imran,
>> One additional quick question---I just want to confirm that I fully
>> understand your comment that "blocks are fetched before they are
>> needed." Typically on our system, we see spikes in cluster bandwidth
>> (with ganglia) at stage boundaries, so I previously assumed that all
>> shuffle read occurred there. Do you mean that the blocks are fetched
>> by the shuffle read iterator, and hence when tasks occur afterwards
>> the necessary blocks have already been fetched?
>> Thanks---I am sorry if this is an obvious question, but I'd like to
>> understand this as precisely as possible.
>> Mike
>>
>> On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>> > Ahhh---forgive my typo: what I mean is,
>> > (t2 - t1) >= (t_ser + t_deser + t_exec)
>> > is satisfied, empirically.
>> >
>> > On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
>> >> 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 <ir...@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 <91...@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 <pw...@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 <91...@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 <rx...@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 <
>> rtreffer@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-unsubscribe@spark.apache.org
>> >>>> >>> For additional commands, e-mail: dev-help@spark.apache.org
>> >>>> >>>
>> >>>> >>
>> >>>> >
>> >>>>
>> >>>>
>> >>>> --
>> >>>> Thanks,
>> >>>> Mike
>> >>>>
>> >>>>
>> >>>> ---------------------------------------------------------------------
>> >>>> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
>> >>>> For additional commands, e-mail: dev-help@spark.apache.org
>> >>>>
>> >>>
>> >>
>> >>
>> >> --
>> >> Thanks,
>> >> Mike
>> >>
>> >
>> >
>> > --
>> > Thanks,
>> > Mike
>> >
>>
>>
>> --
>> Thanks,
>> Mike
>>
>
--
Thanks,
Mike
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
For additional commands, e-mail: dev-help@spark.apache.org
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Mike Hynes <91...@gmail.com>.
Ahhh---forgive my typo: what I mean is,
(t2 - t1) >= (t_ser + t_deser + t_exec)
is satisfied, empirically.
On 6/10/15, Mike Hynes <91...@gmail.com> wrote:
> 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 <ir...@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 <91...@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 <pw...@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 <91...@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 <rx...@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 <rt...@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-unsubscribe@spark.apache.org
>>> >>> For additional commands, e-mail: dev-help@spark.apache.org
>>> >>>
>>> >>
>>> >
>>>
>>>
>>> --
>>> Thanks,
>>> Mike
>>>
>>>
>>> ---------------------------------------------------------------------
>>> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
>>> For additional commands, e-mail: dev-help@spark.apache.org
>>>
>>
>
>
> --
> Thanks,
> Mike
>
--
Thanks,
Mike
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
For additional commands, e-mail: dev-help@spark.apache.org
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Mike Hynes <91...@gmail.com>.
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 <ir...@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 <91...@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 <pw...@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 <91...@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 <rx...@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 <rt...@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-unsubscribe@spark.apache.org
>> >>> For additional commands, e-mail: dev-help@spark.apache.org
>> >>>
>> >>
>> >
>>
>>
>> --
>> Thanks,
>> Mike
>>
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
>> For additional commands, e-mail: dev-help@spark.apache.org
>>
>
--
Thanks,
Mike
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
For additional commands, e-mail: dev-help@spark.apache.org
Re: Stages with non-arithmetic numbering & Timing metrics in event logs
Posted by Imran Rashid <ir...@cloudera.com>.
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 <91...@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 <pw...@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 <91...@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 <rx...@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 <rt...@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-unsubscribe@spark.apache.org
> >>> For additional commands, e-mail: dev-help@spark.apache.org
> >>>
> >>
> >
>
>
> --
> Thanks,
> Mike
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe@spark.apache.org
> For additional commands, e-mail: dev-help@spark.apache.org
>