具有来自文本文件的模式匹配的 Scala RDD

Question

对 Spark 有点陌生。所以这就是问题所在，我有一个具有某种通用格式的 txt 文件，可以说： Time : Message 所以我必须实现两件事：RDD 和模式组以及匹配。

将文件作为rdd：

val rdd1 = sc.textFile(location)

构建模式：

private val f1 = "([0-9]*)"
private val f2 = "([:])"
private val f3 = "(.*?)"
private val regex = s"$f1 $f2 $f3"
private val p = Pattern.compile(regex)

现在我想整合这两者，

rdd1.map(//What to do here)

我想检查每一行是否与常规匹配format.If它不想为不匹配的每一行显示错误消息。

如果匹配，我想为上述模式分组。 f1 是组 1，f2 是组 2，f3 是组 three.Finally 我想在 f3（消息字段）中搜索错误，失败等关键字。

我知道要提前问 for.Thx 很多。

我已经尝试过的：

def parseLine(s1: String): Option[Groups] = {
val matcher = p.matcher(s1)
if (matcher.find) {
  println("General Format correct")
  //group
  Some(group(matcher))
  //after format is verified search f3 for error,failure keyword.

}
else {
  println("Format Wrong")
  None
}
}

def group(matcher: Matcher) = {
Line(
  matcher.group(1),
  matcher.group(2),
  matcher.group(3))}

case class Line(Time: String, colon: String, Message: String)

现在我被困在如何迭代 rdd 以将文本文件的每一行传递给 function.if 我将整个 rdd 传递给函数即 RDD[String] type.Other 元素like 匹配器不起作用，因为它需要 String 类型。在阅读有关 rdd 函数的信息时：（如果我错了请纠正我），foreach 方法应该迭代 rdd 但我得到类型不匹配。目前正在尝试地图功能，但还没有得到它。

正如我所说，我是 spark rdd 的新手's.I 不知道使用分区函数是否可以帮助我解决问题。

我真的需要有经验的人指导一下。提供任何帮助。

Answer 1

对于像这样的简单示例，通常使用 RDD 的方式与使用简单的 Scala 序列的方式相同。如果我们将 Spark 排除在外，一种方法是：

import scala.util.{Failure, Success}

val input = List(
  "123 : Message1 Error",
  "This line doesn't conform",
  "345 : Message2",
  "Neither does this one",
  "789 : Message3 Error"
)

val f1 = "([0-9]*)"
val f2 = "([:])"
val f3 = "(.*)"
val regex = s"$f1 $f2 $f3".r

case class Line(a: String, b: String, c: String)


//Use Success and Failure as the functional way of representing
//operations which might not succeed
val parsed = input.map { str =>
  regex.findFirstMatchIn(str).map(m => Line(m.group(1), m.group(2), m.group(3))) match {
    case Some(l) => Success(l)
    case None => Failure(new Exception(s"Non matching input: $str"))
  }
}

//Now split the parsed result so we can handle the two types of outcome separately
val matching = parsed.filter(_.isSuccess)
val nonMatching = parsed.filter(_.isFailure)

nonMatching.foreach(println)

//Filter for only those messages we're interested in
val messagesWithError = matching.collect{
  case Success(l @ Line(_,_,m)) if m.contains("Error") => l
}

messagesWithError.foreach(println)

我们在 Spark 中做这件事有什么不同？不多：

  val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("Example"))

  import scala.util.{Failure, Success}

  val input = List(
    "123 : Message1 Error",
    "This line doesn't conform",
    "345 : Message2",
    "Neither does this one",
    "789 : Message3 Error"
  )

  val f1 = "([0-9]*)"
  val f2 = "([:])"
  val f3 = "(.*)"
  val regex = s"$f1 $f2 $f3".r

  case class Line(a: String, b: String, c: String)

  val inputRDD = sc.parallelize(input)

  //Use Success and Failure as the functional way of representing
  //operations which might not succeed
  val parsedRDD = inputRDD.map { str =>
    regex.findFirstMatchIn(str).map(m => Line(m.group(1), m.group(2), m.group(3))) match {
      case Some(l) => Success(l)
      case None => Failure(new Exception(s"Non matching input: $str"))
    }
  }

  //Now split the parsed result so we can handle the two types of outcome separately
  val matchingRDD = parsedRDD.filter(_.isSuccess)
  val nonMatchingRDD = parsedRDD.filter(_.isFailure)

  //We use collect() to bring the results back from the Spark workers to the Driver
  nonMatchingRDD.collect().foreach(println)

  //Filter for only those messages we're interested in
  val messagesWithError = matchingRDD.collect {
    case Success(l@Line(_, _, m)) if m.contains("Error") => l
  }

  //We use collect() to bring the results back from the Spark workers to the Driver
  messagesWithError.collect().foreach(println)
}

如果生成的数据集非常大，使用 collect() 将结果提供给驱动程序是不合适的，但使用 println 记录结果也不合适。

具有来自文本文件的模式匹配的 Scala RDD

Scala RDD with pattern matching from a text file

regex

scala

apache-spark

rdd