Spark Scala 2.10 元组限制

Question

我有 66 列要处理的 DataFrame（几乎每个列的值都需要以某种方式更改）所以我正在运行以下语句

    val result = data.map(row=> (
        modify(row.getString(row.fieldIndex("XX"))),
        (...)
        )
    )

直到第 66 列。 由于此版本中的 scala 限制为 22 对的最大元组，因此我不能那样执行。 问题是，有什么解决方法吗？ 在所有行操作之后，我将其转换为具有特定列名的 df

   result.toDf("c1",...,"c66")
   result.storeAsTempTable("someFancyResult")

“修改”功能只是说明我的观点的一个例子

Answer 1

绕过它的方法很繁琐，但它确实有效，试试这个示例代码让你开始吧，你会看到有超过 22 列被访问：

object SimpleApp {
  class Record(val x1: String, val x2: String, val x3: String, ... val x24:String) extends Product with Serializable {
    def canEqual(that: Any) = that.isInstanceOf[Record]

    def productArity = 24

    def productElement(n: Int) = n match {
      case 0 => x1
      case 1 => x2
      case 2 => x3
      ...
      case 23 => x24
    }
  }

  def main(args: Array[String]) {

    val conf = new SparkConf().setAppName("Product Test")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc);

    val record = new Record("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o", "p", "q", "r", "s", "t", "u", "v", "w", "x")

    import sqlContext._
    sc.parallelize(record :: Nil).registerAsTable("records")

    sql("SELECT x1 FROM records").collect()
  }
}

Answer 2

如果您所做的只是修改现有 DataFrame 的值，最好使用 UDF 而不是映射到 RDD：

import org.apache.spark.sql.functions.udf

val modifyUdf = udf(modify)
data.withColumn("c1", modifyUdf($"c1"))

如果由于某种原因上述不符合您的需求，您可以做的最简单的事情是从RDD[Row] 重新创建DataFrame。比如这样：

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructField, StructType, IntegerType}


val result: RDD[Row] = data.map(row => {
  val buffer = ArrayBuffer.empty[Any]

  // Add value to buffer
  buffer.append(modify(row.getAs[String]("c1")))

  // ... repeat for other values

  // Build row
  Row.fromSeq(buffer)
})

// Create schema
val schema = StructType(Seq(
  StructField("c1", StringType, false),
  // ...  
  StructField("c66", StringType, false)
))

sqlContext.createDataFrame(result, schema)