当前位置:   article > 正文

Spark数据读取和创建_spark创建sc

spark创建sc

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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

从简单数据创建

创建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)
  • 1
  • 2
  • 3
  • 4
创建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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

通过 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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

示例:

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)
)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

df的schema展示:

df.printSchema()
df.schema.printTreeString() //等效
  • 1
  • 2

关于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)
  }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

分析参数:
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文件
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
创建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文件
  • 1
  • 2
  • 3
  • 4
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/木道寻08/article/detail/816382
推荐阅读
相关标签
  

闽ICP备14008679号