Spark 与 Scala 中的 ETL 过程

Posted 2023-04-18

【中文标题】Spark 与 Scala 中的 ETL 过程

【英文标题】：ETL process in Spark with Scala

【发布时间】：2015-09-28 05:46:07

【问题描述】：

我正在使用 scala 在 Spark 中进行 ETL 过程。 原始日志文件包含两列名称和年龄。我的 ETL 过程读取并验证原始日志并生成另外两列，即标志和验证消息。 标志列指定行是否有效。（如果行有效= 1 else 0） 验证列指定行无效的原因。

例如。 原始日志文件

RAM,35
SAM,45
JAM,ww

这里最后一行是无效的，所以我的最终输出将是

RAM,35,1,""
SAM,45,1,""
JAM,ww,0,"invalid age"

我的scala代码是

import sqlContext._

val people = sc.textFile("hdfs://..../rawT.csv").map(_.split(","))

val base_people = people.mapr => val formatted_people = if(r(1).matches("^\\d*$")) ("1","") else ("0","Invalid Age");  (r(0), r(1), formatted_people)

但在这段代码中，RDD 生成为 base_people: org.apache.spark.rdd.RDD[(String, String, (String, String))]

我想要没有字符串数组的 RDD， base_people: org.apache.spark.rdd.RDD[(String, String, String, String)]

我的逻辑是正确的还是我需要遵循一些其他的代码。

【参考方案1】：

您正在创建一个三元组，最后一个位置有一对，而不是四元组。怎么样：

val base_people = people
  .mapr => if(r(1).matches("^\\d*$"))
              (r(0), r(1), "1", "") 
            else 
              (r(0), r(1), "0", "Invalid Age")

【参考方案2】：

有很多方法可以实现您想要实现的目标。以下是我的做法：

val base_people = data.map
  r => r(1) match 
   case i:Int => (r(0), r(1), "1", "") 
   case s:String => (r(0), r(1), "0", "invalid age")
  

我将在第二个字段上使用带有案例类（Int、String、Double 等）的模式匹配。你可以使用任何适合你的课程。

我希望这会有所帮助！

【讨论】：

如果您认为答案正确，您可以验证答案并投票