I modified to
detailInputsToGroup.map {
case (detailInput, dataRecord) =>
val key: StringBuilder = new StringBuilder
dimensions.foreach {
dimension =>
key ++= {
Option(dataRecord.get(dimension)).getOrElse(Option(detailInput.get(dimension)).getOrElse("")).toString
}
}
(key.toString, (detailInput, dataRecord))
}.reduceByKey {
case (v1, v2) => {
val v1Detail = v1._1
val v2Detail = v2._1
val v1Data = v1._2
val v2Data = v2._2
* val totalViCount =
Option(v1Data.get("totalViCount").asInstanceOf[Int]).getOrElse(0)*
* v1Data.getRecord.put("totalViCount", totalViCount + 1)*
(v1)
}
}.map {
case (k, v) => {
val schema = SchemaUtil.outputSchema(_detail)
val detailOutputRecord = new DetailOutputRecord(detail, new
SessionRecord(schema))
//Compute dimensions
DataUtil.populateDimensions(schema, dimensions.toArray, v._1,
v._2, detailOutputRecord)
//Construct Output
val wrap = new AvroKey[DetailOutputRecord](detailOutputRecord)
(wrap, NullWritable.get)
}
}
How do i compute unique count ?
On Tue, Jun 30, 2015 at 12:04 PM, Daniel Siegmann <
daniel.siegm...@teamaol.com> wrote:
> If the number of items is very large, have you considered using
> probabilistic counting? The HyperLogLogPlus
> <https://github.com/addthis/stream-lib/blob/master/src/main/java/com/clearspring/analytics/stream/cardinality/HyperLogLogPlus.java>
> class from stream-lib <https://github.com/addthis/stream-lib> might be
> suitable.
>
> On Tue, Jun 30, 2015 at 2:29 PM, ÐΞ€ρ@Ҝ (๏̯͡๏) <deepuj...@gmail.com>
> wrote:
>
>> I have a RDD of type (String,
>>
>> Iterable[(com.ebay.ep.poc.spark.reporting.process.detail.model.DetailInputRecord,
>> com.ebay.ep.poc.spark.reporting.process.model.DataRecord)])]
>>
>> Here String is Key and a list of tuples for that key. I got above RDD
>> after doing a groupByKey. I later want to compute total number of values
>> for a given key and total number of unique values for the same given key
>> and hence i do this
>>
>> val totalViCount = details.size.toLong
>> val uniqueViCount =
>> details.map(_._1.get("itemId").asInstanceOf[Long]).distinct.size.toLong
>>
>> How do i do this using reduceByKey.
>>
>> *Total Code:*
>>
>> val groupedDetail: RDD[(String, Iterable[(DetailInputRecord,
>> DataRecord)])] = detailInputsToGroup.map {
>> case (detailInput, dataRecord) =>
>> val key: StringBuilder = new StringBuilder
>> dimensions.foreach {
>> dimension =>
>> key ++= {
>>
>> Option(dataRecord.get(dimension)).getOrElse(Option(detailInput.get(dimension)).getOrElse("")).toString
>> }
>> }
>> (key.toString, (detailInput, dataRecord))
>> }.groupByKey
>>
>> groupedDetail.map {
>> case (key, values) => {
>> val valueList = values.toList
>>
>> //Compute dimensions // You can skup this
>> val (detailInput, dataRecord) = valueList.head
>> val schema = SchemaUtil.outputSchema(_detail)
>> val detailOutput = new DetailOutputRecord(detail, new
>> SessionRecord(schema))
>> DataUtil.populateDimensions(schema, dimensions.toArray,
>> detailInput, dataRecord, detailOutput)
>>
>>
>> val metricsData = metricProviders.flatMap {
>> case (className, instance) =>
>> val data = instance.getMetrics(valueList)
>> ReflectionUtil.getData(data,
>> _metricProviderMemberNames(className))
>> }
>> metricsData.map { case (k, v) => detailOutput.put(k, v) }
>> val wrap = new AvroKey[DetailOutputRecord](detailOutput)
>> (wrap, NullWritable.get)
>> }
>> }
>>
>>
>> //getMetrics:
>> def getMetrics(details: List[(DetailInputRecord, DataRecord)]) = {
>> val totalViCount = details.size.toLong
>> val uniqueViCount =
>> details.map(_._1.get("itemId").asInstanceOf[Long]).distinct.size.toLong
>> new ViewItemCountMetric(totalViCount, uniqueViCount)
>> }
>>
>>
>> I understand that totalViCount can be implemented using reduceByKey. How
>> can i implement total unique count as i need to have the full list to know
>> the unique values.
>>
>> --
>> Deepak
>>
>>
>
--
Deepak
