在Spark Scala中将RDD[(String，String，String)]转换为RDD[(String，(String，String))]

Question

有两个rdd，我正在尝试加入:当每个rdd中有2个参数时，它将被联接，但是当我在existingGTINs rdd中添加新参数时，我面临以下错误：

代码如下：

newGTS.collect()
(00070137115045,00070137115045)
(00799999150451,00799999150451)

existingGTS.collect()
(00799999150451,(00003306-808b-46da-bc7f-419c5ae223a7,2016-10-10 10:23:12.0))
(00016700000653,(00006d79-94ea-4651-be0c-0ce77958cd45,2021-05-31 01:20:39.291))
(00923846453024,(0000704b-b40d-4b9e-b266-f7c66723df0e,null))
(00610074049265,(0000a7a1-587c-4b13-a155-7846df82fdee,2020-03-20 12:16:55.873))
(00034100516079,(0002495f-6084-49dd-aadb-20cd137d9694,null))


val join1 = newGTINs.leftOuterJoin(existingGTINs) mapValues {
      case (gtin, iUUID, createDt) => (iUUID.isEmpty, iUUID.getOrElse(UUID.randomUUID.toString))
    }


 error: constructor cannot be instantiated to expected type;
 found   : (T1, T2, T3)
 required: (String, Option[(String, String)])
                 case (gtin, iUUID, createDt) => (iUUID.isEmpty, iUUID.getOrElse(UUID.randomUUID.toString))
                      ^

PS: UUID.randomUUID.toString -->此函数用于创建随机id

Answer 1

我猜测join中使用的newGTINs和existingGTINs应该与使用集合显示的newGTS和existingGTS相同。

因为您的newGTSINs看起来是一个RDD[(String, String)]，而existingGTINS是一个RDD[(String, (String, String))]，所以您的newGTINs.leftOuterJoin(existingGTINs)将是一个RDD[(String,(String, Option[(String, String)]))]。

这意味着您的mapValues将需要一个函数(String, Option[(String, String)]) => SomeNewType或作为参数。它还可以接受满足相似类型语义的部分函数。

但是您的{ case (gtin, iUUID, createDt) => (iUUID.isEmpty, iUUID.getOrElse(UUID.randomUUID.toString)) }是一个与(String, String, String) => SomeNewType类型相对应的部分函数。

注意不同之处，因此产生了错误。您可以通过提供适当的部分函数来满足mapValues要求来修复此问题。

val join1 = 
  newGTINs
    .leftOuterJoin(existingGTINs)
    .mapValues {
      case (gtin, Some(iUUID, createDt)) =>
        (iUUID.isEmpty, iUUID.getOrElse(UUID.randomUUID.toString))
      case (gtin, None) =>
        // what heppens for gtins without matching element in existing one's
        (true, UUID.randomUUID.toString)
    }