Spark数据读取和创建

ss表示sparksession
sc表示sparkContext

//Spark 配置代码（2.0 之前的版本）：
import org.apache.spark._
 val conf = new SparkConf().setAppName("applicationName").setMaster("local") // 本地环境运行
 val sc = new SparkContext(conf)
 val sq= new org.apache.spark.sql.SQLContext(sc)

//2.0之后
import org.apache.spark.sql.SparkSession
val ss = SparkSession.builder().enableHiveSupport().getOrCreate()
val sc = ss.sparkContext
val sq=ss.sqlContext

从简单数据创建

创建rdd

//parallelize[T](seq : scala.Seq[T], numSlices : scala.Int): RDD[T] = { /* compiled code */ }
//numSlices 为分区数，如果不填，Spark会尝试根据集群的状况，来自动设定slices的数目
val ar=Array( (8, "bat"),(64, "mouse"),(-27, "horse"))
 val r1=sc.parallelize(ar)

创建dataframe

//引入隐式转换，使toDF()函数生效
import sq.implicits._

//从Seq创建
val someDF = ar.toSeq.toDF("number", "word")

//从RDD创建
r1.toDF()

//RDD+case class创建，这种方法的好处在于可以指定数据类型
// Define the schema using a case class.
// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,
// you can use custom classes that implement the Product interface.
case class Book(word: String, number: Int)
// Create an RDD of Person objects and register it as a table.
val people = r1.map(p => Book(p(0), p(1)))//转为元素为Person的RDD
                 .toDF()//转换为Dataframe

通过 creatDataFrame()函数创建,主要好处在于可定制schema，包括nullable标志

creatDataFrame()共有7种重载方式：

def createDataFrame[A<: scala.Product](rdd : RDD[A]):DataFrame
def createDataFrame[A<: scala.Product](data : scala.Seq[A]): DataFrame 
//多了一个StructType参数指定Schema，要求输入为RDD[Row]
def createDataFrame(rowRDD : RDD[Row], schema : StructType) : DataFrame
//另外还有以下几种方法，少用，省略。
private[sql] def createDataFrame(rowRDD : org.apache.spark.rdd.RDD[org.apache.spark.sql.Row], schema : StructType, needsConversion : scala.Boolean) : DataFrame 
def createDataFrame(rowRDD : JavaRDD[Row], schema : StructType) : DataFrame
def createDataFrame(rdd : RDD[_], beanClass : scala.Predef.Class[_]) :DataFrame 
def createDataFrame(rdd : JavaRDD[_], beanClass : scala.Predef.Class[_]) :DataFrame

示例：

val someData = Seq(
  Row(8, "bat"),
  Row(64, "mouse"),
  Row(-27, "horse")
)

val someSchema = List(
  StructField("number", IntegerType, true),
  StructField("word", StringType, true)
)

val someDF = spark.createDataFrame(
  spark.sparkContext.parallelize(someData),
  StructType(someSchema)
)

df的schema展示：

df.printSchema()
df.schema.printTreeString() //等效

关于scala.product

从外部数据创建

创建rdd

/** * Read a text file from HDFS, a local file system (available on all nodes), or any * Hadoop-supported file system URI, and return it as an RDD of Strings. */
  def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
    
    assertNotStopped()
    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],minPartitions).map(pair => pair._2.toString).setName(path)
  }

分析参数：
path: String 是一个URI，這个URI可以是HDFS、本地文件（全部的节点都可以），或者其他Hadoop支持的文件系统URI返回的是一个字符串类型的RDD，也就是是RDD的内部形式是Iterator[(String)]
minPartitions= math.min(defaultParallelism, 2) 是指定数据的分区，如果不指定分区，当你的核数大于2的时候，不指定分区数那么就是 2。当你的数据大于128M时候，Spark是为每一个快（block）创建一个分片（Hadoop-2.X之后为128M一个block）

val rdd = sc.textFile(“/home/hadoop/data.txt”)

//SparkSession版本 Spark 2.0及以上
val dataRDD1 = ss.read.csv("path/of/csv/file").rdd //读取csv 文件
val dataRDD2 = ss.read.json("path/of/json/file").rdd //读取json 文件
val dataRDD3 = ss.read.textFile("path/of/text/file").rdd//读取text文件

创建DataFrame

官方详解read

//SparkSession版本 Spark 2.0及以上
val df1 = ss.read.csv("path/of/csv/file")//读取csv 文件
val df2 = ss.read.json("path/of/json/file")//读取json 文件
val df3 = ss.read.textFile("path/of/text/file")//读取text文件