实验SparkSQL编程初级实践_②编程实现将rdd转换为dataframe 源文件内容如下(包含id,name,age): 1,el

实验SparkSQL编程初级实践

实践环境：

1. Oracle VM VirtualBox 6.1.12

2. Ubuntu 16.04

3. Hadoop3.1.3

4. JDK1.8.0_162

5. spark2.4.0

6. python3.5

7. Windows11系统下pycharm2019.1专业版

实验目的：

1. 通过实验掌握Spark SQL的基本编程方法；

2. 熟悉RDD到DataFrame的转化方法；

3. 熟悉利用Spark SQL管理来自不同数据源的数据。

实验内容，步骤与实验结果：

1. Spark SQL 基本操作

将下列JSON格式数据复制到Linux系统中，并保存命名为employee.json。

为employee.json创建DataFrame，并写出Python语句完成下列操作：

import os
os.environ[“JAVA_HOME”]=“/usr/lib/jvm/jdk1.8.0_162”
os.environ[“PYSPARK_PYTHON”]=‘/usr/bin/python3.5’
#import SparkSession
from pyspark.sql import SparkSession
#create spar session object
spark=SparkSession.builder.appName(‘data_processing’).getOrCreate()
# Load csv Dataset
df=spark.read.json(“employee.json”)

1 .查询所有数据；
df.show()

2.查询所有数据，并去除重复的数据；
df.distinct().show()

3.查询所有数据，打印时去除id字段；
df.drop(“id”).show()

4.筛选出age>30的记录；
df.filter(“age”>30).show()

5.将数据按age分组；
df.groupBy(“age”).count().show()

6.将数据按name升序排列；
df.sort(df.name.asc()).show()

7.取出前3行数据；
df.show(3)

8. 查询所有记录的name列，并为其取别名为username；
df.distinct().show()

9.查询年龄age的平均值；
df.agg({“age”:“mean”}).show()

10. 查询年龄age的最小值。
df.agg({“age”:“min”}).show()

2．编程实现将RDD转换为DataFrame

源文件内容如下（包含id,name,age）：

请将数据复制保存到Linux系统中，命名为employee.txt，实现从RDD转换得到DataFrame，并按“id:1,name:Ella,age:36”的格式打印出DataFrame的所有数据。请写出程序代码。

#反射机制 – 针对数据项已知
import os
os.environ[“JAVA_HOME”]=“/usr/lib/jvm/jdk1.8.0_162”
os.environ[“PYSPARK_PYTHON”]=‘/usr/bin/python3.5’

# 导入Spark相关包
from pyspark.sql import Row
from pyspark.sql.types import *
from pyspark import SparkContext,SparkConf
from pyspark.sql import SparkSession

# 构建 spark 单元
spark = SparkSession.builder.config(conf=SparkConf()).getOrCreate()
# 构建表头
schemaString = “id name age”
fields = [StructField(field_name ,StringType(),True) for field_name in schemaString.split(" ")]
schema = StructType(fields)

# 加载数据
filename = “employee.txt”
people= spark.sparkContext.textFile(filename)
# print(people.collect())
# 数据预处理
people_data = people.map(lambda x : x.split(“,”))
# print(people_data.collect())
# 处理为 ROW 对象模式
people_rows = people_data.map(lambda attributes : Row(int(attributes[0]),attributes[1],int(attributes[2])))
# 构建 DataFrame
schemapeople = spark.createDataFrame(people_rows,schema)
# 构建临时表
schemapeople.createOrReplaceTempView(“employee”)
# SQL 查询
DF_people = spark.sql(“select * from employee”)
# DF – RDD
people_rdd = DF_people.rdd.map(lambda p : “id:” + p.id + “,” + “name:” + p.name + “,” + “Age:” + str(p.age))
# print(people_rdd.collect())
# print(people_rdd.collect())
for i in people_rdd.collect():
print(i)

3. 编程实现利用DataFrame读写MySQL的数据

（1）在MySQL数据库中新建数据库sparktest，再创建表employee，包含如表5-2所示的两行数据。

表5-2 employee表原有数据

mysql> create database sparktest;

mysql> show databases;

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);

mysql> select * from employee;



（2）配置Spark通过JDBC连接数据库MySQL，编程实现利用DataFrame插入如表5-3所示的两行数据到MySQL中，最后打印出age的最大值和age的总和。

表5-3 employee表新增数据

关于数据库的相关参数 driver – com.mysql.jdbc.Driver 数据库的JDBC驱动 url – 数据库的连接地址 jdbc:mysql://localhost:3306/spark dbtable – 访问的数据表 student user – 数据库的用户名 test password – 123456 数据库的用户密码
查看数据库内容并插入数据代码如下：
# -*- coding:utf-8 -*-
#反射机制 – 针对数据项已知
import os
os.environ[“JAVA_HOME”]=“/usr/lib/jvm/jdk1.8.0_162”
os.environ[“PYSPARK_PYTHON”]=‘/usr/bin/python3.5’
SUBMIT_ARGS = “–packages mysql:mysql-connector-java:5.1.40 pyspark-shell”
os.environ[“PYSPARK_SUBMIT_ARGS”] = SUBMIT_ARGS

from pyspark.sql import Row
from pyspark.sql.types import *
from pyspark import SparkContext,SparkConf
from pyspark.sql import SparkSession

jdbcStr = “jdbc:mysql://localhost:3306/sparktest”
# 构建 spark 单元
spark = SparkSession.builder.config(conf = SparkConf()).getOrCreate()
# 读取数据库中的数据
jdbcDF=spark.read.format(“jdbc”).option(“url”,jdbcStr).option(“driver”,“com.mysql.jdbc.Driver”).option(“dbtable”,“employee”).option(“user”, “root”).option(“password”, “123456”).load()
# 数据查看
jdbcDF.show()

# 下面设置模式信息
schema = StructType([StructField(“id”, IntegerType(), True),StructField(“name”, StringType(), True),StructField(“gender”, StringType(), True),StructField(“age”, IntegerType(), True)])
# 下面设置两条数据，表示两个学生的信息
studentRDD = spark.sparkContext.parallelize([“3 Mary F 26”,“4 Tom M 23”]).map(lambda x:x.split(“\t”))
# 下面创建Row对象，每个Row对象都是rowRDD中的一行
rowRDD = studentRDD.map(lambda p:Row(int(p[0].strip()), p[1].strip(), p[2].strip(), int(p[3].strip())))
# 建立起Row对象和模式之间的对应关系，也就是把数据和模式对应起来
studentDF = spark.createDataFrame(rowRDD, schema)
studentDF.show()

# 写入数据库
prop = {}
prop[‘user’] = ‘root’
prop[‘password’] = ‘123456’
prop[‘driver’] = “com.mysql.jdbc.Driver”
studentDF.write.jdbc(jdbcStr,‘employee’,‘append’, prop)

# 读取数据库中的数据
jdbcDF=spark.read.format(“jdbc”).option(“url”,jdbcStr).option(“driver”,“com.mysql.jdbc.Driver”).option(“dbtable”,“employee”).option(“user”, “root”).option(“password”, “123456”).load()
# 数据查看
jdbcDF.show()

print(type(jdbcDF))
#查询年龄age的最大值
jdbcDF.agg({“age”:“max”}).show()
#查询年龄age的总和值
jdbcDF.agg({“age”:“sum”}).show()


出现的问题与解决方案：
问题一：spark连接mysql时报错，找不到JDBC。
问题原因：可能是实验前没有将jdbc放入spark的jars里面也可能是代码差点什么。

解决方法：
将适合的jdbc放入spark的jars里面，并将jdbc路径添加在spark-env.sh中。

以上步骤操作后重启后并没有解决。
复制spark-defaults.conf.template文件，修改spark-defaults.conf文件内容。

再次重启尝试依旧报错。在代码中加入如下两行：

运行成功！！！（我也不知道是不是只是因为这两行而执行成功的，但总归解决了）