映射函数写入全局 spark rdd

Tricky 发表于 Dev

棘手的

我有一个字符串RDD。每行对应各种日志。

我在一个函数中有多个正则表达式，它们匹配/区分 RDD 的行以应用调整后的正则表达式。

我想将这个独特的函数映射到我的 RDD 上，这样它就可以快速处理每一行，并将处理过的每一行存储在另一个全局 rdd 中。

问题是，因为我希望这个任务被并行化，所以我的全局 RDD 必须可以同时访问以添加每个处理过的行。

我想知道是否有其他方法可以做到这一点或任何事情！我正在寻求提高我的火花技能。

例如，这就是我想做的：

我有一个像这样的txt：

错误：哈哈哈 param_error=8 param_err2=https

警告：呼呼呼呼 param_warn = tchu param_warn2 = wifi

例如，我的正则表达式函数会将包含“ERROR”的行与数组匹配 Array("Error","8","https")

另一个正则表达式函数会将包含“WARNING”的行与数组匹配，例如 Array("Warning","tchu","wifi")

最后，我想RDD[Array[String]]为处理的每一行获取一个。

我如何保持它与 Spark 并行化？

扎克·佐哈尔

首先，重要的是要了解 Spark 中没有像“全局 RDD”这样的东西，也没有理由需要这样的东西。使用 Spark 时，您应该考虑将一个 RDD 转换为另一个 RDD，而不是更新RDD（这是不可能的 - RDD 是不可变的）。每个这样的转换将由 Spark 分布式（并行）执行。

在这种情况下，如果我正确理解您的要求，您希望将map每个记录转换为以下结果之一：

一Array[String]，其中第一项是"ERROR"，或：
一Array[String]，其中第一项是"WARNING"，或：
如果没有与记录匹配的模式，则将其删除

为此，您可以使用map(f)和collect(f)方法RDD：

// Sample data:
val rdd = sc.parallelize(Seq(
  "ERROR : Hahhaha param_error=8 param_err2=https",
  "WARNING : HUHUHUHUH param_warn=tchu param_warn2=wifi",
  "Garbage - not matching anything"
))

// First we can split in " : " to easily identify ERROR vs. WARNING 
val splitPrefix = rdd.map(line => line.split(" : "))

// Implement these parsing functions as you see fit; 
// The input would be the part following the " : ", 
// and the output should be a list of the values (not including the ERROR / WARNING) 
def parseError(v: String): List[String] = ??? // example input: "Hahhaha param_error=8 param_err2=https"
def parseWarning(v: String): List[String] = ??? // example input: "HUHUHUHUH param_warn=tchu param_warn2=wifi"

// Now we can use these functions in a pattern-matching function passed to RDD.collect,
// which will transform each value that matches one of the cases, and will filter out 
// values that don't match anything
val result: RDD[List[String]] = splitPrefix.collect {
  case Array(l @ "ERROR", v) => l :: parseError(v)
  case Array(l @ "WARNING", v) => l :: parseWarning(v)
  // NOT adding a default case, so records that didn't match will be removed
}    

// If you really want Array[String] and not List[String]:    
val arraysRdd: RDD[Array[String]] = result.map(_.toArray)

本文收集自互联网，转载请注明来源。

如有侵权，请联系 [email protected] 删除。