azymnis · December 13, 2013 05:17
diff --git a/ItemSimilarity.scala b/ItemSimilarity.scala
 import com.twitter.scalding._
 import com.twitter.algebird.{ MinHasher, MinHasher32, MinHashSignature }

 /**
 * Computes similar items (with a string itemId), based on approximate
 * Jaccard similarity, using LSH.
 *
 * Assumes an input data TSV file of the following format:
 *
 *    itemId   userId
 *
 * Generates an output file of the following format:
 *
 *    itemId   similarItemId
 *
 * Input arguments:
 *    --input              location of input file
 *    --output             location of output file
 *    --num_hashes         number of hash functions to use, more means
 *                         higher complexity, but higher accuracy
 *    --target_threshold   minimum Jaccard similarity above which two
 *                         items are considered similar
 *
 */
 class ItemSimilarity(args: Args) extends Job(args) {
  import TDsl._

  val targetThreshold = args.optional("target_threshold")
    .map { _.toDouble }.getOrElse(0.8)

  val numHashes = args.optional("num_hashes")
    .map { _.toInt }.getOrElse(50)

  val numBands = MinHasher.pickBands(targetThreshold, numHashes)
  implicit lazy val minHasher = new MinHasher32(numHashes, numBands)

  TypedTsv[(String, Long)](args("input"))

    // First generate minhash signatures
    .map { case (itemId, userId) => (itemId, minHasher.init(userId)) }
    .group[String, MinHashSignature]
    .sum

    // Now generate bands of similar books and aggregate
    .flatMap { case (itemId, sig) =>
      minHasher.buckets(sig).zipWithIndex.map { case (bucket, ind) =>
        ((bucket, ind), Set((itemId, sig)))
      }
    }
    .group[(Long, Int), Set[(String, MinHashSignature)]]
    .sum

    // Now expand all pairs of similar books
    .flatMap { case (_, itemIdSet) =>
      for {
        (itemId1, sig1) <- itemIdSet
        (itemId2, sig2) <- itemIdSet
        sim = minHasher.similarity(sig1, sig2)
         if (itemId1 != itemId2 && sim >= targetThreshold)
      } yield (itemId1, itemId2)
    }
    .distinct
    .write(TypedTsv[(String, String)](args("output")))
 }
	import com.twitter.scalding._
	import com.twitter.algebird.{ MinHasher, MinHasher32, MinHashSignature }

	/**
	* Computes similar items (with a string itemId), based on approximate
	* Jaccard similarity, using LSH.
	*
	* Assumes an input data TSV file of the following format:
	*
	* itemId userId
	*
	* Generates an output file of the following format:
	*
	* itemId similarItemId
	*
	* Input arguments:
	* --input location of input file
	* --output location of output file
	* --num_hashes number of hash functions to use, more means
	* higher complexity, but higher accuracy
	* --target_threshold minimum Jaccard similarity above which two
	* items are considered similar
	*
	*/
	class ItemSimilarity(args: Args) extends Job(args) {
	import TDsl._

	val targetThreshold = args.optional("target_threshold")
	.map { _.toDouble }.getOrElse(0.8)

	val numHashes = args.optional("num_hashes")
	.map { _.toInt }.getOrElse(50)

	val numBands = MinHasher.pickBands(targetThreshold, numHashes)
	implicit lazy val minHasher = new MinHasher32(numHashes, numBands)

	TypedTsv[(String, Long)](args("input"))

	// First generate minhash signatures
	.map { case (itemId, userId) => (itemId, minHasher.init(userId)) }
	.group[String, MinHashSignature]
	.sum

	// Now generate bands of similar books and aggregate
	.flatMap { case (itemId, sig) =>
	minHasher.buckets(sig).zipWithIndex.map { case (bucket, ind) =>
	((bucket, ind), Set((itemId, sig)))
	}
	}
	.group[(Long, Int), Set[(String, MinHashSignature)]]
	.sum

	// Now expand all pairs of similar books
	.flatMap { case (_, itemIdSet) =>
	for {
	(itemId1, sig1) <- itemIdSet
	(itemId2, sig2) <- itemIdSet
	sim = minHasher.similarity(sig1, sig2)
	if (itemId1 != itemId2 && sim >= targetThreshold)
	} yield (itemId1, itemId2)
	}
	.distinct
	.write(TypedTsv[(String, String)](args("output")))
	}