如何在不知道数据模式的情况下将数据从文本文件加载到spark数据框架中

Question

我在hadoop中有一个文本文件，我需要使用spark java api使用其第二列进行排序。我正在使用数据帧，但我不确定它的列。它可能有动态列，这意味着我不知道确切的列数。

我该如何继续?请帮帮我。

提前感谢。

Answer 1

首先，我想用scala(不是java)给出一个csv的例子

你可以使用Spark的csv api来创建数据框，并根据你想要的任何列进行排序。如有任何限制，请见下文。

固定列数:

从下面固定列数的例子开始…你可以照着这个例子做。

数据看起来像ebay.csv:

" 95、8213034705、2.927373 jake7870 0 95117。5,xbox, 3"

//  SQLContext entry point for working with structured data
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// this is used to implicitly convert an RDD to a DataFrame.
import sqlContext.implicits._
// Import Spark SQL data types and Row.
import org.apache.spark.sql._
//define the schema using a case class
case class Auction(auctionid: String, bid: Float, bidtime: Float, bidder: String, bidderrate: Integer, openbid: Float, price: Float, item: String, daystolive: Integer)

 val auction = sc.textFile("ebay.csv").map(_.split(",")).map(p => 
Auction(p(0),p(1).toFloat,p(2).toFloat,p(3),p(4).toInt,p(5).toFloat,p(6).toFloat,p(7),p(8).toInt )).toDF()
// Display the top 20 rows of DataFrame 
auction.show()
// auctionid  bid   bidtime  bidder         bidderrate openbid price item daystolive
// 8213034705 95.0  2.927373 jake7870       0          95.0    117.5 xbox 3
// 8213034705 115.0 2.943484 davidbresler2  1          95.0    117.5 xbox 3 …

// Return the schema of this DataFrame
auction.printSchema()
root
 |-- auctionid: string (nullable = true)
 |-- bid: float (nullable = false)
 |-- bidtime: float (nullable = false)
 |-- bidder: string (nullable = true)
 |-- bidderrate: integer (nullable = true)
 |-- openbid: float (nullable = false)
 |-- price: float (nullable = false)
 |-- item: string (nullable = true)
 |-- daystolive: integer (nullable = true)
auction.sort("auctionid") // this will sort first column i.e auctionid

可变列数(因为Case类与Array参数是可能的):

你可以使用下面的伪代码，其中前4个元素是固定的，其余的都是可变数组…

由于您只插入到第二列进行排序，因此这将有效，并且所有其他数据将存在于该特定行的数组中，以供以后使用。

case class Auction(auctionid: String, bid: Float, bidtime: Float, bidder: String, variablenumberofColumnsArray:String*)
 val auction = sc.textFile("ebay.csv").map(_.split(",")).map(p => 
Auction(p(0),p(1).toFloat,p(2).toFloat,p(3),p(4).toInt, VariableNumberOfColumnsArray or any complex type like Map ).toDF()
    auction.sort("auctionid") // this will sort first column i.e auctionid