Joep Rottinghuis commented on YARN-3901:

Not sure if [~gtCarrera9] is asking what happens if we have consolidated a 
"finished" app into the flow aggregate and then we get an older batched put 
from the same app (from an older AM that wasn't properly killed or something). 
That is currently an open problem that we can address by storing the list of x 
most recently completed apps in a tag on the total sum. We can't quite store 
all app IDs that have been collapsed into the flow sum, because that list could 
potentially be really long. We could keep a list of the last x apps, to 
radically reduce the likelihood that stale batched writes end up messing up the 
aggregation if there were to be a race condition (however rare that might be). 
I think we can add that guarding behavior in a separate jira and not further 
complicate the first cut (that might suffer from this rare race condition).

As [~sjlee0] pointed out and we were discussing offline this morning 
scan.setMaxResultSize(limit); doesn't limit the # rows that are returned, but 
limits the size in bytes. Not sure we want to address that here, or if we'll 
let him adjust that in his patch.
We should limit the number of rows we retrieve from scan and if needed (as 
[~sjlee0] pointed out) add a PageFilter with a limit in addition to the limit 
to further restrict what is prefetched in a buffer (gets more complicated when 
scan spans region servers).

It is a little confusing to read the patch and see where ColumnPrefix has just 
an additional argument in the store method for Attribute... attributes and when 
the method with a String qualifier is changed to one with a byte[]. There seems 
to be a discrepancy in how EntityTablePrefix and ApplicationTablePrefix are 
handles. I'm not sure if is needed to have a getColumnQualifier with a long in 
ColumnHelper, but we may have to review this together interactively behind two 

In HBaseTimelineWriterImpl. onApplicationFinished
You have an old comment:
281        // clear out the hashmap in case the downstream store/coprocessor
282           // added any more attributes during the put for END_TIME
that no longer makes sense. Also, I'd simply do:
storeFlowMetrics(rowKey, metrics, attribute1, 

I'd update the javadoc in AggregationOperations before the enum definitions. 
They are the old style and no longer make sense in the more generic case.
SUM indicates that the values need to be summed up, MIN means that only the 
minimum needs to be kept etc.

Initially I found the method name getIncomingAttributes and corresponding 
member names somewhat confusing (from the method perspective it isn't the 
incoming values, it is the outgoing values). Perhaps combinedAttribute and 
combineAttributes(...) makes more sense, but the provided logic seems correct.

FlowActivityColumnPrefix.IN_PROGRESS_TIME needs a better javadoc description to 
describe its use and meaning.

The coprocessor methods need a little more javadoc to explain what is going on. 
To the casual reader this is total voodoo.
The preGetOp creates a new scanner (ok), then does a single next on it (why?) 
and then bypasses the environment (huh?).
Similarly if in preScannerOpen we already set scan.setMaxVersions(); then why 
is the same still needed in PreGetOp, but in PostScannerOpen we don't do it 
anymore (presumably already done in the preOpen).

I like the more generic FlowRunCoprocessor (although it can have a name that is 
not associated with a table, because behavior is generic, and arg names such as 
frpa are probably artifact from previous version).
In getTagFromAttribute, is it possible to recognize a operation from an 
AggregationCompactionDimension without relying on an exception and catching it? 
For example, can you do AggregationOperation.isA(Attribute a) or something like 

The other thing I realize with the coprocessor is this. It nicely maps 
attributes to tags, but we unnecessarily bloat every single put with the 
We could get creative and use a different column prefix for min and max 
columns. Then the coprocessor can pick that up during read/flush/compaction. 
That makes queries (and filters) much harder. So for now we're probably stuck 
with tagging each value. Perhaps not so bad for min and max given that after 
flush and compact we store only one value.

For SUM we will always have an Aggregation dimension, so adding a SUM tag then 
isn't needed. We assume an aggregation dimension w/o agg operation would 
default to SUM. We do certainly need to tag values with SUM_FINAL.

Aside from that, in FlowRunProcessor.prePut do we have to keep doing 
Tag.fromList(tags) for each cell, or can we create a Tag <list/array> once and 
re-use it?

When reading through the coprocessor it still has Application_ID specific code. 
There is no need to enumerate all the AGG dimensions, that will only be one per 
cell. So the agg dimension itself is an attribute, it has a value. In some 
cases it happens to be APP_ID, but in other cases it is something else.
That doesn't matter to the coprocessor.
In FlowScanner. getAggregationCompactionDimension it doesn't have to be cast to 
String does it? Couldn't it simply be a byte[] with whatever content it is?

In nextInternal there is no need to do getAggregationCompactionDimension unless 
it is a sum (or SUM_FINAL). I see MIN and MAX and SUM handled, but SUM_FINAL is 
ignored. Here SUM_FINAL should be treated as SUM.
addedCnt is increased only for regular cells. It should be increased once per 
collapsed cell as well I think in the case when we haven't seen this.
The behavior should probably differ between a read and a flush/compact.

I can't quite see the coprocessor handle this difference correctly yet. I think 
it should be part of the constructor. For reads we should return exactly one 
value: the final sum. For flushes we need to return the previous sum, plus one 
individual value for each aggregation dimension (modulo ones that can be 

In processMetricCell you see the problem that numbers should be stored through 
Bytes.toBytes(long l) instead of GenericObjectMapper. That is where the tricky 
part comes in with mixing longs and doubles.

Also general comment: check your spacing. There are places where tabs are used 
instead of spaces.

Overall it looks really good, but I need to walk through the store and read 
cases with you and see how the flush/compact scanning works compared to the 
read. I think we're close, but let's review that together.

> Populate flow run data in the flow_run & flow activity tables
> -------------------------------------------------------------
>                 Key: YARN-3901
>                 URL: https://issues.apache.org/jira/browse/YARN-3901
>             Project: Hadoop YARN
>          Issue Type: Sub-task
>          Components: timelineserver
>            Reporter: Vrushali C
>            Assignee: Vrushali C
>         Attachments: YARN-3901-YARN-2928.1.patch, 
> YARN-3901-YARN-2928.2.patch, YARN-3901-YARN-2928.3.patch, 
> YARN-3901-YARN-2928.WIP.2.patch, YARN-3901-YARN-2928.WIP.patch
> As per the schema proposed in YARN-3815 in 
> https://issues.apache.org/jira/secure/attachment/12743391/hbase-schema-proposal-for-aggregation.pdf
> filing jira to track creation and population of data in the flow run table. 
> Some points that are being  considered:
> - Stores per flow run information aggregated across applications, flow version
> RM’s collector writes to on app creation and app completion
> - Per App collector writes to it for metric updates at a slower frequency 
> than the metric updates to application table
> primary key: cluster ! user ! flow ! flow run id
> - Only the latest version of flow-level aggregated metrics will be kept, even 
> if the entity and application level keep a timeseries.
> - The running_apps column will be incremented on app creation, and 
> decremented on app completion.
> - For min_start_time the RM writer will simply write a value with the tag for 
> the applicationId. A coprocessor will return the min value of all written 
> values. - 
> - Upon flush and compactions, the min value between all the cells of this 
> column will be written to the cell without any tag (empty tag) and all the 
> other cells will be discarded.
> - Ditto for the max_end_time, but then the max will be kept.
> - Tags are represented as #type:value. The type can be not set (0), or can 
> indicate running (1) or complete (2). In those cases (for metrics) only 
> complete app metrics are collapsed on compaction.
> - The m! values are aggregated (summed) upon read. Only when applications are 
> completed (indicated by tag type 2) can the values be collapsed.
> - The application ids that have completed and been aggregated into the flow 
> numbers are retained in a separate column for historical tracking: we don’t 
> want to re-aggregate for those upon replay

This message was sent by Atlassian JIRA

Reply via email to