diff options
author | Bryan Newbold <bnewbold@archive.org> | 2019-08-02 17:11:57 -0700 |
---|---|---|
committer | Bryan Newbold <bnewbold@archive.org> | 2019-08-10 19:50:21 -0700 |
commit | ea9e8990139973d6f5fdf52a470bf6516c7d8c2f (patch) | |
tree | 61e63509c28b7f280e7673c27276d28e8b0782ad /scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala | |
parent | ca725ffd9efe847905afb918ff324b421a4d8859 (diff) | |
download | sandcrawler-ea9e8990139973d6f5fdf52a470bf6516c7d8c2f.tar.gz sandcrawler-ea9e8990139973d6f5fdf52a470bf6516c7d8c2f.zip |
FatcatScorable and ScoreSelfFatcat job
Diffstat (limited to 'scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala')
-rw-r--r-- | scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala | 43 |
1 files changed, 43 insertions, 0 deletions
diff --git a/scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala b/scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.scala new file mode 100644 index 0000000..d1a94fe --- /dev/null +++ b/scalding/src/main/scala/sandcrawler/ScoreSelfFatcat.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 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"))) +} |