将RDD[String]拆分为RDD[tuples]

Question

我是Scala和RDD的初学者。我在Spark2.4上使用Scala。我有一个RDDString，上面有这样的行：

(a, b, c, d, ...)

我想在每次昏迷时把这条线分开，得到一个RDD[(String, String, String, ...)]。

显然，下面这样的解决方案对于元素的数量是不可能的。

rdd.map(x => (x.split(",")(0), x.split(",")(1), x.split(",")(2)))

也许有一种方法可以实现自动化？一切都会正常的。

尽管我努力了，但到目前为止我还没有解决我的问题，

非常感谢!

Answer 1

请注意，最大元组大小限制为22，所以列出它们的时间不会太长.

顺便说一句，在第110页的“在行动中的火花”一书中，它写道：

没有优雅的方法将数组转换为元组，因此您必须使用这个丑陋的表达式：

scala> val itPostsRDD = itPostsSplit.map(x => (x(0), x(1), x(2), x(3), x(4), x(5), x(6), x(7), x(8), x(9), x(10), x(11), x(12))
itPostsRDD: org.apache.spark.rdd.RDD[(String, String, ...

Answer 2

如果元素的数量是固定的，则可以执行以下操作：

val tuples =
  rdd
    .map(line => line.replaceAll("[\\(\\)]", "").split(","))
    .collect {
      case Array(col1, col2, ..., coln) => (col1, col2, ..., coln)
    }
// tuples: RDD[(String, String, ..., String)]

Answer 3

一种解决方案是只编写映射函数：

def parse(s: String) = s.split(",") match {
    case Array(a,b,c) => (a,b,c)
}

parse("x,x,x") // (x,x,x)

您可以使用无形状的方法编写更通用的解决方案：

def toTuple[H <: HList](s: String)(implicit ft: FromTraversable[H], t: Tupler[H]) = s.split(",").toHList[H].get.tupled

然后您可以直接使用它：

toTuple[String :: String :: String :: HNil]("x,x,x") // (x,x,x)
toTuple[String :: String :: HNil]("x,x") // (x,x)

或者修复，然后键入并使用它：

def parse3(s: String) = toTuple[String :: String :: String :: HNil](s)

parse3("x,x,x") // (x,x,x)