Repository: mahout Updated Branches: refs/heads/flink-binding b692cbfcc -> eda8a38e4
MAHOUT-1805: Fix previous conflict resolution error Project: http://git-wip-us.apache.org/repos/asf/mahout/repo Commit: http://git-wip-us.apache.org/repos/asf/mahout/commit/eda8a38e Tree: http://git-wip-us.apache.org/repos/asf/mahout/tree/eda8a38e Diff: http://git-wip-us.apache.org/repos/asf/mahout/diff/eda8a38e Branch: refs/heads/flink-binding Commit: eda8a38e40d3aa938ea525183ff484957f2c8597 Parents: b692cbf Author: Andrew Palumbo <[email protected]> Authored: Tue Mar 15 15:52:43 2016 -0400 Committer: Andrew Palumbo <[email protected]> Committed: Tue Mar 15 15:52:43 2016 -0400 ---------------------------------------------------------------------- .../scala/org/apache/mahout/flinkbindings/FlinkEngine.scala | 7 +++++++ 1 file changed, 7 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/mahout/blob/eda8a38e/flink/src/main/scala/org/apache/mahout/flinkbindings/FlinkEngine.scala ---------------------------------------------------------------------- diff --git a/flink/src/main/scala/org/apache/mahout/flinkbindings/FlinkEngine.scala b/flink/src/main/scala/org/apache/mahout/flinkbindings/FlinkEngine.scala index 616512e..0ac6011 100644 --- a/flink/src/main/scala/org/apache/mahout/flinkbindings/FlinkEngine.scala +++ b/flink/src/main/scala/org/apache/mahout/flinkbindings/FlinkEngine.scala @@ -366,6 +366,13 @@ object FlinkEngine extends DistributedEngine { implicit val typeInformation = generateTypeInformation[K] val res = drm.asBlockified.ds.map(par => bmf(par)).reduce(rf) + res.collect().head + } + + private def generateTypeInformation[K: ClassTag]: TypeInformation[K] = { + val tag = implicitly[ClassTag[K]] + + generateTypeInformationFromTag(tag) } private def generateTypeInformationFromTag[K](tag: ClassTag[K]): TypeInformation[K] = {
