Transitive closure in Flink for redirect resolving.

RedirectResolvingProgram.scala

import org.apache.flink.api.scala._

class RedirectResolvingProgram {

	case class LinkWithOrig(from: String, origTo: String, to: String)
	case class Redirect(from: String, to: String)

	override def buildProgram(env: ExecutionEnvironment): Unit = {
		val redirects    = env.readTextFile(redirectPath).map { line =>
			val split = line.split('\t')
			Redirect(split(0), split(1))
		}.name("Redirects")

		val links = env.readTextFile(linksPath).map { line =>
			val split = line.split('\t')
			LinkWithOrig(split(0), split(1), split(1))
		}.name("Links")

		def iterate(s: DataSet[LinkWithOrig], ws: DataSet[LinkWithOrig]): (DataSet[LinkWithOrig], DataSet[LinkWithOrig]) = {
			val resolvedRedirects = redirects.join(ws)
				.where { _.from }
				.equalTo { _.to }
				.map { joinResult => joinResult match {
					case (redirect, link) =>
						link.copy(to = redirect.to)
				}
			}.name("Resolved-Redirects-From-Iteration")
			(resolvedRedirects, resolvedRedirects)
		}


		// resolve redirects via delta iteration
		val resolvedRedirects = links
			.iterateDelta(links, 10, Array("from", "origTo"))(iterate)
		.name("Resolved-Redirects")
		.map { cl => (cl.from, cl.to) }
		.name("Final-Redirect-Result")

		resolvedRedirects.writeAsTsv(resolvedRedirectsPath)
	}
}