我无法定义下面的函数。
首先，moviepairs是一个spark rdd，包含如下数据：
[（“蜘蛛侠”，“睡眠博士”），（“蜘蛛侠”，“丛林书”），（“木兰”，“丛林书”）]
其次，movieratings rdd包含一个电影标题，后跟一个用户分级列表：
格式：（电影名称），[用户分级列表，索引对]即（分级1，索引0），（分级2，索引1）
[（“蜘蛛侠”，[[4，0]，[3，1]，[3，2]]，（“丛林书”，[[4，0]，[3，1]，[3，2]]）]
当我运行下面的代码时，最后一行，交叉操作，给了我一个错误：

val result = moviePairs.map{val filteredMovieList1 = (movie1 : String, movie2 : String) => 
                                            {movieRatings.filter(mRating => mRating._1 == movie1)
                                            .map(mRating => mRating._2).first();}

                                            val filteredMovieList2 = (movie1 : String, movie2 : String) =>  
                                            {movieRatings.filter(mRating => mRating._1 == movie2)
                                            .map(mRating => mRating._2).first();}

                                            (movie1 : String, movie2 : String) => filteredMovieList1.intersection(filteredMovieList2)
                                            }

错误消息：

Task4.scala:29: error: type mismatch;
 found   : (String, String) => (String, String) => Array[(Int, Int)]
 required: ((String, String)) => ?
                                            (movie1 : String, movie2 : String) => filteredMovieList1

我的第一个问题是，如何修复代码，以便不将filteredmovielist1和filteredmovielist2定义为函数？我只需要他们返回的结果。
我的第二个问题是，我是否正确定义了多行函数？我习惯于在spark中编写一行lambda函数，所以这样做对我来说是非常新的。
我之所以编写这一多行函数并将结果存储在许多中间变量中，唯一的原因是因为之前我尝试了这一行：
moviepairs=>movieratings.filter（mrating=>mrating.\u 1==moviepairs.\u 1）.map（mrating=>mrating.\u 2）.intersection（moviepairs=>movieratings.filter（mrating=>mrating.\u 1==moviepairs.\u 2）.map（mrating=>mrating.\u 2））；
但是我得到了一个运行时错误，它暗示了这样的情况：您不能在另一个转换中执行转换或操作（即，我在交集操作中执行转换）。