从Spark RDD中提取数据，并在scala中填充元组

Question

我在Hadoop / Spark框架的顶部使用Scala。

实际上我的数据是这种类型的：

RDD[(List[(String, Int)], Long)]

并且，这是此数据记录中前两行的示例：

(List(("COD_LOCALE_PROGETTO",0), ("CUP",1), ("OC_TITOLO_PROGETTO",2), ("OC_SINTESI_PROGETTO",3), ("OC_LINK",4), ("OC_COD_CICLO",5), ("OC_DESCR_CICLO",6), ("OC_COD_TEMA_SINTETICO",7), ("OC_TEMA_SINTETICO",8), ("COD_GRANDE_PROGETTO",9), ("DESCRIZIONE_GRANDE_PROGETTO",10)),0)

(List(("10CAPORTO-POZZUOLI 1",0), ("J86G08000450003",1), ("INTERVENTO C11 2° LOTTO ¿ 1° STRALCIO FUNZIONALE ¿COLLEGAMENTO TRA TANGENZIALE DI NAPOLI (VIA CAMPANA), RETE VIARIA COSTIERA E PORTO DI POZZUOLI""",2), ("INTERVENTO C11 2° LOTTO ¿ 1° STRALCIO FUNZIONALE ¿COLLEGAMENTO TRA TANGENZIALE DI NAPOLI (VIA CAMPANA), RETE VIARIA COSTIERA E PORTO DI POZZUOLI""",3), ("www.opencoesione.gov.it/progetti/10caporto-pozzuoli-1",4), (1,5), ("Ciclo di programmazione 2007-2013",6), ("07",7), ("Trasporti e infrastrutture a rete",8), (" ",9), (" ",10)),1)

在实际情况下，每行有194列，我总共有160多万条记录。

有了这个数据集，我想填充一个新的列表，类型为:

List[(String, Int, Int, Int)]

其中，第一个“Int”是每行的每个字段（COD_LOCALE_PROGETTO、CUP…），第二个字段是每个字段的大小（19、3、…）；第三个字段是已经编码在变量中的每个字段的位置，就在字符串之后；最后一个“Int”是整个数据集中每行的位置。

我尝试了这个脚本：

     | val Dimensione = item._1.size;
     | for(i <- 0 until Dimensione){
     | ComponentiOpenCoesione :+= (item._1(i)._1.replace("\"","").toString,
     | item._1(i)._1.replace("\"","").toString.size,
     | item._1(i)._2.toInt,
     | item._2.toLong)}
     | })

但它失败了，我称之为“ComponentiOpenCoesione”的元组列表没有填充。

最后，这个变量是这样定义的:

var ComponentiOpenCoesione : List[(String, Int, Int, Long)] = List();

有人能帮我吗？如何从RDD中提取数据并加载到列表中？

非常感谢。

Answer 1

在 scala 中，返回函数的最后一个语句。在这里，您的函数将不返回任何内容，因为它的最后一个语句是不返回任何内容的 for 循环。

要纠正它，您只需将ComponentiOpenCoesione作为您的最后一句话。所以，如果你只是打算映射你的RDD[（List[（String， Int）]， Long）]来获取RDD[List[（String， Int， Int， Long）]]，你的代码应该是：

rdd.map(item => {
  var ComponentiOpenCoesione: List[(String, Int, Int, Long)] = List();
  val Dimensione = item._1.size;
  for (i <- 0 until Dimensione) {
    ComponentiOpenCoesione :+= (item._1(i)._1.replace("\"", "").toString,
      item._1(i)._1.replace("\"", "").toString.size,
      item._1(i)._2.toInt,
      item._2.toLong)
  }
  ComponentiOpenCoesione
})

您可以查看Scala中的Return问题的答案，了解如何在Scala中返回值。