简介
我必须编写分布式应用程序，该应用程序计算3条记录的唯一值的最大数量。我没有这方面的经验，也不了解框架。我的输入可能如下所示:
u1: u2,u3,u4,u5,u6
u2: u1,u4,u6,u7,u8
u3: u1,u4,u5,u9
u4: u1,u2,u3,u6
...
那么结果的开始应该是:
(u1,u2,u3), u4,u5,u6,u7,u8,u9 => count=6
(u1,u2,u4), u3,u5,u6,u7,u8 => count=5
(u1,u3,u4), u2,u5,u6,u9 => count=4
(u2,u3,u4), u1,u5,u6,u7,u8,u9 => count=6
...
因此，我的方法是先合并每两个记录，然后再将每对合并的记录与每条记录合并。
问题
我可以像hadoop / spark这样的框架中执行这样的操作，例如在多个输入行上同时工作(合并)吗？还是我的方法不正确，我应该采取其他方式吗？
任何建议将被认真考虑。
最佳答案
是的你可以。
这取决于数据的大小。如果您的数据很小，则在本地进行处理会更快，更轻松。如果您的数据量巨大(至少数百GB)，通常的策略是将数据保存到HDFS(分布式文件系统)中，并使用Mapreduce / Spark进行分析。
用scala编写的示例spark应用程序:
object MyCounter {
val sparkConf = new SparkConf().setAppName("My Counter")
val sc = new SparkContext(sparkConf)
def main(args: Array[String]) {
val inputFile = sc.textFile("hdfs:///inputfile.txt")
val keys = inputFile.map(line => line.substring(0, 2)) // get "u1" from "u1: u2,u3,u4,u5,u6"
val triplets = keys.cartesian(keys).cartesian(keys)
.map(z => (z._1._1, z._1._2, z._2))
.filter(z => !z._1.equals(z._2) && !z._1.equals(z._3) && !z._2.equals(z._3)) // get "(u1,u2,u3)" triplets
// If you have small numbers of (u1,u2,u3) triplets, it's better prepare them locally.
val res = triplets.cartesian(inputFile).filter(z => {
z._2.startsWith(z._1._1) || z._2.startsWith(z._1._2) || z._2.startsWith(z._1._3)
}) // (u1,u2,u3) only matches line starts with u1,u2,u3, for example "u1: u2,u3,u4,u5,u6"
.reduceByKey((a, b) => a + b) // merge three lines
.map(z => {
val line = z._2
val values = line.split(",")
//count unique values using set
val set = new util.HashSet[String]()
for (value <- values) {
set.add(value)
}
"key=" + z._1 + ", count=" + set.size() // the result from one mapper is a string
}).collect()
for (line <- res) {
println(line)
}
}
}
关于hadoop - 合并分布式应用程序中的输入，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/37628490/