天富娱乐注册注册网站 Schema怎样实现反射机制推断？

在Windows系统下开发Scala代码，可以使用本地环境测试，因此我们首先需要在本地磁盘准备文本数据文件，这里将HDFS中的/spark/person.txt文件下载到本地D:/spark/person.txt路径下。从文件4-1可以看出，当前数据文件共3列，我们可以非常容易的分析出，这三列分别是编号、姓名、年龄。但是计算机无法像人一样直观的感受字段的实际含义，因此我们需要通过反射机制来推断包含特定类型对象的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 = { val spark : SparkSession = SparkSession.builder() .appName("CaseClassSchema ") .master("local[2]") .getOrCreate() val sc : SparkContext =spark.sparkContext sc.setLogLevel("WARN") val data: RDD[Array[String]] = sc.textFile("D://spark//person.txt").map(x=>x.split(" ")) val personRdd: RDD[Person] = data.map(x=>Person(x(0).toInt,x(1),x(2).toInt)) import spark.implicits._ val personDF: DataFrame = personRdd.toDF personDF.show() personDF.printSchema() println(personDF.filter($"age">30).count()) personDF.createOrReplaceTempView("t_person") spark.sql("select * from t_person").show() spark.sql("select * from t_person where name='zhangsan'").show() 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约束的优势是什么？

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