如何在Spark中向数据集添加模式？

我正在尝试将文件加载到spark中.如果我将正常的textFile加载到Spark中,如下所示:

val partFile = spark.read.textFile("hdfs://quickstart:8020/user/cloudera/partfile")

结果是:

partFile: org.apache.spark.sql.Dataset[String] = [value: string]

我可以在输出中看到一个数据集.但是如果我加载一个Json文件:

val pfile = spark.read.json("hdfs://quickstart:8020/user/cloudera/pjson")

结果是具有现成模式的数据框:

pfile: org.apache.spark.sql.DataFrame = [address: struct, age: bigint ... 1 more field]

Json/parquet/orc文件有架构.所以我可以理解这是Spark版本的一个特性:2x,这使得事情变得更容易,因为我们在这种情况下直接获得DataFrame,而对于普通的textFile,你得到的数据集中没有任何架构是有意义的.我想知道的是如何将模式添加到数据集中,该数据集是将textFile加载到spark中的结果.对于RDD,有一个case类/ StructType选项来添加模式并将其转换为DataFrame.谁能让我知道我该怎么做？

使用时textFile,文件的每一行都将是数据集中的字符串行.要使用模式转换为DataFrame,您可以使用toDF:

val partFile = spark.read.textFile("hdfs://quickstart:8020/user/cloudera/partfile")

import sqlContext.implicits._
val df = partFile.toDF("string_column")

val partFile = spark.read.option("header", "true").option("delimiter", ";").csv("hdfs://quickstart:8020/user/cloudera/partfile")

val partFile = spark.read.textFile("hdfs://quickstart:8020/user/cloudera/partfile")

val processedDataset: Dataset[(Int, Int)] = partFile.map {
  line: String => (line(0).toInt, line(3).toInt)
}

import sqlContext.implicits._
val df = processedDataset.toDF("value0", "value3")

case class MyRow(value0: Int, value3: Int)

val partFile = spark.read.textFile("hdfs://quickstart:8020/user/cloudera/partfile")

val processedDataset: Dataset[MyRow] = partFile.map {
  line: String => MyRow(line(0).toInt, line(3).toInt)
}

import sqlContext.implicits._
val df = processedDataset.toDF

root
 |-- value0: integer (nullable = true)
 |-- value3: integer (nullable = true)