You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by "Christoph Viebig (lists)" <li...@christoph-viebig.de> on 2014/06/09 17:10:01 UTC

DeltaIterations: Solution set delta does not depend on workset

Hi,
in an university project at TU Berlin we are implementing the Random Forest
algorithm on Stratosphere. We decided to use DeltaIterations to perform the
computation. The algorithm will grow a random tree (a sort of decision tree)
iteratively. In each iteration a new level of the tree is created and the data
is assigned to the corresponding child node in that level. The solution set is
the set of nodes in the tree. It is appended in each iteration. The workset is
the set of data. It is modified in each iteration as each data record also
stores the assigned node in the tree. Unfortunately we are experiencing a PACT
compiler exception. Can you please give us an advise how to resolve it? What is
the solution set delta and how should it depend on the workset? We append the
solution set in each iteration and use it to manipulate the workset.

Thank you very much in advance!


Best regards
Christoph

```

Exception in thread "main" eu.stratosphere.compiler.CompilerException: In the
given plan, the solution set delta does not depend on the workset. This is a
prerequisite in workset iterations.
at
eu.stratosphere.compiler.PactCompiler$GraphCreatingVisitor.postVisit(PactCompiler.java:1025)
at
eu.stratosphere.compiler.PactCompiler$GraphCreatingVisitor.postVisit(PactCompiler.java:777)
at
eu.stratosphere.api.common.operators.DualInputOperator.accept(DualInputOperator.java:263)
at
eu.stratosphere.api.common.operators.SingleInputOperator.accept(SingleInputOperator.java:180)
at
eu.stratosphere.api.common.operators.GenericDataSink.accept(GenericDataSink.java:409)
at eu.stratosphere.api.common.Plan.accept(Plan.java:298)
at eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:670)
at eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:553)
at eu.stratosphere.client.LocalExecutor.executePlan(LocalExecutor.java:216)
at eu.stratosphere.client.LocalExecutor.execute(LocalExecutor.java:301)
at
de.tu_berlin.impro3.stratosphere.classification.randomforest.RunRandomForest$.main(RandomForest.scala:220)
at
de.tu_berlin.impro3.stratosphere.classification.randomforest.RunRandomForest.main(RandomForest.scala)
```
Code:
```
case class Node(id: Int, split: Split, maxLabel: String, leaf: Boolean = false)
{}

case class Split(dim: Int, value: Double) {}

case class Point(features: Vector, label: String) {}

case class Data(nodeId: Int, point: Point) {}

/* Compute first node in tree.
*/
def computeFirstNode(data: DataSet[Data]): DataSet[Node] = {
// Dummy.
val newNodes = Seq[Node]()
CollectionDataSource(newNodes)
}

/* Create new child nodes if a node has more than class assigned to it
* Find a good split within the data that is assigned
*/
def splitNodes(data: DataSet[Data]): DataSet[Node] = {
// Dummy.
val newNodes = Seq[Node]()
CollectionDataSource(newNodes)
}

/* Assign data to the new nodes.
* If there is no child node k*2 or k*2+1 data is omitted otherwise it's assigned
to the child node depending on the split.
*/
def reassignData(nodes: DataSet[Node], data: DataSet[Data]) : DataSet[Data] = {
// Dummy. It modifies Data.nodeId for each Data instance
(data)
}

/* Grow the tree and create a new level.
*/
def growTree(s: DataSet[Node], ws: DataSet[Data]) = {
val newNodes = splitNodes(ws)
val newWorkSet = reassignData(newNodes, ws)
val newSolutionSet = s.union(newNodes)
(newSolutionSet, newWorkSet)
}

def getScalaPlan(numTrees: Int, maxDepth: Int, bootstrapRatio: Double,
inputFile: String, outputFile: String) = {

val points = DataSource(inputFile, CsvInputFormat[(Double, Double, Double,
String, String, String, String)]("\n", ','))
/* Create case objects from input data
* => (Point)
*/
.map { case (x, y, z, _, _, _, label) => Point(new Vector(List[Double](x, y,
z)), label)}

val data = points
/* Sample data depending on bootstrapRatio and a random number
* (Point) => (Point)
*/
.filter( point => Random.nextDouble < bootstrapRatio )
/* Initialize work set for delta iteration.
* First nodeId is 1 with depth 0.
* (Point) => (nodeId, Point)
*/
.map { x => Data(1, x) }

/* Initialize tree
*/
val initializedTree = computeFirstNode(data);

/* Perform delta iteration on data to grow a tree.
* Aborting when data is empty or maxDepth is reached.
* See
https://github.com/stratosphere/stratosphere/blob/master/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSet.scala
* See
https://github.com/stratosphere/stratosphere/blob/master/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/IterateOperators.scala
(WorkSetIterateMarcros)
*/
val tree = initializedTree.iterateWithDelta(data, {_.id}, growTree, maxDepth)

/* Write tree (solution set) to disk
*/
val output = tree.map{ case(node) => (node.id, node.split.dim, node.split.value,
node.maxLabel, node.leaf) }.write(outputFile, CsvOutputFormat())

new ScalaPlan(Seq(output), "RandomForest")
}
```

Re: DeltaIterations: Solution set delta does not depend on workset

Posted by "Christoph Viebig (lists)" <li...@christoph-viebig.de>.
Hi Stephan, hi Aljoscha,

thank you for your responses!

I am afraid that I explained the idea of the algorithm not good
enaugh. The workset is being constant until we find leafs in the tree.
This is because we will not split leafs therefore we can remove the
training data tuples that are assigned to them. The tuples that are
assigned to non-leaf nodes are emitted again but with another nodeId
property, i.e. they are being replaced. No append to the set is
performed (see the map function below)

Aljoscha's question has a good point. We did not have an implementation
of the work set modification function until yesterday evening. Stephan
is also right, we will join the work set (training data) with the
solution set (nodes in the tree). Within the function that receives the
joined tuples we will perform the reassignment of the training data to
the new child nodes (k*2 or k*2+1 with k as the currently assigned
node). Training data that is assigned to leafs is removed from the work
set. The join looks like this:

val newWorkSet = data.join(nodes.filter{ node => !node.isLeaf() })
  .where{ data => data.nodeID }.isEqualTo{ node => node.nodeID }
  .map{ (data, node) => new Data(
    if (data.point.features.elementAt(node.split.dim) <= node.split.value)
      data.nodeID * 2
    else
      data.nodeID * 2 + 1
    ,data.point) }

Unfortunately the exception is still raised.

Ragarding the solution set growing: We are generating new nodes with ID
2*k and 2*k+1 for each non-leaf node with ID k. Therefore new keys are
generated. We have removed the union now.

But I am afraid that we did not understood the API to emit the new
workset. We do not have keys on the training data. Is it possible to
re-emit a completely new work set in each iteration or can you only
replace existing tuples in the current work set?

Can you also please have a look on our step function invokation?
Currently it is looking like this:

initializedTree.iterateWithDelta(data, {_.nodeID}, growTree, maxDepth)

with the following variables:
  - initializedTree: Node with ID 1 and split point. (Initial solution
set)
  - data: Training data set. (Initial work set)
  - {_.nodeId} Attribute of the training data. But we are not entirely
sure what this parameter is used for.
  - growTree: Step function returning (solutionSetDelta, newWorkSet)
  - maxDepth: Maximum number of iterations.

Thank you very much for your support!

Best regards
Christoph


> Stephan Ewen <se...@apache.org> hat am 10. Juni 2014 um 16:31 geschrieben:
>
>
> Hey!
>
> The way you have written the algorithm strikes me actually as a case for
> bulk iterations. Delta iterations are typically cases where the work set
> sizes goes down (not strictly up, as here).
>
> In your specific code, the problem is that you can actually only
> join/cogroup with the solution set, and produce deltas for it (deltas are
> merged in to the solution set). You do not re-assign the solution set.
>
> In order to grow the solution set by a set of elements, make a delta where
> elements have new keys (do not replace any the current elements). When
> adding the delta to what is already there, you get a union, with the
> special condition that it is a set union on the keys, and if a key existed
> before, it gets replaced by the delta entry with the same key. So the
> solution set delta is just the result of the "newWorkSet" variable. No need
> to union before.
>
> For the workset, you may have to rethink the algorithm. The way you wrote
> it, the workset will always grow. A delta iteration terminates when the
> workset is empty. Try and write your algorithm such that the workset really
> contains only the changes that are relevant to the next iteration. If the
> changes become zero, the work is done and the iteration terminates.
>
> Stephan

Re: DeltaIterations: Solution set delta does not depend on workset

Posted by Stephan Ewen <se...@apache.org>.
Hey!

The way you have written the algorithm strikes me actually as a case for
bulk iterations. Delta iterations are typically cases where the work set
sizes goes down (not strictly up, as here).

In your specific code, the problem is that you can actually only
join/cogroup with the solution set, and produce deltas for it (deltas are
merged in to the solution set). You do not re-assign the solution set.

In order to grow the solution set by a set of elements, make a delta where
elements have new keys (do not replace any the current elements). When
adding the delta to what is already there, you get a union, with the
special condition that it is a set union on the keys, and if a key existed
before, it gets replaced by the delta entry with the same key. So the
solution set delta is just the result of the "newWorkSet" variable. No need
to union before.

For the workset, you may have to rethink the algorithm. The way you wrote
it, the workset will always grow. A delta iteration terminates when the
workset is empty. Try and write your algorithm such that the workset really
contains only the changes that are relevant to the next iteration. If the
changes become zero, the work is done and the iteration terminates.

Stephan

Re: DeltaIterations: Solution set delta does not depend on workset

Posted by Aljoscha Krettek <al...@apache.org>.
Hi,
this might be a stupid question but, in your actual implementation, do you
have code inside the dummy methods? If yes, could you post that as well?

Cheers,
Aljoscha


On Mon, Jun 9, 2014 at 5:10 PM, Christoph Viebig (lists) <
lists@christoph-viebig.de> wrote:

> Hi,
> in an university project at TU Berlin we are implementing the Random Forest
> algorithm on Stratosphere. We decided to use DeltaIterations to perform the
> computation. The algorithm will grow a random tree (a sort of decision
> tree)
> iteratively. In each iteration a new level of the tree is created and the
> data
> is assigned to the corresponding child node in that level. The solution
> set is
> the set of nodes in the tree. It is appended in each iteration. The
> workset is
> the set of data. It is modified in each iteration as each data record also
> stores the assigned node in the tree. Unfortunately we are experiencing a
> PACT
> compiler exception. Can you please give us an advise how to resolve it?
> What is
> the solution set delta and how should it depend on the workset? We append
> the
> solution set in each iteration and use it to manipulate the workset.
>
> Thank you very much in advance!
>
>
> Best regards
> Christoph
>
> ```
>
> Exception in thread "main" eu.stratosphere.compiler.CompilerException: In
> the
> given plan, the solution set delta does not depend on the workset. This is
> a
> prerequisite in workset iterations.
> at
>
> eu.stratosphere.compiler.PactCompiler$GraphCreatingVisitor.postVisit(PactCompiler.java:1025)
> at
>
> eu.stratosphere.compiler.PactCompiler$GraphCreatingVisitor.postVisit(PactCompiler.java:777)
> at
>
> eu.stratosphere.api.common.operators.DualInputOperator.accept(DualInputOperator.java:263)
> at
>
> eu.stratosphere.api.common.operators.SingleInputOperator.accept(SingleInputOperator.java:180)
> at
>
> eu.stratosphere.api.common.operators.GenericDataSink.accept(GenericDataSink.java:409)
> at eu.stratosphere.api.common.Plan.accept(Plan.java:298)
> at eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:670)
> at eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:553)
> at eu.stratosphere.client.LocalExecutor.executePlan(LocalExecutor.java:216)
> at eu.stratosphere.client.LocalExecutor.execute(LocalExecutor.java:301)
> at
>
> de.tu_berlin.impro3.stratosphere.classification.randomforest.RunRandomForest$.main(RandomForest.scala:220)
> at
>
> de.tu_berlin.impro3.stratosphere.classification.randomforest.RunRandomForest.main(RandomForest.scala)
> ```
> Code:
> ```
> case class Node(id: Int, split: Split, maxLabel: String, leaf: Boolean =
> false)
> {}
>
> case class Split(dim: Int, value: Double) {}
>
> case class Point(features: Vector, label: String) {}
>
> case class Data(nodeId: Int, point: Point) {}
>
> /* Compute first node in tree.
> */
> def computeFirstNode(data: DataSet[Data]): DataSet[Node] = {
> // Dummy.
> val newNodes = Seq[Node]()
> CollectionDataSource(newNodes)
> }
>
> /* Create new child nodes if a node has more than class assigned to it
> * Find a good split within the data that is assigned
> */
> def splitNodes(data: DataSet[Data]): DataSet[Node] = {
> // Dummy.
> val newNodes = Seq[Node]()
> CollectionDataSource(newNodes)
> }
>
> /* Assign data to the new nodes.
> * If there is no child node k*2 or k*2+1 data is omitted otherwise it's
> assigned
> to the child node depending on the split.
> */
> def reassignData(nodes: DataSet[Node], data: DataSet[Data]) :
> DataSet[Data] = {
> // Dummy. It modifies Data.nodeId for each Data instance
> (data)
> }
>
> /* Grow the tree and create a new level.
> */
> def growTree(s: DataSet[Node], ws: DataSet[Data]) = {
> val newNodes = splitNodes(ws)
> val newWorkSet = reassignData(newNodes, ws)
> val newSolutionSet = s.union(newNodes)
> (newSolutionSet, newWorkSet)
> }
>
> def getScalaPlan(numTrees: Int, maxDepth: Int, bootstrapRatio: Double,
> inputFile: String, outputFile: String) = {
>
> val points = DataSource(inputFile, CsvInputFormat[(Double, Double, Double,
> String, String, String, String)]("\n", ','))
> /* Create case objects from input data
> * => (Point)
> */
> .map { case (x, y, z, _, _, _, label) => Point(new Vector(List[Double](x,
> y,
> z)), label)}
>
> val data = points
> /* Sample data depending on bootstrapRatio and a random number
> * (Point) => (Point)
> */
> .filter( point => Random.nextDouble < bootstrapRatio )
> /* Initialize work set for delta iteration.
> * First nodeId is 1 with depth 0.
> * (Point) => (nodeId, Point)
> */
> .map { x => Data(1, x) }
>
> /* Initialize tree
> */
> val initializedTree = computeFirstNode(data);
>
> /* Perform delta iteration on data to grow a tree.
> * Aborting when data is empty or maxDepth is reached.
> * See
>
> https://github.com/stratosphere/stratosphere/blob/master/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSet.scala
> * See
>
> https://github.com/stratosphere/stratosphere/blob/master/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/IterateOperators.scala
> (WorkSetIterateMarcros)
> */
> val tree = initializedTree.iterateWithDelta(data, {_.id}, growTree,
> maxDepth)
>
> /* Write tree (solution set) to disk
> */
> val output = tree.map{ case(node) => (node.id, node.split.dim,
> node.split.value,
> node.maxLabel, node.leaf) }.write(outputFile, CsvOutputFormat())
>
> new ScalaPlan(Seq(output), "RandomForest")
> }
> ```