From 952457f9fae1cc25cdeeefc00e19ae20cf86c659 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Sat, 10 Aug 2019 19:50:18 -0700 Subject: please command for groupworksfatcat --- .../scala/sandcrawler/GroupFatcatWorksJob.scala | 43 ++++++++++++++++++++++ .../main/scala/sandcrawler/ScoreSelfFatcat.scala | 43 ---------------------- 2 files changed, 43 insertions(+), 43 deletions(-) create mode 100644 scalding/src/main/scala/sandcrawler/GroupFatcatWorksJob.scala delete mode 100644 scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala (limited to 'scalding/src/main/scala') diff --git a/scalding/src/main/scala/sandcrawler/GroupFatcatWorksJob.scala b/scalding/src/main/scala/sandcrawler/GroupFatcatWorksJob.scala new file mode 100644 index 0000000..46d2038 --- /dev/null +++ b/scalding/src/main/scala/sandcrawler/GroupFatcatWorksJob.scala @@ -0,0 +1,43 @@ +package sandcrawler + +import cascading.pipe.Pipe +import com.twitter.scalding.Args +import com.twitter.scalding.Stat +import com.twitter.scalding.TypedPipe +import com.twitter.scalding.TypedTsv +import parallelai.spyglass.base.JobBase + +class GroupFatcatWorksJob(args: Args) extends JobBase(args) { + + val fatcatRowCount = Stat("fatcat-rows-filtered", "sandcrawler") + val joinedRowCount = Stat("joined-rows", "sandcrawler") + + val fatcatScorable : Scorable = new FatcatScorable() + val fatcatPipe : TypedPipe[(String, ReduceFeatures)] = fatcatScorable + .getInputPipe(args) + .map { r => + fatcatRowCount.inc + r + } + + val joinedPipe = fatcatPipe + .addTrap(TypedTsv(args("output") + ".trapped")) + .join(fatcatPipe) + + // TypedTsv doesn't work over case classes. + joinedPipe + // filter out trivial self-matches (releases are identical) + .filter { case (slug, (fatcatFeaturesLeft, fatcatFeaturesRight)) => + Scorable.selfMatchable(fatcatFeaturesLeft, fatcatFeaturesRight) + } + .map { case (slug, (fatcatFeaturesLeft, fatcatFeaturesRight)) => + joinedRowCount.inc + new ReduceOutput( + slug, + Scorable.computeSimilarity(fatcatFeaturesLeft, fatcatFeaturesRight), + fatcatFeaturesLeft.json, + fatcatFeaturesRight.json) + } + .map { entry => (entry.slug, entry.score, entry.json1, entry.json2) } + .write(TypedTsv[(String, Int, String, String)](args("output"))) +} diff --git a/scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala b/scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala deleted file mode 100644 index d1a94fe..0000000 --- a/scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala +++ /dev/null @@ -1,43 +0,0 @@ -package sandcrawler - -import cascading.pipe.Pipe -import com.twitter.scalding.Args -import com.twitter.scalding.Stat -import com.twitter.scalding.TypedPipe -import com.twitter.scalding.TypedTsv -import parallelai.spyglass.base.JobBase - -class ScoreSelfFatcatJob(args: Args) extends JobBase(args) { - - val fatcatRowCount = Stat("fatcat-rows-filtered", "sandcrawler") - val joinedRowCount = Stat("joined-rows", "sandcrawler") - - val fatcatScorable : Scorable = new FatcatScorable() - val fatcatPipe : TypedPipe[(String, ReduceFeatures)] = fatcatScorable - .getInputPipe(args) - .map { r => - fatcatRowCount.inc - r - } - - val joinedPipe = fatcatPipe - .addTrap(TypedTsv(args("output") + ".trapped")) - .join(fatcatPipe) - - // TypedTsv doesn't work over case classes. - joinedPipe - // filter out trivial self-matches (releases are identical) - .filter { case (slug, (fatcatFeaturesLeft, fatcatFeaturesRight)) => - Scorable.selfMatchable(fatcatFeaturesLeft, fatcatFeaturesRight) - } - .map { case (slug, (fatcatFeaturesLeft, fatcatFeaturesRight)) => - joinedRowCount.inc - new ReduceOutput( - slug, - Scorable.computeSimilarity(fatcatFeaturesLeft, fatcatFeaturesRight), - fatcatFeaturesLeft.json, - fatcatFeaturesRight.json) - } - .map { entry => (entry.slug, entry.score, entry.json1, entry.json2) } - .write(TypedTsv[(String, Int, String, String)](args("output"))) -} -- cgit v1.2.3