opebet网页版-bet亚洲官网手机版-登录

教育行业A股IPO第一股(股票代码 003032)

全国咨询/投诉热线:400-618-4000

传智教育|传智播客

一样的教育,不一样的品质

全国校区

 

Schema怎样实现反射机制推断?

更新时间:2021年03月23日17时09分 来源:传智教育 浏览次数:


在Windows系统下开发Scala代码,可以使用本地环境测试,因此大家首先需要在本地磁盘准备文本数据文件,这里将HDFS中的/spark/person.txt文件下载到本地D:/spark/person.txt路径下。从文件4-1可以看出,当前数据文件共3列,大家可以非常容易的分析出,这三列分别是编号、姓名、年龄。但是计算机无法像人一样直观的感受字段的实际含义,因此大家需要通过反射机制来推断包含特定类型对象的Schema信息。


Schema的开发


接下来大家打开IDEA开发工具,创建名为“spark_chapter04”的Maven工程,讲解实现反射机制推断Schema的开发流程。

1.添加Spark SQL依赖

在pom.xml文件中添加Spark SQL依赖,代码片段如下所示。

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.11</artifactId>
  <version>2.3.2</version>
</dependency>


2.编写代码

实现反射机制推断Schema需要定义一个case class样例类,定义字段和属性,样例类的参数名称会被利用反射机制作为列名,编写代码如文件1所示。

文件1 CaseClassSchema.scala

   import org.apache.spark.SparkContext
   import org.apache.spark.rdd.RDD
   import org.apache.spark.sql.{DataFrame, Row, SparkSession}
   //定义样例类
   case class Person(id:Int,name:String,age:Int)
   object CaseClassSchema {
     def main(args: Array[String]): Unit = {
       //1.构建SparkSession
       val spark : SparkSession = SparkSession.builder()
                       .appName("CaseClassSchema ")
                       .master("local[2]")
                       .getOrCreate()
      //2.获取SparkContext
      val sc : SparkContext =spark.sparkContext
      //设置日志打印级别
      sc.setLogLevel("WARN")
      //3.读取文件
      val data: RDD[Array[String]] =
          sc.textFile("D://spark//person.txt").map(x=>x.split(" "))
      //4.将RDD与样例类关联
      val personRdd: RDD[Person] = 
                 data.map(x=>Person(x(0).toInt,x(1),x(2).toInt))
      //5.获取DataFrame
      //手动导入隐式转换
      import spark.implicits._
      val personDF: DataFrame = personRdd.toDF
      //------------DSL风格操作开始-------------
      //1.显示DataFrame的数据,默认显示20行
      personDF.show()
      //2.显示DataFrame的schema信息
      personDF.printSchema()
      //3.统计DataFrame中年龄大于30的人数
      println(personDF.filter($"age">30).count())
      //-----------DSL风格操作结束-------------
      //-----------SQL风格操作开始-------------
      //将DataFrame注册成表
      personDF.createOrReplaceTempView("t_person")
      spark.sql("select * from t_person").show()
      spark.sql("select * from t_person where name='zhangsan'").show()
      //-----------SQL风格操作结束-------------
      //关闭资源操作
      sc.stop()
      spark.stop()
    }

在文件1中,第5行代码表示定义了一个Person的Case类,这是因为在利用反射机制推断RDD模式时,首先需要定义一个Case类,因为Spark SQL能够自动将包含Case类的RDD隐式转换成DataFrame,Case类定义了Table的结构,Case类的属性通过反射机制变成表的列名。第9-14行代码中通过SparkSession.builder()方法构建名为“spark”的SparkSession对象,并通过spark对象获取SparkContext。第18-26行代码中,通过sc对象读取文件,系统会将文件加载到内存中生成一个RDD,将RDD 与case class Person进行匹配,personRdd对象即为RDD【Person】,toDF()方法是将RDD转换为DataFrame,在调用toDF()方法之前需要手动添加“spark.implicits._”包。第29-39行代码表示当前创建DataFrame对象后,使用DSL和SQL两种语法操作风格进行数据查询。DataFrame操作和之前在Spark-Shell操作示例大致相同,因此这里将不再展示实行效果。



猜你喜欢:

Scala的方法和函数先容【大数据文章】

Sequence File是什么?简单先容几种文件储存格式

Spark SQL架构的工作原理和工作流程是什么?

Schema约束的优势是什么?

传智教育大数据项目开发培训

opebet网页版|bet亚洲官网手机版

XML 地图 | Sitemap 地图