在spark sql中,我们知道操作的数据都是DataFrame,因此从这篇文章开始,我们介绍几种创建DataFrame的方式,这篇文章介绍从文件中读取数据创建DataFrame。
其实从文件中读取文件创建DataFrame的案例我们再签名已经演示过了,所以这里简单的介绍一下即可:
package org.example import org.apache.spark.SparkConf import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType} object StudentCal { def main(args: Array[String]): Unit = { val conf = new SparkConf() //这里的master模式,在提交的时候我们可以使用命令进行修改,但是在自动化部署提交的平台里面我们不能随意去添加各种命令,因此在这里设置即可。 conf.set("spark.master", "local[*]") //这里是配置spark应用程序执行的cpu核数 conf.set("spark.executor.cores", "2") //这里是配置spark应用程序执行的时候堆大小 conf.set("spark.executor.memory", "4g") //这里是配置spark应用程序执行的时候需要多少个Executor进程来执行整个任务 conf.set("spark.executor.instances", "6") //这里是设置spark应用程序数据本地化等待时长,这个意思代表数据本地化等待时长,spark driver对于application分配的task尽量在数据节点上, //这里补充一下,设置这个时间,也就是等待driver把task分配到数据所处的节点上,如果超过这个时长,则数据会被分配到就近的一个节点上。 conf.set("spark.locality.wait", "0") //这里是spark的序列化信息设置 conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); val session = SparkSession.builder() //把上面的conf设置进来 .config(conf) .appName("demo") .getOrCreate() val sc = session.sparkContext // 数据源文件 val file = "C:\\Users\\Administrator\\Desktop\\fsdownload\\users.txt" val usersDF = session.read.format("csv") .option("sep", ";") // 字段使用;分隔符 .option("inferSchema", "true") // 指定自动推断模式 .option("samplingRatio", 0.001) // 根据抽样进行模式推断 .option("header", "true") // 说明有标题行 .load(file) //打印dataframe的schema信息。 usersDF.printSchema() //再打印下userDF的整个dataframe表信息 usersDF.show() } case class User(name:String,age:Long,sex:String){ } }
备注:
1、spark对于文件中加载数据有非常完善的api
2、在遇到大文件的话,建议指定schema,同时关闭自动推荐的模式。
还没有评论,来说两句吧...