创建数据架构的有效方法是什么？

Question

我是个新手，我发现有两种方法可以创建数据框架的模式。我有一个RDD: empRDD和数据(除以"，")

+---+-------+------+-----+
|  1|   Mark|  1000|   HR|
|  2|  Peter|  1200|SALES|
|  3|  Henry|  1500|   HR|
|  4|   Adam|  2000|   IT|
|  5|  Steve|  2500|   IT|
|  6|  Brian|  2700|   IT|
|  7|Michael|  3000|   HR|
|  8|  Steve| 10000|SALES|
|  9|  Peter|  7000|   HR|
| 10|    Dan|  6000|   BS|
+---+-------+------+-----+

val empFile = sc.textFile("emp")
val empData = empFile.map(e => e.split(","))

创建模式的第一种方法是使用case类：

case class employee(id:Int, name:String, salary:Int, dept:String)
val empRDD = empData.map(e => employee(e(0).toInt, e(1), e(2).toInt, e(3)))
val empDF = empRDD.toDF()

第二种方法是使用StructType：

    val empSchema = StructType(Array(StructField("id", IntegerType, true),
                    StructField("name", StringType, true), 
                    StructField("salary", IntegerType, true), 
                    StructField("dept", StringType, true)))
    val empRDD = empdata.map(e => Row(e(0).toInt, e(1), e(2).toInt, e(3)))
    val empDF  = sqlContext.createDataFrame(empRDD, empSchema)

就我个人而言，我更喜欢使用StructType编写代码。但我不知道在实际的行业项目中推荐哪种方式。有人能告诉我最喜欢的方法吗？

Answer 1

您可以使用火花-csv库来读取csv文件，根据我们的要求，这个库有大量的options。

您可以将csv文件读取为

val df = sqlContext.read
    .format("com.databricks.spark.csv")
    .option("header", "true") // Use first line of all files as header
    .option("inferSchema", "true") // Automatically infer data types
    .load("data.csv")

但是，您也可以手动提供架构。

我认为最好的方法是将csv与spark-csv作为dataset读取为

val cities = spark.read
  .option("header", "true")
  .csv(location)
  .as[employee]

阅读dataset相对于rdd和dataframe 这里的优势。

如果已经有了case类，也可以从case类生成架构。

import org.apache.spark.sql.Encoders
val empSchema = Encoders.product[Employee].schema

希望这能有所帮助

Answer 2

在这种情况下，当您从CSV文件(或任何分隔文件)创建RDD时，您可以像@Shankar Koirala所提到的那样自动推断模式。

如果您正在从不同的源创建RDD，那么：

答:当您有较少的字段(少于22个)，您可以创建它使用案例类。

B.当您有超过22个字段时，需要以编程方式创建模式

指向星火编程指南的链接

Answer 3

如果您的输入文件是分隔的文件，您可以使用数据库的火花-csv库。

使用这种方式：

// For spark < 2.0
DataFrame df = sqlContext.read()
            .format("com.databricks.spark.csv")
            .option("header", "true")
            .option("nullValue", "")
            .load("./data.csv");
df.show();

火花2.0；

DataFrame df = sqlContext.read()
            .format("csv")
            .option("header", "true")
            .option("nullValue", "")
            .load("./data.csv");
df.show();

在命令中可以使用option进行大量定制。例如：

1. .option("inferSchema", "true")自动推断每列的数据类型。
2. 定义压缩编解码器的.option("codec", "org.apache.hadoop.io.compress.GzipCodec")
3. .option("delimiter", ",")将分隔符指定为','

Databrick的星火-csv库被移植到spark 2.0中。

使用此库将使您免于分析分隔文件的各种用例的困难。

参考：https://github.com/databricks/spark-csv