1. spark的作用
  1. spark替代hadoop实现大数据的离线分析统计
  2. Spark通过内存计算能力极大地提高了大数据处理速度。
  3. Spark还支持SQL查询，流式计算，图计算，机器学习等。通过对Java、Python、Scala、R等语言的支持，极大地方便了用户的使用
2. spark的特点
  1. 快速处理能力。

随着实时大数据应用越来越多，Hadoop作为离线的高吞吐、低响应框架已不能满足这类需求。Hadoop MapReduce的Job将中间输出和结果存储在HDFS中，读写HDFS造成磁盘IO成为瓶颈。Spark允许将中间输出和结果存储在内存中，节省了大量的磁盘IO。同时Spark自身的DAG执行引擎也支持数据在内存中的计算。Spark官网声称性能比Hadoop快100倍。即便是内存不足需要磁盘IO，其速度也是Hadoop的10倍以上。

1. 1. 易于使用

Spark现在支持Java、Scala、Python和R等语言编写应用程序，大大降低了使用者的门槛。自带了80多个高等级操作符，允许在Scala，Python，R的shell中进行交互式查询。

1. 1. 支持查询

Spark支持SQL及Hive SQL对数据查询。

1. 1. 支持流式计算

与MapReduce只能处理离线数据相比，Spark还支持实时的流计算。Spark依赖Spark Streaming对数据进行实时的处理，其流式处理能力还要强于Storm。

1. 1. 可用性高

Spark自身实现了Standalone部署模式，此模式下的Master可以有多个，解决了单点故障问题。此模式完全可以使用其他集群管理器替换，比如YARN、Mesos、EC2等。

1. spark的生态

正在上传…重新上传取消

1. spark的基本架构

正在上传…重新上传取消

spark的运行原理

2.1.spark的运行原理图

正在上传…重新上传取消

2.2.spark与hadoop之间的对比

Spark 是在借鉴了 MapReduce 之上发展而来的，继承了其分布式并行计算的优点并改进了 MapReduce 明显的缺陷:

(1)Spark 把中间数据放到内存中，迭代运算效率高。MapReduce 中计算结果需要落地，保存到磁盘上，这样势必会影响整体速度，而 Spark 支持 DAG 图的分布式并行计算的编程框架，减少了迭代过程中数据的落地，提高了处理效率。（延迟加载）

(2)Spark 容错性高。Spark 引进了弹性分布式数据集 RDD (Resilient DistributedDataset) 的抽象，它是分布在一组节点中的只读对象集合，这些集合是弹性的，如果数据集一部分丢失，则可以根据“血统”（即允许基于数据衍生过程）对它们进行重建。另外在RDD 计算时可以通过 CheckPoint 来实现容错。

(3)Spark 更加通用。mapreduce 只提供了 Map 和 Reduce 两种操作，Spark 提供的数据集操作类型有很多，大致分为：Transformations 和 Actions 两大类。Transformations包括 Map、Filter、FlatMap、Sample、GroupByKey、ReduceByKey、Union、Join、Cogroup、MapValues、Sort 等多种操作类型，同时还提供 Count, Actions 包括 Collect、Reduce、Lookup 和 Save 等操作

spark的运行环境配置

3.1.spark的运行环境

3.2.spark的部署模式

（1）local(本地模式)：常用于本地开发测试，本地还分为local单线程和local-cluster多线程

（2）standalone(集群模式)：典型的Mater/slave模式，不过也能看出Master是有单点故障的；Spark支持ZooKeeper来实现 HA

（3）on yarn(集群模式)：运行在 yarn 资源管理器框架之上，由 yarn 负责资源管理，Spark 负责任务调度和计算

（4）on mesos(集群模式)：运行在 mesos 资源管理器框架之上，由 mesos 负责资源管理，Spark 负责任务调度和计算

3.3.在windows10下配置spark的运行环境

3.1.1.解压缩spark-3.0.3-bin-hadoop2.7(win10可用).tgz到指定的磁盘

本机使用d:\hadoop目录作为配置目录

正在上传…重新上传取消

3.3.2.spark的配置环境变量

1.建立spark环境

正在上传…重新上传取消

2.建立pythonpath环境变量

正在上传…重新上传取消

3.安装py4j日志组件

pip install py4j

将spark的bin目录添加到环境变量中

Path=D:\hadoop\spark-3.0.3-bin-hadoop2.7\bin

正在上传…重新上传取消

点击确定，使用管理员身份进入cmd,运行pyspark出现如下图所示界面，表示spark配置成功

正在上传…重新上传取消

3.3.3.window10下安装hadoop3

1.解压缩hadoop-3.3.1-aarch64.tar.gz文件，拷贝解压缩目录中的hadoop-3.3.1目录到d:\hadoop目录中

2.解压缩apache-hadoop-3.1.0-winutils-master (1).rar文件得到apache-hadoop-3.1.0-winutils-master (1)\apache-hadoop-3.1.0-winutils-maste目录结构

拷贝apache-hadoop-3.1.0-winutils-maste\bin目录下的所有文件

正在上传…重新上传取消

覆盖D:\hadoop\hadoop-3.3.1\bin目录下的所有文件

正在上传…重新上传取消

将hadoop在win10下的执行文件拷贝到此处。

3.将hadoop.dll复制到windows的system32目录下

正在上传…重新上传取消

复制到

正在上传…重新上传取消

4.配置hadoop的环境变量

HADOOP_HOME=D:\hadoop\hadoop-3.3.1

PATH=D:\hadoop\hadoop-3.3.1\bin

正在上传…重新上传取消

配置path路径

正在上传…重新上传取消

3.3.4.配置hadoop的配置文件

1.将安装完成的java目录复制到c盘根目录下，注意，不要有中文和空格等特殊字符

正在上传…重新上传取消

修改 hadoop的环境变量参数

打开 \etc\hadoop\hadoop-env.cmd文件,修改JAVA_HOME为C:\Java\jdk1.8.0_131 。

正在上传…重新上传取消

set JAVA_HOME=C:\Java\jdk1.8.0_131

修改hadoop的核心配置文件core-site.xml

<name>fs.default.name</name>

<value>hdfs://localhost:9000</value>

</property>

</configuration>

修改etc/hadoop/mapred-site.xml配置（是mapred-site.xml.template,去掉template）

<name>mapreduce.framework.name</name>

</property>

</configuration>

在D:\hadoop目录下创建数据存储目录data目录,目录结构如下：

正在上传…重新上传取消

修改etc/hadoop/hdfs-site.xml配置，指定数据存储的位置

<name>dfs.replication</name>

</property>

<name>dfs.namenode.name.dir</name>

<value>file:/hadoop/data/dfs/namenode</value>

</property>

<name>dfs.datanode.data.dir</name>

<value>file:/hadoop/data/dfs/datanode</value>

</property>

</configuration>

修改etc/hadoop/yarn-site.xml配置

正在上传…重新上传取消

<name>yarn.nodemanager.aux-services</name>

<value>mapreduce_shuffle</value>

</property>

<name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>

<value>org.apache.hadoop.mapred.ShuffleHandler</value>

</property>

</configuration>

检查Hadoop是否安装成功

运行“Hadoop version”,成功如下：

正在上传…重新上传取消

启动hadoop

用管理员身份打开cmd:

正在上传…重新上传取消

首先格式化hdfs存储空间,输入下面指令

hdfs namenode -format

正在上传…重新上传取消

启动hadoop服务

进入D:\hadoop\hadoop-3.3.1\sbin目录下输入下面指令，启动hadoop服务：

Start-all.cmd

正在上传…重新上传取消

使用浏览器查看hadoop是否配置成功：

正在上传…重新上传取消

10.再次在cmd中输入pyspark就不会有异常提示了

正在上传…重新上传取消

使用pycharm开发spark程序
1. 在pycharm中安装pyspark组件

Pip install pyspark

Pip install py4j

正在上传…重新上传取消

1. 在pycharm的py文件中使用代码设置环境变量

4.2.1在程序中设置环境变量

import os

import sys

os.environ['spark_home'] = 'c:\xxx\spark-2.2.1-bin-hadoop2.7'

sys.path.append('c:\xxx\spark-2.2.1-bin-hadoop2.7\python')

4.3.在pycharm中添加spark环境

正在上传…重新上传取消

如果在windows的系统环境变量中配置过spark的环境变量，则下面的配置可以省略,如果没有配置，则需要进行下面的配置步骤

正在上传…重新上传取消

配置spark环境：总共3个（SPARK_HOME、HADOOP_HOME、PYTHONPATH）

正在上传…重新上传取消

1. 编写spark程序的步骤
  1. 引入以下的组件库

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

1. 1. 初始化并创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

1. 1. 创建spark的上下文关系对象sparkContext

sc = spark.sparkContext

1. 1. 使用spark上下文关系对象创建RDD(弹性分布式数据)对象

input_rdd=sc.textFile("f:\\ajk.csv")#加载数据文件创建rdd对象

1. 1. 使用rdd对象处理分析数据

print(input_rdd.count())#词条数量

完整示例代码如下:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

#3使用上下文关系对象加载数据文件f:\\ajk.csv创建rdd对象

input_rdd=sc.textFile("f:\\ajk.csv")

#4统计文档中的条目数量

num=input_rdd.count()

print("磁条的数目为:",num)

正在上传…重新上传取消

spark的RDD和Value转换

5.1.RDD概念

（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变（只读）、可分区、里面的元素可并行计算的集合。

5.2.RDD特性：

(1)只读：不能修改，只能通过转换操作生成新的 RDD。

(2)分布式：可以分布在多台机器上进行并行处理。

(3)弹性：计算过程中内存不够时它会和磁盘进行数据交换。

(4)基于内存：可以全部或部分缓存在内存中，在多次计算间重用。

5.3.RDD分区运算

可以将 RDD 理解为一个分布式对象集合，本质上是一个只读的分区记录集合。每个 RDD 可以分成多个分区，每个分区就是一个数据集片段。

分区是RDD内部并行计算的一个计算单元，RDD的数据集在逻辑上被划分为多个分片，每一个分片称为分区，分区的格式决定了并行计算的粒度，而每个分区的数值计算都是在一个任务中进行的，因此任务的个数，也是由RDD的分区数决定。

正在上传…重新上传取消

5.4.RDD分区的原则

RDD分区的一个分区原则是使得分区个数尽量等于集群中的CPU核心（core）数量。分区过多并不会增加执行速度。

例如，我们集群有10个core，我们分5个区，每个core执行一个分区操作，剩下5个core浪费。如果，我们分20分区，一个core执行一个分区，剩下的10分区将会排队等待。

默认分区数目：

spark.default.parallelism这个参数值，来配置默认分区。

5.5.RDD通过Lineage（血统）实现数据的容错性。

正在上传…重新上传取消

5.5.1.RDD依赖

RDD通过操作算子进行转换，转换得到的新RDD包含了从其他RDDs衍生所必需的信息，RDD之间维护着这种血缘关系，也称之为依赖。依赖包括两种，一种是窄依赖，另一种是宽依赖。

（1）窄依赖

每个父RDD的一个Partition最多被子RDD的一个Partition所使用（1:1 或 n:1）。例如map、filter、union等操作都会产生窄依赖；子RDD分区通常对应常数个父RDD分区。

(2)宽依赖

一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey

、reduceByKey、sortByKey等操作都会产生宽依赖；（1:m 或 n:m）。

正在上传…重新上传取消

5.6.获取RDD对象的步骤

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

#1.创建spark对象

spark=SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

#2.创建spark的上下文关系对象

sc=spark.sparkContext;

#3.创建RDD对象

rdd1=sc.parallelize([2,3,4,5,6])

窄依赖:每个父RDD的一个Partition最多被子RDD的一个Partition所使用

宽依赖:一个父RDD的Partition会被多个子RDD的Partition所使用

Value数据类型的Transformation算子

6.1.算子分类

算子实际是进行数据运算的函数。

1、Value数据类型的Transformation算子，这种变换不触发提交作业，针对处理的数据项是Value型的数据。

2、Key-Value数据类型的Transformation算子，这种变换不触发提交作业，针对处理的数据项是Key-Value型的数据。

3、Action算子，这类算子会触发SparkContext提交作业。

6.1.1.map算子

将原来 RDD 的每个数据项通过 map 中的用户自定义函数 f 映射转变为一个新的元素。源码中 map 算子相当于初始化一个 RDD，新 RDD 叫做 MappedRDD(this, sc.clean(f))。

正在上传…重新上传取消

示例：RDD使用map转换为collect集合，并显示ajk文件中的词条数目和内容信息:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

#3.读取文件获取rdd对象

rdd1=sc.textFile("ajk.csv")

#4.使用map函数通过rdd对象转换为集合

list=rdd1.map(lambda x:x.split(",")).collect()

print(list)

#获取ajk.scsv文件中的词条个数

num=len(list)

print('词条个数:',num)

for s in list:

print(s)

for s1 in s:

print(s1)

print('======================================')

正在上传…重新上传取消

6.1.2.flatMap算子

将原来 RDD 中的每个元素通过函数 f 转换为新的元素，并将生成的 RDD 的每个集合中的元素合并为一个集合，内部创建 FlatMappedRDD(this，sc.clean(f)),flatMap为一个扁平化操作。

正在上传…重新上传取消

flatMap的示例

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

#3使用flatmap转换为集合函数

rdd=sc.parallelize(['hello spark','hello spark','hello hadoop'])

#4使用flatmap转换,使用空格拆分

list=rdd.flatMap(lambda x:x.split(" ")).collect()

print(type(list))

print(list)

num=len(list)

print("词条个数:",num)

正在上传…重新上传取消

6.1.3.filter算子

过滤算子，根据函数的规则返回(true的)一个过滤过后的rdd

过滤掉指定数据后返回的一个RDD对象

示例如下:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

#3使用flatmap转换为集合函数

rdd=sc.parallelize(['hello spark','hello spark','hello hadoop'])

#4使用flatmap转换,使用空格拆分

list=rdd.flatMap(lambda x:x.split(" ")).collect()

print(type(list))

print(list)

num=len(list)

print("词条个数:",num)

#使用filter算子，过滤掉不包含hello spark的数据,==表示包含的意思

list1=rdd.filter(lambda x:x=="hello spark").collect()

print(list1)

正在上传…重新上传取消

6.1.4.mapPartitions

mapPartitions 函数获取到每个分区的迭代器,在函数中通过这个分区整体的迭代器对整个分区的元素进行操作。内部实现是生成MapPartitionsRDD。

正在上传…重新上传取消

上面图示中过滤掉小于3的所有分区值。

MapPartitions的优点：

如果是普通的map，比如一个partition中有1万条数据。ok，那么你的function要执行和计算1万次。使用MapPartitions操作之后，一个task仅仅会执行一次function，function一次接收所有的partition数据。只要执行一次就可以了，性能比较高。

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

#3使用flatmap转换为集合函数

rdd=sc.parallelize(['hello spark','hello spark','hello hadoop'])

#4使用mapPartitions转换,使用空格拆分

list=rdd.mapPartitions(lambda x:x)

print(type(list))

print(list)

6.1.5.union算子

使用 union 函数时需要保证两个 RDD 元素的数据类型相同，返回的 RDD 数据类型和被合并的 RDD 元素数据类型相同，并不进行去重操作，保存所有元素。如果想去重可以使用 distinct()。同时 Spark 还提供更为简洁的使用 union 的 API，通过 ++ 符号相当于 union 函数操作。

示例:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

rdd1=sc.parallelize([1,4,5])

rdd2=sc.parallelize([2,4,7])

rdd3=rdd1.union(rdd2)

list=rdd3.map(lambda x:x).collect()

print(list)

正在上传…重新上传取消

6.1.5.cartesian(笛卡尔)算子

对两个RDD内的所有元素进行笛卡尔积操作。操作后，内部实现返回CartesianRDD。

示例如下:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

rdd1=sc.parallelize([1,2,3])

rdd2=sc.parallelize([2,3,4,5])

rdd3=rdd1.cartesian(rdd2)#调用笛卡尔算子

list=rdd3.map(lambda x:x).collect()

print(list)

正在上传…重新上传取消

6.1.6.groupBy算子

groupBy ：将元素通过函数生成相应的 Key，数据就转化为 Key-Value 格式，之后将 Key 相同的元素分为一组。

示例:

将偶数标注为a，奇数标注为b

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from pyspark.resultiterable import ResultIterable

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

rdd1=sc.parallelize([1,2,3,4,5,6,7,8,9])

rdd2=rdd1.groupBy(lambda x:x%2==0)

list=rdd2.map(lambda x1:x1).collect()

print(list)

for ls in list:

if ls[0]==False:#奇数值

print('奇数值为:')

for ls11 in ls[1]:

print(ls11)

print("======================")

else:

for ls11 in ls[1]:

print('偶数值为:')

for ls11 in ls[1]:

print(ls11)

print("======================")

正在上传…重新上传取消

6.1.7.distinct算子

distinct将RDD中的元素进行去重操作。每个方框代表一个RDD分区，通过distinct函数，将数据去重。

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from pyspark.resultiterable import ResultIterable

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

rdd1=sc.parallelize([1,2,2,3,4,5,5,5,6])

list=rdd1.distinct().collect()

print(list)

正在上传…重新上传取消

6.1.8.subtract（差值运算）

subtract相当于进行集合的差操作，但返回在RDD中出现，并且不在otherRDD中出现的元素，不去重。

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

rdd1=sc.parallelize([1,2,3])

rdd2=sc.parallelize([2,3,4,5])

#两个rdd中的数据进行差值运算

list=rdd1.subtract(rdd2).collect()

print(list)

正在上传…重新上传取消

6.1.9.sample样本算子

sample 将 RDD 这个集合内的元素进行采样，获取所有元素的子集。用户可以设定是否有放回的抽样、百分比、随机种子，进而决定采样方式。

withReplacement=true，表示有放回的抽样。

withReplacement=false，表示无放回的抽样。

示例:

(从rdd1中随机且有放回的抽出10%的数据，随机种子值为3（即可能以1 2 3的其中一个起始值）)

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

rdd1=sc.parallelize([1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,24,56,7,8,545,33])

rdd2=rdd1.sample(True,0.1,10);#从rdd1中随机且有放回的抽出10%的数据，随机种子值为10（即可能以1 2 3...的其中一个起始值）

list=rdd2.collect()

print(list)

正在上传…重新上传取消

6.1.10.takeSample算子

takeSample（）函数和上面的sample函数是一个原理，但是不使用相对比例采样，而是按设定的采样个数进行采样，同时返回结果不再是RDD，而是相当于对采样后的数据进行Collect（），返回结果的集合为单机的数组。

示例如下：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

# 1.创建spark对象

spark = SparkSession.builder.appName("mytestapp").master('local').getOrCreate()

# 2.创建spark的上下文关系对象

sc = spark.sparkContext;

rdd1=sc.parallelize([1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,24,56,7,8,545,33])

list=rdd1.takeSample(True,5)#取出集合5个随机的数据

print(list)

正在上传…重新上传取消

6.2.spark中的key-value算子和action算子

6.2.1.mapValues算子

针对（Key,Value）型数据中的 Value 进行Map操作，而不对Key进行处理。

示例如下:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list=[('1班',22),('2班',12),('3班',23),('4班',5)];

rdd=sc.parallelize(list)

rdd1=rdd.mapValues(lambda x:x+2)#在原值的基础上+2 ，例如1班,22人+2=24人

list1=rdd1.collect()

print(list1)

正在上传…重新上传取消

6.2.2.flatMapValues算子

同基本转换操作中的flatMap，只不过flatMapValues是针对[K,V]中的V值进行flatMap操作。

示例:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list=[('1班',22),('2班',12),('3班',23),('4班',5)];

rdd=sc.parallelize(list)

list1=rdd.flatMapValues(lambda x:[x,'1班']).collect()

print(list1)

正在上传…重新上传取消

6.2.3.combineByKey算子(统计算子)

1.常用的基于key的聚合函数，返回的类型可以和输入的类型不一样

2.许多基于key的聚合函数有用到了它，像groupByKey()

3.combineByKey 函数的定义：

combineByKey[C](createCombiner:(V) C,

mergeValue:(C, V) C,

mergeCombiners:(C, C) C,

partitioner:Partitioner,

mapSideCombine:Boolean=true,

serializer:Serializer=null):RDD[(K,C)]

4.示例

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

#向集合中设置元素a,并定义a集合

def to_list(a):

return [a]

#向a集合中加入元素b

def append(a,b):

a.append(b)

return a

#将a集合扩充到b集合

def extend(a,b):

a.extend(b)

return a

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list=[('1月',10),('1月',20),('1月',30),('2月',15),('2月',15),('3月',25),('3月',10),('4月',45),('4月',25)]

rdd=sc.parallelize(list)

list1=rdd.combineByKey(to_list,append,extend).collect()

print(list1)

正在上传…重新上传取消

6.2.4.foldByKey算子

foldByKey操作作用于RDD[K,V]根据K将V做折叠、合并处理。

示例:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list=[('1月',10),('1月',20),('1月',30),('2月',15),('2月',15),('3月',25),('3月',10),('4月',45),('4月',25)]

rdd=sc.parallelize(list)

list1=rdd.foldByKey(0,add).collect()#对每月数据进行统计计算操作

print(list1)

正在上传…重新上传取消

6.2.5.reduceByKey算子

reduceByKey就是对元素为KV对的RDD中Key相同的元素的Value进行binary_function的reduce操作，因此，Key相同的多个元素的值被reduce为一个值，然后与原RDD中的Key组成一个新的KV对。

示例:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list=[('1月',10),('1月',20),('1月',30),('2月',15),('2月',15),('3月',25),('3月',10),('4月',45),('4月',25)]

rdd=sc.parallelize(list)

list1=rdd.reduceByKey(add).collect()

print(list1)

正在上传…重新上传取消

6.2.6.groupByKey算子

groupByKey是对每个key进行合并操作，但只生成一个sequence

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list=[('苹果',10),('苹果',10),('苹果',30),('香蕉',15),('香蕉',15),('樱桃',25),('樱桃',10),('樱桃',45),('石榴',25)]

rdd=sc.parallelize(list)#加载list集合并获取rdd对象

list1=rdd.groupByKey().mapValues(len).collect()#对相同键进行统计,类似于统计类别的个数

print(list1)

正在上传…重新上传取消

对键进行统计，不统计值

正在上传…重新上传取消

6.2.7.sortByKey算子

sortByKey 对key进行排序，默认为升序

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list=[('5月',10),('1月',20),('6月',30),('2月',15),('2月',15),('3月',25),('3月',10),('4月',45),('4月',25)]

rdd=sc.parallelize(list)

list1=rdd.sortByKey().collect()

print(list1)

正在上传…重新上传取消

6.2.8.cogroup算子

cogroup

对两个RDD(如:(K,V)和(K,W))相同Key的元素先分别做聚合，最后返回(K,Iterator<V>,Iterator<W>)形式的RDD,numPartitions设置分区数，提高作业并行度

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list1=[('1月',10),('2月',20),('3月',30)]

list2 = [('1月', '梅花'), ('2月', "梨花"), ('3月', "桃花")]

rdd1=sc.parallelize(list1)

rdd2 = sc.parallelize(list2)

'''

rdd3=rdd1.cogroup(rdd2)

list3=rdd3.map(lambda x:x).collect()

print(list3)

for ls in list3:

key=ls[0]

print(key)

vals=ls[1]

for val in vals:

print(val)

'''

list4= [(rdd1, tuple(map(list,rdd2))) for rdd1, rdd2 in sorted(list(rdd1.cogroup(rdd2).collect()))]

print(list4)

正在上传…重新上传取消

6.2.9.join算子

join

内连接的join，以某一个表为基础，KEY相同的打印出来，不相同的不打印

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list1=[('1月',10),('2月',20),('3月',30),('4月',90)]

list2 = [('1月', '梅花'), ('2月', "梨花"), ('3月', "桃花")]

rdd1=sc.parallelize(list1)

rdd2 = sc.parallelize(list2)

rdd3=rdd1.join(rdd2)

list3=rdd3.map(lambda x:x).collect()

print(list3)

正在上传…重新上传取消

6.2.10.leftOuterJoin

leftOuterJoin返回数据集左边的全部数据和数据集左边与右边有交集的数据

示例:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list1=[('1月',10),('2月',20),('3月',30),('4月',90)]

list2 = [('1月', '梅花'), ('2月', "梨花"), ('3月', "桃花")]

rdd1=sc.parallelize(list1)

rdd2 = sc.parallelize(list2)

rdd3=rdd1.leftOuterJoin(rdd2)

list3=rdd3.map(lambda x:x).collect()

print(list3)

正在上传…重新上传取消

6.2.11.fullOuterJoin算子

返回左右数据集的全部数据，左右有一边不存在的数据以None填充

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

'''

使用mapVlues的示例

'''

#定义集合

list1=[('1月',10),('2月',20),('3月',30),('4月',90)]

list2 = [('1月', '梅花'), ('2月', "梨花"), ('3月', "桃花"),('5月','玫瑰')]

rdd1=sc.parallelize(list1)

rdd2 = sc.parallelize(list2)

rdd3=rdd1.fullOuterJoin(rdd2)

list3=rdd3.map(lambda x:x).collect()

print(list3)

正在上传…重新上传取消

Actions算子

7.1.foreach算子

foreach用于遍历RDD,将函数f应用于每一个元素。

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=['苹果','西瓜','香蕉','橘子','菠萝','葡萄','香瓜']

rdd1=sc.parallelize(list)

rdd1.foreach(lambda x:print(x))

正在上传…重新上传取消

7.2.collect算子

collect():以数据的形式返回数据集中的所有元素给Driver程序，为防止Driver程序内存溢出，一般要控制返回的数据集大小

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=['苹果','西瓜','香蕉','橘子','菠萝','葡萄','香瓜']

rdd1=sc.parallelize(list)

list1=rdd1.collect()

print(list1)

正在上传…重新上传取消

7.3.take算子

take(n):以数组的形式返回数据集上的前n个元素

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=['苹果','西瓜','香蕉','橘子','菠萝','葡萄','香瓜']

rdd1=sc.parallelize(list)

#显示前3个水果信息

print(rdd1.take(3))

正在上传…重新上传取消

7.4.top算子

top(n):按默认或者指定的排序规则返回前n个元素，默认按降序输出

示例:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=['苹果','西瓜','香蕉','橘子','菠萝','葡萄','香瓜']

rdd1=sc.parallelize(list)

#显示前3个水果信息

print(rdd1.top(3))

正在上传…重新上传取消

7.5.first算子

first():返回数据集的第一个元素

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=['苹果','西瓜','香蕉','橘子','菠萝','葡萄','香瓜']

rdd1=sc.parallelize(list)

#显示前3个水果信息

print(rdd1.first())

正在上传…重新上传取消

7.6.reduce算子

reduce(func):通过函数func先聚集各分区的数据集，再聚集分区之间的数据，func接收两个参数，返回一个新值，新值再做为参数继续传递给函数func，直到最后一个元素

示例：

将数组的值进行统计:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=[12,23,45,6,78,9,11]

rdd1=sc.parallelize(list)

print(rdd1.reduce(add))

正在上传…重新上传取消

7.7.collectAsMap算子

collectAsMap():作用于K-V类型的RDD上，作用与collect不同的是collectAsMap函数不包含重复的key，对于重复的key。后面的元素覆盖前面的元素

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=[('code1','葡萄'),('code2','菠萝'),('code3','西瓜'),('code4','榴莲'),('code5','山竹')]

rdd1=sc.parallelize(list)

lsit2=rdd1.collectAsMap()

print(lsit2)

for item in lsit2.items():

print(item)

正在上传…重新上传取消

作业：

完成课堂案例，并提交到svn上

7.8.countByKey算子

countByKey():作用于K-V类型的RDD上，统计每个key的个数，返回(K,K的个数)，返回主键的个数值。

示例:

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=[('code1','葡萄'),('code2','菠萝'),('code3','西瓜'),('code4','榴莲'),('code5','山竹')]

rdd1=sc.parallelize(list)

countnum=rdd1.countByKey()

print("key的数量：", len(countnum))

正在上传…重新上传取消

7.9.lookup算子（查找算子）

lookup(k)：作用于K-V类型的RDD上，返回指定K的所有V值，获取主键对应的所有value值。

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=[('技术部','王勇'),('技术部','张青'),('技术部','马楠楠'),('行政部','曹莲花'),

('行政部','周东南'),('行政部','吕不韦'),('行政部','韩寒'),

('财务部','马俊杰'),('财务部','马冬梅'),('财务部','孙可可'),

('市场部','江湖接'),('市场部','兰可可'),('市场部','马花花'),('市场部','周飞跃')]

rdd1=sc.parallelize(list)

list1=rdd1.lookup('技术部')#查找行政部中的所有员工

print(list1)

正在上传…重新上传取消

7.10.foldz算子

fold()与reduce()类似，接收与reduce接收的函数签名相同的函数，区别再与：加上一个初始值作为第一次调用的结果。

示例

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=[12,23,45,6,78,9,11]

rdd1=sc.parallelize(list)

print(rdd1.fold(0,add))#将list集合中的数据进行累加运算

正在上传…重新上传取消

7.11.aggregate 算子

通常为我们的spark程序计算是分布式的，所以我们通常需要聚合的数据都分部在不同的分区，不同的机器上。

该函数它会首先对每个分区内的数据基于初始值进行一个首次聚合，然后将每个分区聚合的结果，通过使用给定的聚合函数，再次基于初始值进行分区之间的聚合，并且最终函数的返回结果的类型，可以是与该RDD的类型相同

示例：

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

list=[12,23,45,6,78,9,11]

rdd1=sc.parallelize(list)

seq0=(lambda x,y:((x[0]+y),x[1]+1))

comOp=(lambda x,y:(x[0]+y[0],x[1]+y[1]))

print(rdd1.aggregate((0,0),seq0,comOp))

正在上传…重新上传取消

使用spark分析安居客csv文件中的信息，并打印输入

import os

import re

from pyspark import SparkContext

from pyspark.sql import SparkSession#spark2.0使用此组件构建spark上下文关系对象

from operator import add

if __name__ == '__main__':

#创建spark对象

spark = SparkSession \

.builder \

.appName("PythonWordCount") \

.master("local") \

.getOrCreate()

#2创建spark的上下文关系对象

sc=spark.sparkContext

#读取ajk.csv文件，并转换为RDD

rdd1=sc.textFile("ajk.csv")

#将ajk.csv文件中的数据转为list集合输出

list=rdd1.map(lambda x:x.split(",")).collect()

print(list)

print('============================================================')

for ajk in list:

print(ajk)

for ak in ajk:

print(ak)

print('==========')

print('============================================================')

print("获取ajk.csv文件的前两条记录:")

#获取前2条记录值

print(rdd1.take(2))

课堂练习:

完成课堂案例

实现将爬取的csv文件使用spark的算子进行转换和分析

spark广播变量

8.1.广播变量的主要作用

如果我们要在分布式计算里面分发大对象，例如：字典，集合，黑白名单等，这个都会由Driver端进行分发，一般来讲，如果这个变量不是广播变量，那么每个task就会分发一份，这在task数目十分多的情况下Driver的带宽会成为系统的瓶颈，而且会大量消耗task服务器上的资源，如果将这个变量声明为广播变量，那么只是每个executor拥有一份，这个executor启动的task会共享这个变量，节省了通信的成本和服务器的资源。

8.2.广播变量的定义方式

import pyspark

from pyspark import SparkContext

from pyspark.sql import SparkSession

if __name__ == '__main__':

spark=SparkSession.builder.appName("myapp").master('local').getOrCreate()

sc=spark.sparkContext

#定义数组

list=[12,2,3,44,5]

#定义广播变量

bordcastval=sc.broadcast(list)

print(bordcastval.value)

没有使用广播的图示：

正在上传…重新上传取消

使用广播变量的图示：

正在上传…重新上传取消

8.3.累加器

spark应用程序中，我们经常会有这样的需求，如异常监控，调试，记录符合某特性的数据的数目，这种需求都需要用到计数器，如果一个变量不被声明为一个累加器，那么它将在被改变时不会再driver端进行全局汇总，即在分布式运行时每个task运行的只是原始变量的一个副本，并不能改变原始变量的值，但是当这个变量被声明为累加器后，该变量就会有分布式计数的功能。

import pyspark

from pyspark import SparkContext

from pyspark.sql import SparkSession

if __name__ == '__main__':

spark=SparkSession.builder.appName("myapp").master('local').getOrCreate()

sc=spark.sparkContext

#定义数组

list=[1,2,3,4,5]

#定义累加器变量

accum2=sc.accumulator(0,0)

print(accum2)

#定义rdd对象

rdd=sc.parallelize(list)

rdd.map(lambda x:accum2.add(x))

print(accum2)

8.3.spark的输入与输出

8.3.1.spark支持的输入输出方式

正在上传…重新上传取消

8.3.2.spark读取文本文件和写出文本文件数据

1.读取文件的操作

rdd=sparkContext.textFile(“本地文件文件路径”)

读取hadoop的hdfs上的数据

读取hdfs上的一个文件

hdfsfile1 = ssparkContext.textFile("hdfs://h201:9000/xxx.txt")

import pyspark

from pyspark import SparkContext

from pyspark.sql import SparkSession

if __name__ == '__main__':

spark=SparkSession.builder.appName("myapp").master('local').getOrCreate()

sc=spark.sparkContext

rdd=sc.textFile("f:\\ajk.csv")

print(rdd.take(4))#读取文件的前四记录

正在上传…重新上传取消

读取json格式的文件操作

Test.json文件格式

{"sid":"1","sname":"张飞","sex":"男","address":"西安","birthday":"1998-12-11"}

{"sid":"2","sname":"马超","sex":"男","address":"汉中","birthday":"1991-11-11"}

{"sid":"3","sname":"赵云","sex":"男","address":"常山","birthday":"1999-01-11"}

{"sid":"4","sname":"关羽","sex":"男","address":"运城","birthday":"1997-11-12"}

{"sid":"5","sname":"黄忠","sex":"男","address":"长沙","birthday":"1995-10-21"}

import json

import pyspark

from pyspark import SparkContext

from pyspark.sql import SparkSession

from pymysql.constants.FIELD_TYPE import JSON

if __name__ == '__main__':

spark=SparkSession.builder.appName("myapp").master('local').getOrCreate()

sc=spark.sparkContext

rdd=sc.textFile("test.json")

print(rdd.take(4))#读取文件的前四记录

list1=rdd.map(lambda s:s.replace("\\",'')).collect()

print(list1)

for st in list1:

st1=json.loads(st)#将字符串转为json格式的数据

print(st1)

print(st1['sid'])

print(st1['sname'])

print(st1['sex'])

print(st1['address'])

print(st1['birthday'])

print("================")

正在上传…重新上传取消

sequencefile读取hadoop存储的二进制顺序文件

import pyspark

from pyspark import SparkContext

from pyspark.sql import SparkSession

from pymysql.constants.FIELD_TYPE import JSON

if __name__ == '__main__':

spark=SparkSession.builder.appName("myapp").master('local').getOrCreate()

sc=spark.sparkContext

list=[(12,22),(13,24),(21,20),(15,11),(17,18)]

#rdd=sc.parallelize(list)

#存储为hadoop的二进制的队列文件

#rdd.saveAsSequenceFile("f:\\dd")

#读取存储的hadoop队列文件

seq=sc.sequenceFile('f:\\dd\\part-00000')

list1=seq.collect()

print(list1)

正在上传…重新上传取消

3.对象文件

对象文件是将对象序列化后保存的文件，采用 Java 的序列化机制。可以通过

objectFile 函数接收一个路径，读取对象文件，返回对应的 RDD，也可以通过调用 saveAsObjectFile() 实现对对象文件的输出。

示例如下：

from pyspark import SparkContext

from pyspark.sql import SparkSession

if __name__ == '__main__':

spark=SparkSession.builder.appName("myapp").master("local").getOrCreate()

sc=spark.sparkContext#获取spark上下文关系对象

#准备集合

list=[1,2,3,4,5,6,7]

#写文件操作

#rdd=sc.parallelize(list)#将集合转为rdd对象

#rdd.saveAsPickleFile('myobj')

#读取pick文件

rdd=sc.pickleFile("myobj/part-00000")

lits1=rdd.collect()

print(lits1)

使用spark访问mysql数据库

将mysql的数据库驱动包mysql-connector-java-8.0.17.jar拷贝到D:\hadoop\spark-3.0.3-bin-hadoop2.7\jars目录下

正在上传…重新上传取消

在python使用spark中访问mysql8.x版本的数据库

from pyspark import SparkContext

from pyspark.sql import SparkSession,SQLContext

if __name__ == '__main__':

spark=SparkSession.builder.appName("myapp").master("local").getOrCreate()

sc=spark.sparkContext#获取spark上下文关系对象

sqlsc=SQLContext(sc)#创建spark之sql的上下文关系对象

#使用sqlContext对象获取数据集df（DataFrame）对象

df = sqlsc.read.format("jdbc").options(url="jdbc:mysql://localhost:3307/zfdb?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&user=root&password=java",dbtable="zf").load()

df.show()

sc.stop()

正在上传…重新上传取消

spark的sql开发
1. spark sql的原理
  1. 历史

2011的时候，Hive可以说是SQL On Hadoop的唯一选择，负责将SQL解析成MR任务运行在大数据上，实现交互式查询、报表等功能。就在那个时候，Spark社区的小伙伴就意识到可以使用Spark作为执行引擎替换Hive中的MR，这样可以使Hive的执行效率得到极大提升。这个思想的产物就是Shark，所以从实现功能上来看，Shark更像一个Hive On Spark实现版本。

1. 1. sparkSQL体系结构

正在上传…重新上传取消

1. Catalyst工作流程

Catalyst为SparkSQL的优化器，Catalyst支持基于规则和基于成本的优化。

Catalyst的核心是使用一个通用库生成树并使用规则操作这些树。在该框架的基础上，构建了用于关系查询处理库（例如表达式，逻辑查询计划）和处理执行查询不同阶段的几组规则：分析、逻辑优化、物理计划和代码生成。

(1)Parser模块被解析为语法树

正在上传…重新上传取消

(2)Analyzer（分析）

正在上传…重新上传取消

Optimizer

正在上传…重新上传取消

9.3.RDD、DataFrame 与DataSet

正在上传…重新上传取消

DataSet包含RddS和DataFrames

Rdd和DataFrames是平行关系

9.3.1.RDD

(a)RDD是一个懒执行的不可变的可以支持Lambda表达式的并行数据集合。

(b)RDD的最大好处就是简单，API的人性化程度很高。

(c)RDD的劣势是性能限制，它是一个JVM驻内存对象，这也就决定了存在GC的限制和数据增加时Java序列化成本的升高。

9.3.2.Dataframe

与RDD类似，DataFrame也是一个分布式数据容器。然而DataFrame更像传统数据库的二维表格，除了数据以外，还记录数据的结构信息，即schema。同时，与Hive类似，DataFrame也支持嵌套数据类型（struct、array和map）。从API易用性的角度上看，DataFrame API提供的是一套高层的关系操作，比函数式的RDD API要更加友好，门槛更低。

正在上传…重新上传取消

1. spark sql基本操作
  1. 使用spark对象获取json格式的数据,并转换为DataFrame对象

import pyspark

from pyspark.sql import SparkSession,SQLContext

from pyspark import SparkContext

if __name__ == '__main__':

#创建spark对象

spark=SparkSession.builder.appName("myapp").master("local").getOrCreate()

#创建sparkContext对象

sc=spark.sparkContext

#读取stu.json文件,并获取datafram对象

df =spark.read.json("stu.json")

df.show();

正在上传…重新上传取消

1. 1. 查看表的结构

#查看表的结构

df.printSchema()

正在上传…重新上传取消

1. 1. 获取指定字段的列值

#获取指定字段的列值

df.select("id","sname").show()

正在上传…重新上传取消

9.3.4.过滤DataFrme中指定列的数据

显示年龄大于30岁的学生信息

#显示年龄大于30岁的学生信息

df.filter("age>=30").show()

正在上传…重新上传取消

1. 1. 统计操作

按照性别统计学生的个数

#按照性别统计学生的个数

df.groupBy("sex").count().show()

正在上传…重新上传取消

9.3.5.使用spark获取csv文件的数据转为DataFrame对象

import pyspark

from pyspark.sql import SparkSession,SQLContext

from pyspark import SparkContext

if __name__ == '__main__':

#创建spark对象

spark=SparkSession.builder.appName("myapp").master("local").getOrCreate()

#创建sparkContext对象

sc=spark.sparkContext

#使用spark读取csv文件获取DataFrame对象

ajkdf=spark.read.csv("ajkzj.csv",header=True,encoding="gbk")

#查看ajkdf中的数据

ajkdf.show()

#按照小区名称统计楼盘的数量

print("每个楼盘数量为:")

ajkdf.groupBy("xqname").count().show()

正在上传…重新上传取消

1. 1. 运行标准sql

import pyspark

from pyspark.sql import SparkSession,SQLContext

from pyspark import SparkContext

import matplotlib.pyplot as plt #导入matplotlib绘图库

plt.rcParams['font.sans-serif']=['SimHei'] #解决图形中中文字体显示问题

plt.rcParams['axes.unicode_minus']=False #解决负号显示为方块的问题

from pyecharts import options as opts

from pyecharts.charts import Pie

from pyecharts.charts import Bar

if __name__ == '__main__':

#创建spark对象

spark=SparkSession.builder.appName("myapp").master("local").getOrCreate()

#创建sparkContext对象

sc=spark.sparkContext

#使用spark读取csv文件获取DataFrame对象

ajkdf=spark.read.csv("ajkzj.csv",header=True,encoding="gbk")

#使用sql操作dataframe中的数据

#1，使用dataframe对象创建临时视图anjuketable

ajkdf.createOrReplaceTempView("anjuketable")

sqldf=spark.sql("select * from anjuketable")

sqldf.show()

sqldf=spark.sql("select xqname,avg(jiage) pingjun from anjuketable group by xqname")

print("小区的平均租金为:")

sqldf.show()

lsxqname=sqldf.select("xqname").rdd.collect()#获取小区名称的listy集合

lspingjunjiage=sqldf.select("pingjun").rdd.collect()#获取平均价格的集合

for xqname,avjprice in zip(lsxqname,lspingjunjiage):

print(xqname)

print(avjprice)

print("==========")

1. 1. 将dataframe转换为rdd并获取其中的元素值

from pyspark.sql import SparkSession

from pyspark import SparkContext

if __name__ == '__main__':

#获取spark对象

spark=SparkSession.builder.appName("myapp").master("local").getOrCreate()

sc=spark.sparkContext #获取spark的上下文关系对象

#读取csv文件，并获取DataFrame对象,显示第一行的标题

ajkdf=spark.read.csv("ajkzj.csv",encoding="gbk",header=True)

#ajkdf.show()#查看csv文件的前20数据

#创建临时表，使用sql语句操作csv文档,使用dataframe对象创建内存临时表

ajkdf.createOrReplaceTempView("anjuke")

#使用sprk对象操作sql查询anjuke临时内存表的数据

anjkdf=spark.sql("select * from anjuke")

anjkdf.show()

#使用sql的分组语句按照小区的名称统计小区的租金平均价格

sql="select xqname,avg(jiage) avjprice from anjuke group by xqname"

anjkgroupdf=spark.sql(sql)#使用spark对象执行分组的sql语句

anjkgroupdf.show()

#根据分组的ddataFrame对象获取分组列的dataframe对象

xqmcdf=anjkgroupdf.select("xqname")#获取小区名称的dataframe对象

print("小区名称")

xqmcdf.show()

ajgpricedf=anjkgroupdf.select("avjprice")#获取小区平均价格的dataframe对象

print("小区的平均价格")

ajgpricedf.show()

#将小区名称和平均价格的dataframe对象anjkgroupdf转换为rdd对象，并将rdd对象转为值集合

listgroup=anjkgroupdf.rdd.collect()

for xqpricerow in listgroup:

#print(xqprice)

print(xqpricerow['xqname'],'----',xqpricerow['avjprice'])

正在上传…重新上传取消

1. 创建dataset
  1. 编写json文件Person.json

{"name": "张飞","age": "22"}

{"name": "马超","age": "23"}

{"name": "关羽","age": "19"}

{"name": "赵云","age": "28"}

9.4.2.读取json文件并转为DataSet集合数据

from pyspark.sql import SparkSession

from pyspark import SparkContext

class Person(object):

name=None

age=0

def __init__(self,name,age):

self.name=name

self.age=age

if __name__ == '__main__':

spark=SparkSession.builder.appName("myperson").master("local").getOrCreate()

#读取person.json文件，并创建dataSet对象

psdataset=spark.read.json("person.json").alias("Person")

psdataset.show()

正在上传…重新上传取消

1. rdd转换为dataframes

from pyspark.sql import SparkSession

from pyspark import SparkContext

class Stu(object):

sname=None

age=10

def __init__(self,name,age):

self.sname=name

self.age=age

if __name__ == '__main__':

#获取spark对象

spark=SparkSession.builder.appName("myapp").master("local").getOrCreate()

sc=spark.sparkContext #获取spark的上下文关系对象

#创建集合

list=[('张飞',22),('林冲',32),('李阔',21),('晁盖',34),('宋江',21)]

#创建Rdd对象

rdd=sc.parallelize(list)

#使用 rdd转为 DataFrame对象

stdf=rdd.map(lambda x:x).toDF()

stdf.show()

#使用dataframe对象创建临时内存表

stdf.createOrReplaceTempView("st_table")

stddf=spark.sql("select * from st_table")

stddf.show()

正在上传…重新上传取消

1. StructType
  1. 结构体类型

9.6.2.StructField

作用：定义字段结构, 一个结构体内部的一个StructField就像一个SQL中的一个字段一样

1. 1. StructType

StructType对象，可以有多个StructField,同时也可以用名字（name）来提取,就想当于Map可以用key来提取value，但是他StructType提取的是整条字段的信息

scala> import org.apache.spark.sql.types._

val innerStruct =

StructType(

StructField("f1", IntegerType, true) ::

StructField("f2", LongType, false) ::

StructField("f3", BooleanType, false) :: Nil)

#使用以下语句返回的就是structType类型

stddf.printSchema()

正在上传…重新上传取消

使用pyecharts绘制基本图表

10.1.安装pyecharts

pip install pyecharts

10.2.使用pyecharts绘制正弦散点图

from pyecharts.charts import Funnel #漏斗图

from pyecharts.charts import Liquid #水球图

from pyecharts.charts import Gauge #仪表盘图

from pyecharts.charts import Geo #地图

from pyecharts.charts import Scatter # 导入散点图

from pyecharts.charts import Line # 导入折线图

from pyecharts.charts import Pie # 导入饼图

from pyecharts.charts import Geo # 导入地图

#"下面绘制的是：正弦曲线的散点图"

from pyecharts.charts import Scatter

import numpy as np

#设置x轴y轴数据

x = np.linspace(0,2 * np.pi,100)

y = np.sin(x)

(

#调用类

Scatter()

#添加x轴

.add_xaxis(xaxis_data=x)

#添加y轴

.add_yaxis(series_name="散点图",y_axis=y)

).render("zhengxuan.html")#直接在jupyter notebook运行显示

正在上传…重新上传取消

10.3.使用pyecharts绘制柱状图

from pyecharts.charts import Bar #导入柱状图的组件

from pyecharts import options as opts

#//设置行名

columns = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]

#//设置数据

data1 = [2.0, 4.9, 7.0, 23.2, 25.6, 76.7, 135.6, 162.2, 32.6, 20.0, 6.4, 3.3]

data2 = [2.6, 5.9, 9.0, 26.4, 28.7, 70.7, 175.6, 182.2, 48.7, 18.8, 6.0, 2.3]

#//设置柱状图的主标题与副标题

#//添加柱状图的数据及配置项

bar=(

Bar()

.add_xaxis(xaxis_data=columns)

.add_yaxis("降雨量", y_axis=data1)

.add_yaxis("增发量", y_axis=data2)

.set_global_opts(title_opts=opts.TitleOpts(title="Bar-基本示例", subtitle="我是副标题"))

)

bar.render("zhutu.html")

正在上传…重新上传取消

10.4.使用pyecharts绘制饼图

#普通方式画饼图

from pyecharts.charts import Pie

import pyecharts.options as opts

num = [110, 136, 108, 48, 111, 112, 103]

lab = ['哈士奇', '萨摩耶', '泰迪', '金毛', '牧羊犬', '吉娃娃', '柯基']

x = [(i, j)for i, j in zip(lab, num)]

print(x)

pie = Pie(init_opts=opts.InitOpts(width="700px",height="300px"))

pie.add(series_name='',data_pair=[(i, j)for i, j in zip(lab, num)])

pie.render('bingtu.html')

正在上传…重新上传取消

课堂练习：

实现课堂案例中的示例，修改其中的数据查看修改后的效果。

10.5.使用pyecharts绘制折线图

from pyecharts.charts import Line #引入折线图组件

from pyecharts import options as opts

#绘制折线图

if __name__ == '__main__':

columns = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]

# //设置数据

data1 = [2.0, 4.9, 7.0, 23.2, 25.6, 76.7, 135.6, 162.2, 32.6, 20.0, 6.4, 3.3]

data2 = [2.6, 5.9, 9.0, 26.4, 28.7, 70.7, 175.6, 182.2, 48.7, 18.8, 6.0, 2.3]

line = (

# 调用类

Line()

# 添加x轴

.add_xaxis(xaxis_data=columns)

# 添加y轴

.add_yaxis(series_name="折线图1", y_axis=data1)

.add_yaxis(series_name="折线图2", y_axis=data2)

)

line.render('zhexiantu.html')

正在上传…重新上传取消

10.6.使用pyecharts绘制雷达图

from pyecharts.charts import Radar #雷达图组件

if __name__ == '__main__':

radar = Radar()

# //由于雷达图传入的数据得为多维数据，所以这里需要做一下处理

radar_data1 = [[2.0, 4.9, 7.0, 23.2, 25.6, 76.7, 135.6, 162.2, 32.6, 20.0, 6.4, 3.3]]

radar_data2 = [[2.6, 5.9, 9.0, 26.4, 28.7, 70.7, 175.6, 182.2, 48.7, 18.8, 6.0, 2.3]]

# //设置column的最大值，为了雷达图更为直观，这里的月份最大值设置有所不同

schema = [

("Jan", 5), ("Feb", 10), ("Mar", 10),

("Apr", 50), ("May", 50), ("Jun", 200),

("Jul", 200), ("Aug", 200), ("Sep", 50),

("Oct", 50), ("Nov", 10), ("Dec", 5)

]

# //传入坐标

radar.add_schema(schema)

radar.add("降水量", radar_data1)

# //一般默认为同一种颜色，这里为了便于区分，需要设置item的颜色

radar.add("蒸发量", radar_data2, color="#1C86EE")

radar.render('leida.html')

正在上传…重新上传取消

10.7.使用pyecharts绘制地图热力图

zhoujielun1.csv

city,num

广东,2426

上海,4567

西安,3290

南京,786

北京,3210

天津,780

郑州,900

重庆,5670

成都,7600

兰州,800

拉萨,100

深圳,8900

香港,8700

长沙,900

厦门,870

杭州,6000

昆明,1000

from pyecharts.charts import Pie, Bar, Map, WordCloud

from pyecharts import options as opts

from pyspark.sql import SparkSession

from pyspark import SparkContext

if __name__ == '__main__':

spark=SparkSession.builder.appName("appname").master("local").getOrCreate()

df=spark.read.csv("zhoujielun1.csv",encoding="gbk",header=True)

df.show()

#将df转为集合

list=df.rdd.collect()

print(list)

store_location_num=[]

for cmrow in list:

store_location_num.append((cmrow['city'],cmrow['num']))

print(store_location_num)

map_ = (

Map()

# 添加标题,传入列表数据[('广东', 2426), ('上海', 560), ('浙江', 435)...]

# 'china为中国地图

.add("中国地图", [(data) for data in store_location_num], "china")

.set_global_opts(

title_opts=opts.TitleOpts(title="演唱会地点分布热力图（连续型）"),

visualmap_opts=opts.VisualMapOpts(max_=1000)) #'这里设置最大值,如果超过则不显示'

)

map_.render('演唱会.html')#

正在上传…重新上传取消

课堂练习：

完成课堂案例中的图形制作，可以改变数据

10.8.使用pyecharts绘制词云图

使用spark分析小区提供的房源数量，按照房源数量，绘制小区的词云图

from pyspark.sql import SparkSession

from pyspark import SparkContext

from pyecharts.charts import Pie, Bar, Map, WordCloud#词云图

from pyecharts import options as opts

from pyecharts.faker import Faker

from pyecharts.globals import ThemeType

from pyecharts.globals import SymbolType

if __name__ == '__main__':

#获取spark对象

spark=SparkSession.builder.appName("zfciyunapp").master("local").getOrCreate()

sc=spark.sparkContext

#使用spark对象读取ajkzj1.csv文件并创建DataFrame对象

zfdf=spark.read.csv("ajkzj1.csv",encoding="utf-8",header=True)

#zfdf.show()

#创建临时表用于按照小区的名称获取小区的房源数量

zfdf.createOrReplaceTempView("zftable")

#使用zftable进行分组统计

sql="select xqname,count(*) fynum from zftable group by xqname"

zfdf1=spark.sql(sql)

zfdf1.show()

#将统计的结果dataframe对象转为集合对象

list=zfdf1.collect()

print(list)

keyword_list=[]

for zfrow in list:

keyword_list.append((zfrow['xqname'],zfrow['fynum']))

print(keyword_list)

# 1. 高频词汇词云

word_cloud = (

WordCloud()

# '传入列表，word_size_range为字体大小，shape为词云的形状'

.add("", keyword_list, word_size_range=[15, 100], shape=SymbolType.DIAMOND)

.set_global_opts(title_opts=opts.TitleOpts(title="小区的房源统计的词云图"))

)

word_cloud.render('词云图.html') #'默认在当前目录生成一个render.html，也可以自定义文件名称'

正在上传…重新上传取消

10.9.使用pyecharts绘制上下组合

from pyecharts.charts import Pie, Bar, Map, WordCloud,Line,Grid#词云图

from pyecharts import options as opts

from pyecharts.faker import Faker

from pyecharts.globals import ThemeType

from pyecharts.globals import SymbolType

if __name__ == '__main__':

# 1、准备数据

country = ['巴西', '俄罗斯', '印度', '英国', '西班牙', '伊朗', '德国', '土耳其', '法国']

quezheng = [923189,553301,354065,298136,244328,192439,188252,181298,157716]

siwang = [45241,7478,11903,41969,27136,9065,8802,4842,29547]

# 2、绘制柱形图

bar = (

Bar()

.add_xaxis(country)

.add_yaxis("确诊人数", quezheng)

.add_yaxis("死亡人数", siwang)

.set_global_opts(title_opts=opts.TitleOpts(title="我是标题", subtitle="我是副标题"))

)

# 3、绘制线图

line = (

Line()

.add_xaxis(country)

.add_yaxis("quzheng人数", quezheng)

.add_yaxis("siwang人数", siwang)

.set_global_opts(legend_opts=opts.LegendOpts(pos_bottom="45%")))

# 4、创建组合图

(Grid(init_opts=opts.InitOpts(width='750px', height='350px'))

.add(bar, grid_opts=opts.GridOpts(pos_bottom="60%"))

.add(line, grid_opts=opts.GridOpts(pos_top="60%"))

).render("zuhetu.html")

正在上传…重新上传取消

10.10.使用pyecharts绘制左右组合图

from pyecharts.charts import Pie, Bar, Map, WordCloud,Line,Grid#词云图

from pyecharts import options as opts

from pyecharts.faker import Faker

from pyecharts.globals import ThemeType

from pyecharts.globals import SymbolType

if __name__ == '__main__':

# 1、准备数据

country = ['巴西', '俄罗斯', '印度', '英国', '西班牙', '伊朗', '德国', '土耳其', '法国']

quezheng = [923189,553301,354065,298136,244328,192439,188252,181298,157716]

siwang = [45241,7478,11903,41969,27136,9065,8802,4842,29547]

# 2、绘制柱形图

bar = (

Bar()

.add_xaxis(country)

.add_yaxis("确诊人数", quezheng, label_opts=opts.LabelOpts(is_show=False))

.add_yaxis("死亡人数", siwang, label_opts=opts.LabelOpts(is_show=False))

.set_global_opts(title_opts=opts.TitleOpts(title="我是标题", subtitle="我是副标题"),

legend_opts=opts.LegendOpts(pos_right="20%"))

)

# 3、绘制线图

line = (

Line()

.add_xaxis(country)

.add_yaxis("quzheng人数", quezheng, label_opts=opts.LabelOpts(is_show=False))

.add_yaxis("siwang人数", siwang, label_opts=opts.LabelOpts(is_show=False))

.set_global_opts(legend_opts=opts.LegendOpts(pos_left="20%"))

)

# 4、创建组合图

(Grid(init_opts=opts.InitOpts(width='750px', height='350px'))

.add(bar, grid_opts=opts.GridOpts(pos_left="55%"))

.add(line, grid_opts=opts.GridOpts(pos_right="55%"))

).render("zuhetu1.html")

正在上传…重新上传取消

10.11.使用pyecharts绘制一轴多图

from pyecharts.charts import Pie, Bar, Map, WordCloud,Line,Grid#词云图

from pyecharts import options as opts

from pyecharts.faker import Faker

from pyecharts.globals import ThemeType

from pyecharts.globals import SymbolType

if __name__ == '__main__':

# 1、准备数据

country = ['巴西', '俄罗斯', '印度', '英国', '西班牙', '伊朗', '德国', '土耳其', '法国']

quezheng = [923189,553301,354065,298136,244328,192439,188252,181298,157716]

siwang = [45241,7478,11903,41969,27136,9065,8802,4842,29547]

# 2、绘制柱形图

bar = (

Bar(init_opts=opts.InitOpts(width='750px', height='350px'))

.add_xaxis(country)

.add_yaxis("确诊人数", quezheng)

.add_yaxis("死亡人数", siwang)

.set_global_opts(title_opts=opts.TitleOpts(title="我是标题", subtitle="我是副标题"))

)

label_opts=opts.LabelOpts(is_show=False)

# 3、绘制线图

line = (

Line()

.add_xaxis(country)

.add_yaxis("确诊人数", quezheng, label_opts=opts.LabelOpts(is_show=False))

.add_yaxis("死亡人数", siwang, label_opts=opts.LabelOpts(is_show=False))

)

# 4、创建组合图

bar.overlap(line).render('test.html')

正在上传…重新上传取消

练习课堂案例

使用Django、mysql、spark、pyechart构建租房的分析与呈现

11.1.使用爬虫先爬取安居客租房网的信息到mysql数据库

11.2.使用django分页显示爬取的信息

11.3.使用spark访问mysql数据库中安居客租房的信息，并对mysql中数据库表的信息转换为dataframe进行分析。

11.3.1.代码如下:bingtu.html

<!DOCTYPE html>

<head>

<title>按照小区名称统计平均租金的饼形图</title>

</head>

<body>

// 基于准备好的dom，初始化echarts实例

var myChart = echarts.init(document.getElementById('main'));

console.log(jsdata)

// 指定图表的配置项和数据

var option = {

title: {

text: ' 按照小区名称统计平均租金的饼形图',

subtext: '平均租金饼图',

left: 'center'

tooltip: {

trigger: 'item'

legend: {

orient: 'vertical',

left: 'left'

series: [

{

name: 'Access From',

type: 'pie',

radius: '50%',

data: {{ datas|safe }},

emphasis: {

itemStyle: {

shadowBlur: 10,

shadowOffsetX: 0,

shadowColor: 'rgba(0, 0, 0, 0.5)'

}

]

};

// 使用刚指定的配置项和数据显示图表。

myChart.setOption(option);

</script>

</body>

</html>

data: {{ datas|safe }},其中safe 表示不要将json中的单引号进行转义

11.3.2.views.py中代码如下

from django.shortcuts import render

import pymysql #导入数据库访问组件

from pyspark.sql import SparkSession

from pyspark import SparkContext, SQLContext

from pyecharts.charts import Pie, Bar, Map, WordCloud,Line #词云图

from pyecharts import options as opts

from pyecharts.faker import Faker

from pyecharts.globals import ThemeType

from pyecharts.globals import SymbolType

#将表的记录值转为json对象的函数

def convertToJsonObject(obj):

jsonobj={

'zid':obj[0],

'title':obj[1],

'shi': obj[2],

'ting': obj[3],

'louceng': obj[4],

'mianji': obj[5],

'price': obj[6],

'address': obj[7],

'xiaoquname': obj[8],

'note': obj[9]

}

return jsonobj

#转换为饼图需要的json对象

def convertToJsonGroupByxqmc(obj):

jsonobj={

"name":obj['xiaoquname'],

"value":obj['avgpric']

}

return jsonobj

# Create your views here.

def showAll(request):

#获取数据库的链接对象

conn=pymysql.connect(host='localhost',port=3307,user='root',password='java',db="zfdb",charset='utf8')

#获取游标对象

cs=conn.cursor()

#获取当前页数和每页记录数

spage=request.GET.get('page')

spagesize=request.GET.get('pagesize')

if spage==None:

spage=1

if spagesize==None:

spagesize=5

page=int(spage)

pagesize=int(spagesize)

print(page)

print(pagesize)

if page<1:

page=1

if pagesize<1:

pagesize=5

if pagesize>20:#限制每页最多获取20条记录

pagesize=20

#编写sql语句(mysql的分页语句)

sql="SELECT * FROM zf where 1=1 order by zid limit %s,%s"

#计算当前页记录位置

pagenum=(page-1)*pagesize

#执行

cs.execute(sql,(pagenum,pagesize))

#获取结果集对象

rst=cs.fetchall()

print(rst)

#上页值

uppage=page-1

if uppage<1 :

uppage=1

#下页值

nextpage=page+1

#获取总记录数

sql1="select count(*) from zf"

cs.execute(sql1)

maxrow=cs.fetchone()[0]#获取总记录数

print("maxrow--->",maxrow)

#计算总页数================

maxpage=1

if maxrow%pagesize==0:

maxpage=maxrow/pagesize

else:

maxpage=maxrow/pagesize+1

if maxpage==0:

maxpage=1

#==========================

if nextpage>maxpage:#下页值大于最大页数，则下页值就是最大页数值

nextpage=maxpage

if page>maxpage:#限制当前页数不能大于总页数

page=maxpage

#将结果集合转为json集合

lszf=[]

for zfobj in rst:

obj=convertToJsonObject(zfobj)#调用转换函数，将表的记录转为json对象

lszf.append(obj)#将json对象设置到数组中

print(lszf)

#将分页数据封装到json对象pg对象中

pgobj={'lszf':lszf,'page':page,'pagesize':pagesize,'uppage':uppage,'nextpage':nextpage,'maxpage':int(maxpage)}

return render(request,"zufanglist.html",{'pg':pgobj})

#按照小区名称统计平均租金的饼形图

def showPipdemo(request):

#创建spark对象

spark=SparkSession.builder.appName("myzfapp").master("local").getOrCreate()

sc=spark.sparkContext#获取spark上下文关系对象

sqlsc = SQLContext(sc) # 创建spark之sql的上下文关系对象

df = sqlsc.read.format("jdbc").options(url="jdbc:mysql://localhost:3307/zfdb?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&user=root&password=java",dbtable="zf").load()

#df.show()

#创建临时数据表

df.createOrReplaceTempView("zttable")

sql='select xiaoquname,avg(price) avgpric from zttable group by xiaoquname'

zfgroupdf=spark.sql(sql)

#zfgroupdf.show()

listrow=zfgroupdf.collect()

#print(listrow)

listzftotal=[]

for zfrow in listrow:

jsonobj=convertToJsonGroupByxqmc(zfrow)#将小区名称和平均价转为json对象[{'name': '香榭兰岛', 'value': 1850.0}, {'name': '高山流水和城', 'value': 766.6666666666666}, ...]

listzftotal.append(jsonobj)#设置大数组中

print(listzftotal)

#data = listzftotal.decode("unicode-escape")

return render(request,'bingtu.html',{'datas':listzftotal})

11.3.3.在urls.py文件中设置访问路径

"""prjzufangwang URL Configuration

The `urlpatterns` list routes URLs to views. For more information please see:

https://docs.djangoproject.com/en/4.0/topics/http/urls/

Examples:

Function views

1. Add an import: from my_app import views

2. Add a URL to urlpatterns: path('', views.home, name='home')

Class-based views

1. Add an import: from other_app.views import Home

2. Add a URL to urlpatterns: path('', Home.as_view(), name='home')

Including another URLconf

1. Import the include() function: from django.urls import include, path

2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))

"""

# from django.contrib import admin

from django.urls import path

from myzufang_app import views

urlpatterns = [

# path('admin/', admin.site.urls),

path('',views.showAll),

path('showPipdemo/',views.showPipdemo),

]

11.3.4.在zufanglist.html页面中编写调用该方法的按钮

<!DOCTYPE html>

<head>

<title>租房信息列表</title>

$(function(){

//点击更改每页记录数的按钮

$("#btpagesize").click(function(){

//获取每页记录数文本框中的值

var cpsize=$("#pagesize").val();

if(isNaN(cpsize)){

alert('请输入正确的整数值！');

$("#pagesize").val({{ pg.pagesize }});

return;

}

window.location='/?page={{ pg.page }}&pagesize='+cpsize;

});

//点击跳转到第n页的事件

$("#btpage").click(function () {

//获取文本框中的页数值

var cpage=$("#page").val();

if(isNaN(cpage)){

alert('请输入正确的整数数值！');

$("#page").val({{ pg.page }});

return;

}

window.location='/?page='+cpage+'&pagesize={{ pg.pagesize }}';

});

</script>

</head>

<body>

<h1 align="center">房源信息列表</h1>

</tr>

{% for zf in pg.lszf %}

<td>{{ zf.title }}</td>

<td>{{ zf.louceng }}</td>

<td>{{ zf.mianji }}平方米</td>

<td>{{ zf.price }}元</td>

<td>{{ zf.address }}</td>

<td>{{ zf.xiaoquname }}</td>

</tr>

{% endfor %}

</table>

<td>

{% if pg.page > 1 %}

{% endif %}

首页</a></td>

<td>

{% if pg.page > 1 %}

{% endif %}

上页</a></td>

<td>

{% if pg.page < pg.maxpage %}

{% endif %}

下页</a></td>

<td>

{% if pg.page < pg.maxpage %}

{% endif %}

末页</a></td>

<td>

每页<input type="text" name="pagesize" value="{{ pg.pagesize }}" size="2" id="pagesize">条记录

</td>

<td>跳转到第<input type="text" name="page" value="{{ pg.page }}" size="2" id="page">页

<td>{{ pg.page }}/{{ pg.maxpage }}页</td>

</tr>

</table>

<table align="center" width="1200" border="1" cellspacing="1" cellpadding="1">

<tr align="center">

<td><input class="btn-lg" type="button" value="按照小区名称统计平均租金的饼形图" οnclick="findavgpriceByAXMC()"></td>

正在上传…重新上传取消 <td><input type="button" value="按照小区名称统计小区房源数量柱状图"></td>

<td><input type="button" value="按照楼层统计小区平均租金的折线图"></td>

<td><input type="button" value="按照小区名称的房源数量绘制词云图"></td>

</tr>

</table>

//按照小区名称统计平均组件的事件

function findavgpriceByAXMC(){

window.location="/showPipdemo/"

}

</script>

</body>

</html>

11.4.将分析的结果使用pyecharts生成饼形图、柱状图、折线图、词语图进行呈现

正在上传…重新上传取消

11.5.在views.py中编写柱状图的方法

11.5.1.views.py中构建柱状图的方法

from django.shortcuts import render

import pymysql #导入数据库访问组件

from pyspark.sql import SparkSession

from pyspark import SparkContext, SQLContext

from pyecharts.charts import Pie, Bar, Map, WordCloud,Line #词云图

from pyecharts import options as opts

from pyecharts.faker import Faker

from pyecharts.globals import ThemeType

from pyecharts.globals import SymbolType

#将表的记录值转为json对象的函数

def convertToJsonObject(obj):

jsonobj={

'zid':obj[0],

'title':obj[1],

'shi': obj[2],

'ting': obj[3],

'louceng': obj[4],

'mianji': obj[5],

'price': obj[6],

'address': obj[7],

'xiaoquname': obj[8],

'note': obj[9]

}

return jsonobj

#转换为饼图需要的json对象

def convertToJsonGroupByxqmc(obj):

jsonobj={

"name":obj['xiaoquname'],

"value":obj['avgpric']

}

return jsonobj

# Create your views here.

def showAll(request):

#获取数据库的链接对象

conn=pymysql.connect(host='localhost',port=3307,user='root',password='java',db="zfdb",charset='utf8')

#获取游标对象

cs=conn.cursor()

#获取当前页数和每页记录数

spage=request.GET.get('page')

spagesize=request.GET.get('pagesize')

if spage==None:

spage=1

if spagesize==None:

spagesize=5

page=int(spage)

pagesize=int(spagesize)

print(page)

print(pagesize)

if page<1:

page=1

if pagesize<1:

pagesize=5

if pagesize>20:#限制每页最多获取20条记录

pagesize=20

#编写sql语句(mysql的分页语句)

sql="SELECT * FROM zf where 1=1 order by zid limit %s,%s"

#计算当前页记录位置

pagenum=(page-1)*pagesize

#执行

cs.execute(sql,(pagenum,pagesize))

#获取结果集对象

rst=cs.fetchall()

print(rst)

#上页值

uppage=page-1

if uppage<1 :

uppage=1

#下页值

nextpage=page+1

#获取总记录数

sql1="select count(*) from zf"

cs.execute(sql1)

maxrow=cs.fetchone()[0]#获取总记录数

print("maxrow--->",maxrow)

#计算总页数================

maxpage=1

if maxrow%pagesize==0:

maxpage=maxrow/pagesize

else:

maxpage=maxrow/pagesize+1

if maxpage==0:

maxpage=1

#==========================

if nextpage>maxpage:#下页值大于最大页数，则下页值就是最大页数值

nextpage=maxpage

if page>maxpage:#限制当前页数不能大于总页数

page=maxpage

#将结果集合转为json集合

lszf=[]

for zfobj in rst:

obj=convertToJsonObject(zfobj)#调用转换函数，将表的记录转为json对象

lszf.append(obj)#将json对象设置到数组中

print(lszf)

#将分页数据封装到json对象pg对象中

pgobj={'lszf':lszf,'page':page,'pagesize':pagesize,'uppage':uppage,'nextpage':nextpage,'maxpage':int(maxpage)}

return render(request,"zufanglist.html",{'pg':pgobj})

#按照小区名称统计平均租金的饼形图

def showPipdemo(request):

#创建spark对象

spark=SparkSession.builder.appName("myzfapp").master("local").getOrCreate()

sc=spark.sparkContext#获取spark上下文关系对象

sqlsc = SQLContext(sc) # 创建spark之sql的上下文关系对象

df = sqlsc.read.format("jdbc").options(url="jdbc:mysql://localhost:3307/zfdb?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&user=root&password=java",dbtable="zf").load()

#df.show()

#创建临时数据表

df.createOrReplaceTempView("zttable")

sql='select xiaoquname,avg(price) avgpric from zttable group by xiaoquname'

zfgroupdf=spark.sql(sql)

#zfgroupdf.show()

listrow=zfgroupdf.collect()

#print(listrow)

listzftotal=[]

for zfrow in listrow:

jsonobj=convertToJsonGroupByxqmc(zfrow)#将小区名称和平均价转为json对象[{'name': '香榭兰岛', 'value': 1850.0}, {'name': '高山流水和城', 'value': 766.6666666666666}, ...]

listzftotal.append(jsonobj)#设置大数组中

print(listzftotal)

#data = listzftotal.decode("unicode-escape")

return render(request,'bingtu.html',{'datas':listzftotal})

#编写按照小区名称统计小区房源数量柱状图的方法

def showBarByxqmc(request):

# 创建spark对象

spark = SparkSession.builder.appName("myzfapp").master("local").getOrCreate()

sc = spark.sparkContext # 获取spark上下文关系对象

sqlsc = SQLContext(sc) # 创建spark之sql的上下文关系对象

df = sqlsc.read.format("jdbc").options(

url="jdbc:mysql://localhost:3307/zfdb?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&user=root&password=java",

dbtable="zf").load()

# df.show()

# 创建临时数据表

df.createOrReplaceTempView("zttable")

sql = 'select xiaoquname,avg(price) avgpric from zttable group by xiaoquname'

zfgroupdf = spark.sql(sql)

# zfgroupdf.show()

listrow = zfgroupdf.collect()

# print(listrow)

xqmclist = []

xqjiagelist=[]

for zfrow in listrow:

xqmclist.append(zfrow['xiaoquname'])#['香榭兰岛', '高山流水和城', '天地源万熙天地', '陕西省测绘地理信息局住宅区', '奥达文景观园', ..]

xqjiagelist.append(zfrow['avgpric'])#[1850.0, 766.6666666666666, 616.6666666666666,...]

print(xqmclist)

print(xqjiagelist)

return render(request,'zhuzhuangtu.html',{'xqnames':xqmclist,'jiages':xqjiagelist})

11.5.2.在urls.py中设置访问路径

"""prjzufangwang URL Configuration

The `urlpatterns` list routes URLs to views. For more information please see:

https://docs.djangoproject.com/en/4.0/topics/http/urls/

Examples:

Function views

1. Add an import: from my_app import views

2. Add a URL to urlpatterns: path('', views.home, name='home')

Class-based views

1. Add an import: from other_app.views import Home

2. Add a URL to urlpatterns: path('', Home.as_view(), name='home')

Including another URLconf

1. Import the include() function: from django.urls import include, path

2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))

"""

# from django.contrib import admin

from django.urls import path

from myzufang_app import views

urlpatterns = [

# path('admin/', admin.site.urls),

path('',views.showAll),

path('showPipdemo/',views.showPipdemo),#显示饼形图的路径

path('showBarByxqmc/',views.showBarByxqmc),#显示柱状图的路径

]

11.5.3.zhuzhuangtu.html页面

<!DOCTYPE html>

<head>

<title>按照小区名称统计小区房源数量柱状图</title>

<script src="/static/js/echarts.js"></script>

</head>

<body>

// 基于准备好的dom，初始化echarts实例

var myChart = echarts.init(document.getElementById('main'));

var dataxqs={{ xqnames|safe}};

console.log(dataxqs);

// 指定图表的配置项和数据

var option = {

xAxis: {

type: 'category',

data: {{ xqnames|safe }}

yAxis: {

type: 'value'

series: [

{

data: {{ jiages|safe }},

type: 'bar'

}

]

};

// 使用刚指定的配置项和数据显示图表。

myChart.setOption(option);

</script>

</body>

</html>

11.5.4.在列表页面添加事件访问

<!DOCTYPE html>

<head>

<title>租房信息列表</title>

$(function(){

//点击更改每页记录数的按钮

$("#btpagesize").click(function(){

//获取每页记录数文本框中的值

var cpsize=$("#pagesize").val();

if(isNaN(cpsize)){

alert('请输入正确的整数值！');

$("#pagesize").val({{ pg.pagesize }});

return;

}

window.location='/?page={{ pg.page }}&pagesize='+cpsize;

});

//点击跳转到第n页的事件

$("#btpage").click(function () {

//获取文本框中的页数值

var cpage=$("#page").val();

if(isNaN(cpage)){

alert('请输入正确的整数数值！');

$("#page").val({{ pg.page }});

return;

}

window.location='/?page='+cpage+'&pagesize={{ pg.pagesize }}';

});

</script>

</head>

<body>

<h1 align="center">房源信息列表</h1>

</tr>

{% for zf in pg.lszf %}

<td>{{ zf.title }}</td>

<td>{{ zf.louceng }}</td>

<td>{{ zf.mianji }}平方米</td>

<td>{{ zf.price }}元</td>

<td>{{ zf.address }}</td>

<td>{{ zf.xiaoquname }}</td>

</tr>

{% endfor %}

</table>

<td>

{% if pg.page > 1 %}

{% endif %}

首页</a></td>

<td>

{% if pg.page > 1 %}

{% endif %}

上页</a></td>

<td>

{% if pg.page < pg.maxpage %}

{% endif %}

下页</a></td>

<td>

{% if pg.page < pg.maxpage %}

{% endif %}

末页</a></td>

<td>

每页<input type="text" name="pagesize" value="{{ pg.pagesize }}" size="2" id="pagesize">条记录

</td>

<td>跳转到第<input type="text" name="page" value="{{ pg.page }}" size="2" id="page">页

<td>{{ pg.page }}/{{ pg.maxpage }}页</td>

</tr>

</table>

</tr>

</table>

//按照小区名称统计平均组件的事件

function findavgpriceByAXMC(){

window.location="/showPipdemo/"

}

//按照小区名称统计平均组件的事件

function findBarByxqmc(){

window.location="/showBarByxqmc/"

}

</script>

</body>

</html>

正在上传…重新上传取消

11.6.绘制折线图

Views.py

from django.shortcuts import render

import pymysql #导入数据库访问组件

from pyspark.sql import SparkSession

from pyspark import SparkContext, SQLContext

from pyecharts.charts import Pie, Bar, Map, WordCloud,Line #词云图

from pyecharts import options as opts

from pyecharts.faker import Faker

from pyecharts.globals import ThemeType

from pyecharts.globals import SymbolType

#将表的记录值转为json对象的函数

def convertToJsonObject(obj):

jsonobj={

'zid':obj[0],

'title':obj[1],

'shi': obj[2],

'ting': obj[3],

'louceng': obj[4],

'mianji': obj[5],

'price': obj[6],

'address': obj[7],

'xiaoquname': obj[8],

'note': obj[9]

}

return jsonobj

#转换为饼图需要的json对象

def convertToJsonGroupByxqmc(obj):

jsonobj={

"name":obj['xiaoquname'],

"value":obj['avgpric']

}

return jsonobj

# Create your views here.

def showAll(request):

#获取数据库的链接对象

conn=pymysql.connect(host='localhost',port=3307,user='root',password='java',db="zfdb",charset='utf8')

#获取游标对象

cs=conn.cursor()

#获取当前页数和每页记录数

spage=request.GET.get('page')

spagesize=request.GET.get('pagesize')

if spage==None:

spage=1

if spagesize==None:

spagesize=5

page=int(spage)

pagesize=int(spagesize)

print(page)

print(pagesize)

if page<1:

page=1

if pagesize<1:

pagesize=5

if pagesize>20:#限制每页最多获取20条记录

pagesize=20

#编写sql语句(mysql的分页语句)

sql="SELECT * FROM zf where 1=1 order by zid limit %s,%s"

#计算当前页记录位置

pagenum=(page-1)*pagesize

#执行

cs.execute(sql,(pagenum,pagesize))

#获取结果集对象

rst=cs.fetchall()

print(rst)

#上页值

uppage=page-1

if uppage<1 :

uppage=1

#下页值

nextpage=page+1

#获取总记录数

sql1="select count(*) from zf"

cs.execute(sql1)

maxrow=cs.fetchone()[0]#获取总记录数

print("maxrow--->",maxrow)

#计算总页数================

maxpage=1

if maxrow%pagesize==0:

maxpage=maxrow/pagesize

else:

maxpage=maxrow/pagesize+1

if maxpage==0:

maxpage=1

#==========================

if nextpage>maxpage:#下页值大于最大页数，则下页值就是最大页数值

nextpage=maxpage

if page>maxpage:#限制当前页数不能大于总页数

page=maxpage

#将结果集合转为json集合

lszf=[]

for zfobj in rst:

obj=convertToJsonObject(zfobj)#调用转换函数，将表的记录转为json对象

lszf.append(obj)#将json对象设置到数组中

print(lszf)

#将分页数据封装到json对象pg对象中

pgobj={'lszf':lszf,'page':page,'pagesize':pagesize,'uppage':uppage,'nextpage':nextpage,'maxpage':int(maxpage)}

return render(request,"zufanglist.html",{'pg':pgobj})

#按照小区名称统计平均租金的饼形图

def showPipdemo(request):

#创建spark对象

spark=SparkSession.builder.appName("myzfapp").master("local").getOrCreate()

sc=spark.sparkContext#获取spark上下文关系对象

sqlsc = SQLContext(sc) # 创建spark之sql的上下文关系对象

df = sqlsc.read.format("jdbc").options(url="jdbc:mysql://localhost:3307/zfdb?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&user=root&password=java",dbtable="zf").load()

#df.show()

#创建临时数据表

df.createOrReplaceTempView("zttable")

sql='select xiaoquname,avg(price) avgpric from zttable group by xiaoquname'

zfgroupdf=spark.sql(sql)

#zfgroupdf.show()

listrow=zfgroupdf.collect()

#print(listrow)

listzftotal=[]

for zfrow in listrow:

jsonobj=convertToJsonGroupByxqmc(zfrow)#将小区名称和平均价转为json对象[{'name': '香榭兰岛', 'value': 1850.0}, {'name': '高山流水和城', 'value': 766.6666666666666}, ...]

listzftotal.append(jsonobj)#设置大数组中

print(listzftotal)

#data = listzftotal.decode("unicode-escape")

return render(request,'bingtu.html',{'datas':listzftotal})

#编写按照小区名称统计小区房源数量柱状图的方法

def showBarByxqmc(request):

# 创建spark对象

spark = SparkSession.builder.appName("myzfapp").master("local").getOrCreate()

sc = spark.sparkContext # 获取spark上下文关系对象

sqlsc = SQLContext(sc) # 创建spark之sql的上下文关系对象

df = sqlsc.read.format("jdbc").options(

url="jdbc:mysql://localhost:3307/zfdb?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&user=root&password=java",

dbtable="zf").load()

# df.show()

# 创建临时数据表

df.createOrReplaceTempView("zttable")

sql = 'select xiaoquname,avg(price) avgpric from zttable group by xiaoquname'

zfgroupdf = spark.sql(sql)

# zfgroupdf.show()

listrow = zfgroupdf.collect()

# print(listrow)

xqmclist = []

xqjiagelist=[]

for zfrow in listrow:

xqmclist.append(zfrow['xiaoquname'])#['香榭兰岛', '高山流水和城', '天地源万熙天地', '陕西省测绘地理信息局住宅区', '奥达文景观园', ..]

xqjiagelist.append(zfrow['avgpric'])#[1850.0, 766.6666666666666, 616.6666666666666,...]

print(xqmclist)

print(xqjiagelist)

return render(request,'zhuzhuangtu.html',{'xqnames':xqmclist,'jiages':xqjiagelist})

#显示词云图的方法

def showciyun(request):

return render(request,'ciyuntu.html')

#显示折线图的方法,根据校区的名称统计房源数量

def showLine(request):

# 创建spark对象

spark = SparkSession.builder.appName("myzfapp").master("local").getOrCreate()

sc = spark.sparkContext # 获取spark上下文关系对象

sqlsc = SQLContext(sc) # 创建spark之sql的上下文关系对象

df = sqlsc.read.format("jdbc").options(

url="jdbc:mysql://localhost:3307/zfdb?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&user=root&password=java",

dbtable="zf").load()

# df.show()

# 创建临时数据表

df.createOrReplaceTempView("zttable")

sql = 'select xiaoquname,count(*) nums from zttable group by xiaoquname'

zfgroupdf = spark.sql(sql)

# zfgroupdf.show()

listrow = zfgroupdf.collect()

# print(listrow)

xqmclist = []

numslist = []

for zfrow in listrow:

xqmclist.append(zfrow['xiaoquname']) # ['香榭兰岛', '高山流水和城', '天地源万熙天地', '陕西省测绘地理信息局住宅区', '奥达文景观园', ..]

numslist.append(zfrow['nums']) # [1850.0, 766.6666666666666, 616.6666666666666,...]

print(xqmclist)

print(numslist)

return render(request, 'zhexiantu.html', {'xqnames': xqmclist, 'nums': numslist})

Urls.py中显示折线图的路径:

"""prjzufangwang URL Configuration

The `urlpatterns` list routes URLs to views. For more information please see:

https://docs.djangoproject.com/en/4.0/topics/http/urls/

Examples:

Function views

1. Add an import: from my_app import views

2. Add a URL to urlpatterns: path('', views.home, name='home')

Class-based views

1. Add an import: from other_app.views import Home

2. Add a URL to urlpatterns: path('', Home.as_view(), name='home')

Including another URLconf

1. Import the include() function: from django.urls import include, path

2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))

"""

# from django.contrib import admin

from django.urls import path

from myzufang_app import views

urlpatterns = [

# path('admin/', admin.site.urls),

path('',views.showAll),

path('showPipdemo/',views.showPipdemo),#显示饼形图的路径

path('showBarByxqmc/',views.showBarByxqmc),#显示柱状图的路径

path('showciyun/',views.showciyun),#显示词云图

path('showLine/',views.showLin),#显示折线图

]

Zhexiantu.html

<!DOCTYPE html>

<head>

<title>按照小区名称统计小区房源数量折线图</title>

</head>

<body>

// 基于准备好的dom，初始化echarts实例

var myChart = echarts.init(document.getElementById('main'));

var dataxqs={{ xqnames|safe}};

console.log(dataxqs);

// 指定图表的配置项和数据

var option = {

xAxis: {

type: 'category',

data: {{ xqnames|safe }}

yAxis: {

type: 'value'

series: [

{

data: {{ nums|safe }},

type: 'line'

}

]

};

// 使用刚指定的配置项和数据显示图表。

myChart.setOption(option);

</script>

</body>

</html>

修改zufanglist.html页面加入访问路径

<!DOCTYPE html>

<head>

<title>租房信息列表</title>

$(function(){

//点击更改每页记录数的按钮

$("#btpagesize").click(function(){

//获取每页记录数文本框中的值

var cpsize=$("#pagesize").val();

if(isNaN(cpsize)){

alert('请输入正确的整数值！');

$("#pagesize").val({{ pg.pagesize }});

return;

}

window.location='/?page={{ pg.page }}&pagesize='+cpsize;

});

//点击跳转到第n页的事件

$("#btpage").click(function () {

//获取文本框中的页数值

var cpage=$("#page").val();

if(isNaN(cpage)){

alert('请输入正确的整数数值！');

$("#page").val({{ pg.page }});

return;

}

window.location='/?page='+cpage+'&pagesize={{ pg.pagesize }}';

});

</script>

</head>

<body>

<h1 align="center">房源信息列表</h1>

</tr>

{% for zf in pg.lszf %}

<td>{{ zf.title }}</td>

<td>{{ zf.louceng }}</td>

<td>{{ zf.mianji }}平方米</td>

<td>{{ zf.price }}元</td>

<td>{{ zf.address }}</td>

<td>{{ zf.xiaoquname }}</td>

</tr>

{% endfor %}

</table>

<td>

{% if pg.page > 1 %}

{% endif %}

首页</a></td>

<td>

{% if pg.page > 1 %}

{% endif %}

上页</a></td>

<td>

{% if pg.page < pg.maxpage %}

{% endif %}

下页</a></td>

<td>

{% if pg.page < pg.maxpage %}

{% endif %}

末页</a></td>

<td>

每页<input type="text" name="pagesize" value="{{ pg.pagesize }}" size="2" id="pagesize">条记录

</td>

<td>跳转到第<input type="text" name="page" value="{{ pg.page }}" size="2" id="page">页

<td>{{ pg.page }}/{{ pg.maxpage }}页</td>

</tr>

</table>

</tr>

</table>

//按照小区名称统计平均组件的事件

function findavgpriceByAXMC(){

window.location="/showPipdemo/"

}

//按照小区名称统计平均组件的事件

function findBarByxqmc(){

window.location="/showBarByxqmc/"

}

//显示词云图

function showciyun(){

window.location="/showciyun/";

}

//显示折线图

function showline(){

window.location="/showLine/";

}

</script>

</body>

</html>

正在上传…重新上传取消

效果如下:

正在上传…重新上传取消

完成课堂案例

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/772383

spark数据分析基础_spark大数据分析

一.Spark概述

2.1.spark的运行原理图

2.2.spark与hadoop之间的对比

3.1.spark的运行环境

3.2.spark的部署模式

（1）local(本地模式)：常用于本地开发测试，本地还分为local单线程和local-cluster多线程

（2）standalone(集群模式)：典型的Mater/slave模式，不过也能看出Master是有单点故障的；Spark支持ZooKeeper来实现 HA

（3）on yarn(集群模式)： 运行在 yarn 资源管理器框架之上，由 yarn 负责资源管理，Spark 负责任务调度和计算

（4）on mesos(集群模式)： 运行在 mesos 资源管理器框架之上，由 mesos 负责资源管理，Spark 负责任务调度和计算

3.3.在windows10下配置spark的运行环境

3.1.1.解压缩spark-3.0.3-bin-hadoop2.7(win10可用).tgz到指定的磁盘

3.3.2.spark的配置环境变量

3.3.3.window10下安装hadoop3

3.3.4.配置hadoop的配置文件

4.2.1在程序中设置环境变量

4.3.在pycharm中添加spark环境

5.1.RDD概念

5.2.RDD特性：

5.3.RDD分区运算

5.4.RDD分区的原则

5.5.RDD通过Lineage（血统）实现数据的容错性。

5.5.1.RDD依赖

（1）窄依赖

(2)宽依赖

5.6.获取RDD对象的步骤

6.1.算子分类

1、Value数据类型的Transformation算子，这种变换不触发提交作业，针对处理的数据项是Value型的数据。

2、Key-Value数据类型的Transformation算子，这种变换不触发提交作业，针对处理的数据项是Key-Value型的数据。

3、Action算子，这类算子会触发SparkContext提交作业。

6.1.1.map算子

6.1.2.flatMap算子

6.1.3.filter算子

6.1.4.mapPartitions

MapPartitions的优点：

6.1.5.union算子

6.1.5.cartesian(笛卡尔)算子

6.1.6.groupBy算子

6.1.7.distinct算子

6.1.8.subtract（差值运算）

6.1.9.sample样本算子

6.1.10.takeSample算子

6.2.spark中的key-value算子和action算子

6.2.1.mapValues算子

6.2.2.flatMapValues算子

6.2.3.combineByKey算子(统计算子)

6.2.4.foldByKey算子

6.2.5.reduceByKey算子

6.2.6.groupByKey算子

6.2.7.sortByKey算子

6.2.8.cogroup算子

6.2.9.join算子

6.2.10.leftOuterJoin

6.2.11.fullOuterJoin算子

7.1.foreach算子

7.2.collect算子

7.3.take算子

7.4.top算子

7.5.first算子

7.6.reduce算子

7.7.collectAsMap算子

7.8.countByKey算子

7.9.lookup算子（查找算子）

7.10.foldz算子

7.11.aggregate 算子

8.1.广播变量的主要作用

8.2.广播变量的定义方式

8.3.累加器

8.3.spark的输入与输出

8.3.1.spark支持的输入输出方式

8.3.2.spark读取文本文件和写出文本文件数据

(2)Analyzer（分析）

9.3.RDD、DataFrame 与DataSet

9.3.1.RDD

9.3.2.Dataframe

9.3.4.过滤DataFrme中指定列的数据

9.3.5.使用spark获取csv文件的数据转为DataFrame对象

9.4.2.读取json文件并转为DataSet集合数据

9.6.2.StructField

10.1.安装pyecharts

（3）on yarn(集群模式)：运行在 yarn 资源管理器框架之上，由 yarn 负责资源管理，Spark 负责任务调度和计算

（4）on mesos(集群模式)：运行在 mesos 资源管理器框架之上，由 mesos 负责资源管理，Spark 负责任务调度和计算