0

我正在加入两个数据集——第一个来自流，第二个来自 HDFS。

我在火花中使用斯卡拉。加入两个数据集后，我需要对加入的数据集应用过滤器，但在这里我遇到了问题。请协助解决。

我正在使用下面的代码，

 val streamkv = streamrecs.map(_.split("~")).map(r => ( r(0), (r(5), r(6)))) 
val HDFSlines = sc.textFile("/user/Rest/sample.dat").map(_.split("~")).map(r => ( r(1), (r(0) r(3),r(4),))) 
val streamwindow = streamkv.window(Minutes(1)) 

val join1 = streamwindow.transform(joinRDD => { joinRDD.join(HDFSlines)} )

使用过滤器时出现以下错误

val tofilter = join1.filter {
     | case (r(0), (r(5), r(6)),(r(0),r(3),r(4))) =>
     | r(4).contains("iPhone")
     | }.count()
<console>:48: error: constructor cannot be instantiated to expected type;
found   : (T1, T2, T3)
required: (String, ((String, String), (String, String, String)))
       case (r(0), (r(5), r(6)),(r(0),r(3),r(4))) =>
4

2 回答 2

0

在 scala（这是一种类型化语言仅供参考）中，模式匹配（case）允许“提取”块中的变量以供本地使用。所以

case (r(0), (r(5), r(6)),(r(0),r(3),r(4))) =>

是非法的，因为你不是在这里调用 r 函数，而是提取值（想想函数声明）。

假设生成的流/集合 join1 中的对象遵循此类型签名(_, (_, _), (_, _, String))，您应该尝试：

val tofilter = join1.filter {
 | case (_, (_, _),(_,_,device)) =>
 | device.contains("iPhone")
 | }.count()
于 2014-09-11T16:54:59.940 回答
0

错误的原因是缺少一对圆括号。

case (r(0), (r(5), r(6)),(r(0),r(3),r(4)))应该是case (_, ((_, _),(_,_,device))) =>- 注意最后两个元素周围的括号 -和Tuple2对象Tuple3。

于 2015-12-07T20:51:28.403 回答