Hi Flinker,
I try to implement a quadratic distribution i.e. I would like to choose an
element from a dataset with probability proportional to it's squared value.
In Python this would look like this:
s = numpy.cumsum(residual**2)
x = numpy.random.rand() * s[-1]
return residual[numpy.sum(x > s)]
With Flink it is somewhat more complicated, I gave it a try:
import util.Random
val X = DataSource(XFile, CsvInputFormat[Float])
val Y = DataSource(YFile, CsvInputFormat[Float])
// take square of them
val X_2 = X map { x => (x*x, x) }
// calc sum of squares
val X_sum = X_2 reduce { (x1, x2) => (x1._1 + x2._1, 0) } map { x => x._1 }
// choose random value in our range
val y = X_sum map { Random.nextFloat * _ }
// make cummulative sum and find value we search for
val center = X_2 map {
x => (0.0f, x._1, x._2) //sum, x^2, x
} reduce {
(x1, x2) =>
if(x1._1 > y){// already found value we searched for
x1
} else {
if(x1._1 + x2._2 > y){// this is the value we search for
(x1._1 + x2._2, x2._2, x2._3)
} else {
(x1._1 + x2._2, x1._2, x2._3) // just go on with cummulative sum
}
}
} map { _._3 } // we just need the initial value
val output = center //map { x => println(x); x }
val sink = output.write("/tmp/test", CsvOutputFormat[Float], "Center
output")
My problem here is now, I need to get the information stored in y into the
reduce statement to gather the center value. Unfortunately I have no idea
how to achieve that. If somebody knows a way I would be rather thankful. If
someone would know a easier way to solve this problem too!
Many thanks in advance!
Cheers Max
