Spark编程底子：（实行四）Sark SQL**编程初级实践**

Sark SQL**编程初级实践**
一、实行环境

操作系统:Ubunt 16.04.
Spark 版本:2.4.0。
数据库:MySQL。
Python 版本:3.4.3.
from pyspark import SparkContext, SparkConf
 from pyspark.sql import SparkSession
 import os
 os.environ["JAVA_HOME"]="/home/spark021/servers/jdk"
 os.environ["YSPARK_PYTHON"]="/home/spark021/servers/anaconda3/envs/pyspark/bin/python3.6"
 os.environ["YSPARK_DRIVER_PYTHON"]="/home/spark021/servers/anaconda3/envs/pyspark/bin/python3.6"
 spark= SparkSession.builder.appName("employee021").config(conf=SparkConf()).getOrCreate()

二、 实行内容与完成情况

1. Spark SQL根本操作

为 employee.json 创建 DataFrame，并编写 Python 语句完成下列操作:
(1)查询所有数据;
(2)查询所有数据，并去除重复的数据;
(3)查询所有数据，打印时去除id 字段;
(4)筛选出 age>30 的记载;
(5)将数据按 age 分组;
(6)将数据按name升序分列;
(7)取出前3行数据;
(8)查询所有记载的 name 列，并为其取别名为 userame;
(9)查询年龄 age的平均值;
(10)查询年龄 age 的最小值。

# (1) 加载 JSON 文件并查询所有数据
 df = spark.read.json("file:///home/spark021/data/employee021.json")
 df.show()

# (2) 去除重复的数据
 unique_df = df.dropDuplicates()
 unique_df.show()

# (3) 去除 id 字段
 df_no_id = df.selectExpr(*[c for c in df.columns if c != 'id'])
 df_no_id.show()

# (4) 筛选出 age > 30 的记载
 older_than_30 = df.filter(df['age'] > 30)
 older_than_30.show()

# (5) 按 age 分组
 grouped_by_age = df.groupBy('age')
 grouped_by_age.count().show()

# (6) 按 name 升序分列
 sorted_by_name = df.orderBy('name')
 sorted_by_name.show()

# (7) 取出前3行数据
 first_three = df.limit(3)
 first_three.show()

# (8) 重定名 name 列为 username
 renamed_df = df.withColumnRenamed('name', 'username')
 renamed_df.show()

# （9）利用聚合函数avg()盘算年龄的平均值
 average_age = df.agg({"age": "avg"}).collect()0
 print("年龄平均值:", average_age)
 # （10）利用聚合函数min()盘算年龄的最小值
 min_age = df.agg({"age": "min"}).collect()0
 print("年龄最小值:", min_age)
 # 关闭 SparkSession
 spark.stop()

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

 from pyspark import SparkConf
 from pyspark.sql import SparkSession,Row
 from pyspark.sql.types import *
 import os
 os.environ["JAVA_HOME"]="/home/spark021/servers/jdk"
 os.environ["YSPARK_PYTHON"]="/home/spark021/servers/anaconda3/envs/pyspark/bin/python3.6"
 os.environ["YSPARK_DRIVER_PYTHON"]="/home/spark021/servers/anaconda3/envs/pyspark/bin/python3.6"
 spark = SparkSession.builder.appName("RDD转换成DataFrame").config(conf=SparkConf()).getOrCreate()
 *#**配置Spark应用程序
* employee021 = spark.sparkContext.\
 textFile("file:///home/spark021/data/employee021.txt").\
 map(lambda line: line.split(",")).\
 map(lambda x: Row(id=x[0],name=x[1],age=x[1]))
 employee021.foreach(print)
 *#创建DataFrame(具有模式)
* schemaPeople = spark.createDataFrame(employee021)
 *#注册为暂时表，才可以通过sql查询
* schemaPeople.createOrReplaceTempView("people")
 peopleDF = spark.sql("select * from people ")
 peopleDF.show()
 peopleRDD = peopleDF.rdd.map(lambda p: "id"+":"+p.id+","+"name"+":"+p.name+","+"age"+":"+p.age)
 peopleRDD.foreach(print)

3.编码实现利用DataFrame读写Mysql的数据

（1）创建数据库

创建表
create table employee021 (id int(4), name char(20), gender char(4), age int(4));
插入数据
insert into employee021 values(1,'Alice','F',22);
insert into employee021 values(2,'John','M',25);
查看表
select * from employee021;

（2）进入saprk，配置 Spark通过 JDBC 连按数据库 MySQL,编程实现利用 DataFrame 插入数据到 MySQL 中,末了打印出 age 的最大值和 age 的总和。
from pyspark.sql import Row
from pyspark.sql.types import *
from pyspark import SparkContext, SparkConf
from pyspark.sql import  SparkSession
import os
os.environ["JAVA_HOME"]="/home/spark021/servers/jdk"
os.environ["YSPARK_PYTHON"]="/home/spark021/servers/anaconda3/envs/pyspark/bin/python3.6"
os.environ["YSPARK_DRIVER_PYTHON"]="/home/spark021/servers/anaconda3/envs/pyspark/bin/python3.6"

spark=SparkSession.builder.config("spark.jars", "/home/spark021/servers/spark-local/jars/mysql-connector-java-5.1.40-bin.jar").getOrCreate()

#配置 Spark通过 JDBC 连按数据库 MySQL
employeeDF = spark.read \
    .format("jdbc")\
    .option("driver","com.mysql.jdbc.Driver")\
    .option("url","jdbc:mysql://localhost:3306/sparktest")\
    .option("dbtable","employee021")\
    .option("user","spark021")\
    .option("password","123456")\
    .load()
employeeDF.show()

#编程实现利用 DataFrame 插入数据到 MySQL 中
#设置模式信息
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"])\
            .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']='spark021'
prop['password']='123456'
prop['driver']="com.mysql.jdbc.Driver"
employeeDF.write.jdbc("jdbc:mysql://localhost:3306/sparktest",'employee021','append', prop)

employeeDF.show()

#末了打印出 age 的最大值和 age 的总和。
jdbcDF=spark.read.format("jdbc")\
    .option("driver","com.mysql.jdbc.Driver") .option("url","jdbc:mysql://localhost:3306/sparktest")\
    .option("dbtable","employee021") .option("user","spark021")\
    .option("password","123456")\
    .load()
jdbcDF.createOrReplaceTempView("employee021")
spark.sql("select MAX(age) from employee021").show()
spark.sql("select Sum(age) from employee021").show()

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。