0


实验SparkSQL编程初级实践

实验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。
{ “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”:4 , “name”:“Jim”,“age”:28 } { “id”:5 , “name”:“Damon” } { “id”:5 , “name”:“Damon” }
为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):
1,Ella,36 2,Bob,29 3,Jack,29
请将数据复制保存到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表原有数据
id name gender Age 1 Alice F 22 2 John M 25
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表新增数据
id name gender age 3 Mary F 26 4 Tom M 23
关于数据库的相关参数 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文件内容。
在这里插入图片描述
再次重启尝试依旧报错。在代码中加入如下两行:
在这里插入图片描述
运行成功!!!(我也不知道是不是只是因为这两行而执行成功的,但总归解决了)

标签: spark mysql sql

本文转载自: https://blog.csdn.net/Alicia_LX/article/details/124542830
版权归原作者 小李今天工地加班 所有, 如有侵权,请联系我们删除。

“实验SparkSQL编程初级实践”的评论:

还没有评论