1

我是 scala(2.11) 和 spark (1.6.0) 的新程序员，他试图在没有 spark-csv 包的情况下将 RDD 转换为 DF（用于练习，但也因为一些技术问题）。在阅读了 Spark 的入门指南和 stackoverflow 的所有相关帖子之后，我不知道如何使一些方法（4）起作用——只有一个对我有用，我不知道为什么——：

对他们中的任何一个的每一次帮助都将是惊人的！

我有一个简单的表格，例如 txt 文件：

Jorgito 10 1 Soltero
Juanito 20 2 Casado
Jaimito 30 3 Divociado

我编写了一些初步代码：

var RDD_filas = RDD_datos.map(_.split("\t"))
var esquema = new StructType()
.add("Nombre", StringType)
.add("Edad", IntegerType)
.add("Hijos",IntegerType)
.add("EC",StringType)

import org.apache.spark.sql._
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType};
import org.apache.spark.sql.SQLContext

case class X(Nombre: String, Edad: Int, Hijos: Int, EC: String)

然后，我应用了我见过的所有不起作用的方法：

var DF_datos = RDD_filas.map({case Array(s0, s1, s2, s3) => X(s0, s1.trim.toInt, s2.trim.toInt, s3)}).toDF("Nombre","Edad","Hijos","EC")
var DF_datos2 = RDD_filas.map(p => X(p(0), p(1).trim.toInt,p(2).trim.toInt,p(3))).toDF("Nombre","Edad","Hijos","EC")
var DF_datos3 = RDD_filas.map(Array(s0, s1, s2, s3) => Array(s0, s1.trim.toInt, s2.trim.toInt, s3)).toDF("Nombre","Edad","Hijos","EC")
var DF_datos4 = sqlContext.createDataFrame(RDD_filas,esquema)

前三种方法允许我创建 DF 并打印它们的模式，但它们没有标题（DF_datos.header() 返回第一行），如果我尝试 DF_datos.show() 会出错一个（对我来说）是第 4 位，因为它应该是最“规范”的方式。

只有这对我有用：

var a = RDD_datos.map(_.split(" ")).take(3)
val rdd = sc.makeRDD(a)
val df = rdd.map {case Array(s0, s1, s2, s3) => X(s0, s1.toInt, s2.toInt, s3)}.toDF()
4

1 回答 1

3

为了使用sqlContext.createDataFrame，您需要有一个RDD[Row]位置，您的行中的条目类型对应于您的架构中的类型。因此，您需要在适当的时候将一些条目从 转换String为。Int

这是一个例子：

scala> val data = sc.textFile("./junk/dat.txt")
data: org.apache.spark.rdd.RDD[String] = ./junk/dat.txt MapPartitionsRDD[20] at textFile at <console>:28

scala> data.foreach{println}
Jorgito 10 1 Soltero
Juanito 20 2 Casado
Jaimito 30 3 Divociado

scala> :pa
// Entering paste mode (ctrl-D to finish)

var esquema = new StructType()
.add("Nombre", StringType)
.add("Edad", IntegerType)
.add("Hijos",IntegerType)
.add("EC",StringType)

// Exiting paste mode, now interpreting.

esquema: org.apache.spark.sql.types.StructType = StructType(StructField(Nombre,StringType,true), StructField(Edad,IntegerType,true), StructField(Hijos,IntegerType,true), StructField(EC,StringType,true))

scala> val rowRDD = data.map(l => l.split(" ")).map{case Array(a,b,c,d) => Row(a, b.toInt, c.toInt, d)}
rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[22] at map at <console>:30

scala> val df = sqlContext.createDataFrame(rowRDD, esquema)
df: org.apache.spark.sql.DataFrame = [Nombre: string, Edad: int ... 2 more fields]

scala> df.show
+-------+----+-----+---------+
| Nombre|Edad|Hijos|       EC|
+-------+----+-----+---------+
|Jorgito|  10|    1|  Soltero|
|Juanito|  20|    2|   Casado|
|Jaimito|  30|    3|Divociado|
+-------+----+-----+---------+

但是，这是很多工作，因此我建议您 a) 升级到 Spark 2.0 并使用内置加载程序或 b) 查看. 在这两种情况下，您只需将分隔符设置为或根据需要。DataFrameReader csvspark-csv\s\t

于 2017-01-04T18:02:11.223 回答