吴桥县网站建设公司,wordpress后台添加底部菜单,网站运营需要服务器吗,做电子请柬用什么网站目录
一、目的与要求
二、实验内容
三、实验步骤
1、Spark SQL基本操作
2、编程实现将RDD转换为DataFrame
3、编程实现利用DataFrame读写MySQL的数据
四、结果分析与实验体会 一、目的与要求
1、通过实验掌握Spark SQL的基本编程方法#xff1b; 2、熟悉RDD到DataFram…目录
一、目的与要求
二、实验内容
三、实验步骤
1、Spark SQL基本操作
2、编程实现将RDD转换为DataFrame
3、编程实现利用DataFrame读写MySQL的数据
四、结果分析与实验体会 一、目的与要求
1、通过实验掌握Spark SQL的基本编程方法 2、熟悉RDD到DataFrame的转化方法 3、熟悉利用Spark SQL管理来自不同数据源的数据。
二、实验内容
1、Spark SQL基本操作 将下列JSON格式数据复制到Linux系统中并保存命名为employee.json。 { id:1 , name:Ella , age:36 } { id:2, name:Bob,age:29 } { id:3 , name:Jack,age:29 } { id:4 , name:Jim,age:28 } { id:5 , name:Damon } { id:5 , name:Damon } 为employee.json创建DataFrame并写出Python语句完成下列操作 1查询所有数据 2查询所有数据并去除重复的数据 3查询所有数据打印时去除id字段 4筛选出age30的记录 5将数据按age分组 6将数据按name升序排列 7取出前3行数据 8查询所有记录的name列并为其取别名为username 9查询年龄age的平均值 10查询年龄age的最小值。
2、编程实现将RDD转换为DataFrame 源文件内容如下包含id,name,age 1,Ella,36 2,Bob,29 3,Jack,29 请将数据复制保存到Linux系统中命名为employee.txt实现从RDD转换得到DataFrame并按“id:1,name:Ella,age:36”的格式打印出DataFrame的所有数据。请写出程序代码。
3、编程实现利用DataFrame读写MySQL的数据
1在MySQL数据库中新建数据库sparktest再创建表employee包含如表所示的两行数据。 2配置Spark通过JDBC连接数据库MySQL编程实现利用DataFrame插入如表所示的三行数据到MySQL中最后打印出age的最大值和age的总和。 三、实验步骤
1、Spark SQL基本操作 将下列JSON格式数据复制到Linux系统中并保存命名为employee.json。 { id:1 , name:Ella , age:36 } { id:2, name:Bob,age:29 } { id:3 , name:Jack,age:29 } { id:4 , name:Jim,age:28 } { id:5 , name:Damon } { id:5 , name:Damon } 为employee.json创建DataFrame并写出Python语句完成下列操作 sparkSparkSession.builder.getOrCreate()df spark.read.json(file:///home/zhc/mycode/sparksql/employee.json)
1查询所有数据 df.show() 2查询所有数据并去除重复的数据 df.distinct().show() 3查询所有数据打印时去除id字段 df.drop(id).show() 4筛选出age30的记录 df.filter(df.age 30).show()5将数据按age分组 df.groupBy(age).count().show() 6将数据按name升序排列 df.sort(df.name.asc()).show() 7取出前3行数据 df.take(3) 8查询所有记录的name列并为其取别名为username df.select(df.name.alias(username)).show() 9查询年龄age的平均值 df.agg({age: mean}).show() 10查询年龄age的最小值。 df.agg({age: min}).show() 2、编程实现将RDD转换为DataFrame 源文件内容如下包含id,name,age 1,Ella,36 2,Bob,29 3,Jack,29 请将数据复制保存到Linux系统中命名为employee.txt实现从RDD转换得到DataFrame并按“id:1,name:Ella,age:36”的格式打印出DataFrame的所有数据。请写出程序代码。
首先在“/home/zhc/mycode/sparksql”目录下创建文件employee.txt
[rootbigdata sparksql]# vi employee.txt
然后在该目录下新建一个py文件命名为rddtodf.py然后写入如下py程序
[rootbigdata sparksql]# vi rddtodf.py
#/home/zhc/mycode/sparksql/rddtodf.py
from pyspark.conf import SparkConf
from pyspark.sql.session import SparkSession
from pyspark import SparkContext
from pyspark.sql.types import Row
from pyspark.sql import SQLContext
if __name__ __main__:sc SparkContext(local,Simple App)sparkSparkSession(sc)peopleRDD spark.sparkContext.textFile(file:home/zhc/mycode/sparksql/employee.txt)rowRDD peopleRDD.map(lambda line : line.split(,)).map(lambda attributes : Row(int(attributes[0]),attributes[1],int(attributes[2]))).toDF()rowRDD.createOrReplaceTempView(employee)personsDF spark.sql(select * from employee)personsDF.rdd.map(lambda t : id:str(t[0]),Name:t[1],age:str(t[2])).foreach(print)最后运行该程序
[rootbigdata sparksql]# python3 rddtodf.py 3、编程实现利用DataFrame读写MySQL的数据
1在MySQL数据库中新建数据库sparktest再创建表employee包含如表所示的两行数据。 2配置Spark通过JDBC连接数据库MySQL编程实现利用DataFrame插入如表所示的三行数据到MySQL中最后打印出age的最大值和age的总和。 首先启动mysql服务并进入到mysql数据库中
[rootbigdata sparksql]# systemctl start mysqld.service
[rootbigdata sparksql]# mysql -u root -p然后开始接下来的操作。
1在MySQL数据库中新建数据库sparktest再创建表employee包含如表所示的两行数据。
mysql create database sparktest;
mysql use sparktest;
mysql create table employee (id int(4), name char(20), gender char(4), age int(4));
mysql insert into employee values(1,Alice,F,22);
mysql insert into employee values(2,John,M,25); 2配置Spark通过JDBC连接数据库MySQL编程实现利用DataFrame插入如表所示的三行数据到MySQL中最后打印出age的最大值和age的总和。
首先在“/home/zhc/mycode/sparksql”目录下面新建一个py程序并命名为mysqltest.py。
[rootbigdata sparksql]# vi mysqltest.py
接着写入如下py程序
#/home/zhc/mycode/sparksql/mysqltest.py
from pyspark.sql import Row
from pyspark.sql.types import *
from pyspark import SparkContext,SparkConf
from pyspark.sql import SparkSession
spark SparkSession.builder.config(conf SparkConf()).getOrCreate()
#下面设置模式信息
schema StructType([StructField(id,IntegerType(),True),StructField(name, StringType(), True),StructField(gender, StringType(), True),StructField(age,IntegerType(), True)])
employeeRDD spark.sparkContext.parallelize([3 Mary F 26,4 Tom M 23,5 zhanghc M 21]).map(lambda x:x.split( ))
#下面创建Row对象每个Row对象都是rowRDD中的一行
rowRDD employeeRDD.map(lambda p:Row(int(p[0].strip()), p[1].strip(), p[2].strip(), int(p[3].strip())))
#建立起Row对象和模式之间的对应关系也就是把数据和模式对应起来
employeeDF spark.createDataFrame(rowRDD, schema)
#写入数据库
prop {}
prop[user] root
prop[password] MYsql123!
prop[driver] com.mysql.jdbc.Driver
employeeDF.write.jdbc(jdbc:mysql://localhost:3306/sparktest?useSSLfalse,employee,append, prop)
employeeDF.collect()
employeeDF.agg({age: max}).show()
employeeDF.agg({age: sum}).show()然后直接运行该py程序即可得到结果
[rootbigdata sparksql]# python3 mysqltest.py 最后到MySQL Shell中即可查看employee表中的所有信息。
mysql select * from employee; 四、结果分析与实验体会 Spark SQL是Apache Spark中用于处理结构化数据的模块。它提供了一种类似于SQL的编程接口可以用于查询和分析数据。通过实验掌握了Spark SQL的基本编程方法SparkSession支持从不同的数据源加载数据并把数据转换成DataFrame并且支持把DataFrame转换成SQLContext自身中的表然后使用SQL语句来操作数据。 在使用Spark SQL之前需要创建一个SparkSession对象。可以使用SparkSession的read方法加载数据。可以使用DataFrame的createOrReplaceTempView方法将DataFrame注册为一个临时视图。可以使用SparkSession的sql方法执行SQL查询。除了使用SQL查询外还可以使用DataFrame的API进行数据操作和转换。可以使用DataFrame的write方法将数据写入外部存储。在使用完SparkSession后应该调用其close方法来关闭SparkSession。 最后还掌握了RDD到DataFrame的转化方法并可以利用Spark SQL管理来自不同数据源的数据。
