四：spark Sql， dataframe, Dataset
4.1:Spark SQL 的用法之一是执行 SQL 查询, 它也可以从现有的 Hive 中读取数据
SparkSession: Spark 中所有功能的入口是 SparkSession 类
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Python Spark SQL basic example").config("spark.some.config.option", "some-value").getOrCreate()
4.2: Dataset 和 DataFrame
Dataset 是一个分布式数据集，它是 Spark 1.6 版本中新增的一个接口, 它结合了 RDD（强类型，可以使用强大的 lambda 表达式函数） 和 Spark SQL 的优化执行引擎的好处。Dataset 可以从 JVM 对象构造得到，随后可以使用函数式的变换（map，flatMap，filter 等） 进行操作
DataFrame 是按命名列方式组织的一个 Dataset。从概念上来讲，它等同于关系型数据库中的一张表或者 R 和 Python 中的一个 data frame， 只不过在底层进行了更多的优化。DataFrame 可以从很多数据源构造得到，比如：结构化的数据文件，Hive 表，外部数据库或现有的 RDD
4.3：同过spark读取的文件是dataframe：
df = spark.read.json('./dense_report.json') 都json文件
df.show()
df.select('summary').show()
df.select(df['options'], df['summary']).show()
df.filter(df['summary']>20).show()
Count people by age 按年龄计算人数
df.groupBy('summary').count().show()
4.4：如何需要运用sql：
df.createOrReplaceTempView("people")
sqldf = spark.sql('select * from people')
或者：
df.createGlobalTempView("people")
spark.newSession().sql('select * from global_temp.people').show()
sqldf.show()
4.5如果读取的是txt文件：
sc = SparkContext('local', 'sql app')
lines = sc.textFile('data.txt')
parts = lines.map(lambda l: l.split(','))
people = parts.map(lambda p: Row(name=p[1], age=p[6]))
spark = SparkSession(sc)
schemaPeople = spark.createDataFrame(people)
schemaPeople.createOrReplaceTempView("people")
# schemaPeople.show()
teenagers = spark.sql('select * from people where age>500 and age<1000 ')
# 对查询的数据进行修改
teeNames = teenagers.rdd.map(lambda p: "Name: "+p.name).collect()
print(f"teeNames:{teeNames}")
for name in teeNames:
print(name)
4.6：Parquet 文件
Parquet 是一种列式存储格式，很多其它的数据处理系统都支持它。Spark SQL 提供了对 Parquet 文件的读写支持，而且 Parquet 文件能够自动保存原始数据的 schema。写 Parquet 文件的时候，所有列都自动地转化成 nullable，以便向后兼容
'''
将json文件转化为parquet，然后读取parquet文件创建视图进行查询
'''
sc = SparkContext('local', 'parquet')
spark = SparkSession(sc)
df = spark.read.json("./people.json")
df.write.parquet("people.parquet")
people_df = spark.read.parquet("people.parquet")
people_df.createOrReplaceTempView("people")
teenage = spark.sql("select * from people where age>=13 and age<=19")
teenage.show()
'''
生成2个parquet 文件， 将其进行合并
'''
sc = SparkContext('local', 'merge parquet')
spark = SparkSession(sc)
df1 = spark.createDataFrame(sc.parallelize(range(1, 6)).map(lambda l: Row(single=l, double=l**2)))
df1.write.parquet("./test_table/key=1")
df2 = spark.createDataFrame(sc.parallelize(range(6, 11)).map(lambda p: Row(single=p, triple=p**3)))
df2.write.parquet("./test_table/key=2")
merge_df = spark.read.option('mergeSchema', "true").parquet("./test_table")
# merge_df.printSchema()
merge_df.createOrReplaceTempView('merge')
merge_df.show()
# 或者直接在文件上运行sql
df = spark.sql("select * from parquet.`./users.parquet`")
4.7：spark集成hive从hive上读取数据
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").appName("hive app").enableHiveSupport().getOrCreate()
spark.sql("CREATE TABLE IF NOT EXISTS srcpp (key INT, value STRING) USING hive")
spark.sql("load data local inpath './kv1.txt' into table srcpp")
# Queries are expressed in HiveQL 查询以HiveQL表示
spark.sql('select * from srcpp').show()
其中关键要加上：enableHiveSupport()
4.8： spark集成mysql：
Spark SQL也包括一个可以使用JDBC从其它数据库读取数据的数据源。该功能应该优于使用JdbcRDD，因为它的返回结果是一个DataFrame，而在Spark SQL中DataFrame处理简单，且和其它数据源进行关联操作。JDBC数据源在Java和Python中用起来很简单，因为不需要用户提供一个ClassTag
连接数据库读取：
方法一：
dbcDF = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql:dbserver") \
.option("dbtable", "schema.tablename") \
.option("user", "username") \
.option("password", "password") \
.load()
方法二：
jdbcDF2 = spark.read \
.jdbc("jdbc:postgresql:dbserver", "schema.tablename",
properties={"user": "username", "password": "password"})
连接数据库写入：
方法一：
# Saving data to a JDBC source
jdbcDF.write \
.format("jdbc") \
.option("url", "jdbc:postgresql:dbserver") \
.option("dbtable", "schema.tablename") \
.option("user", "username") \
.option("password", "password") \
.save()
方法二
jdbcDF2.write \
.jdbc("jdbc:postgresql:dbserver", "schema.tablename",
properties={"user": "username", "password": "password"})
方法三
# Specifying create table column data types on write
jdbcDF.write \
.option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)") \
.jdbc("jdbc:postgresql:dbserver", "schema.tablename",
properties={"user": "username", "password": "password"})
例子：
sc = SparkContext('local', 'mysql view')
spark = SparkSession(sc)
df = spark.read.parquet("./people.parquet")
df.show()
# 将结果写进mysql数据库
url = "jdbc:mysql://127.0.0.1:3306/data_test?useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8"
table = "person"
mode = "overwrite"
properties = {'user': 'root', 'password': 'hgh15070417641'}
df.write.jdbc(url, table, mode, properties=properties)
其中需要下载mysql-connector-java.jar, 下载路径：
https://mvnrepository.com/artifact/mysql/mysql-connector-java
选择对应的版本 我的java版本：11.0.8 pyspark: 3.0.1, Scala version 2.12.10, 选择了mysql-connector-java-8.0.21.jar
4.9：性能调优：
对于有一定计算量的Spark任务，可以将数据放入内存缓存或开启一些试验选项来提升性能
通过调用 spark.cacheTable(“tableName”) 或者 dataFrame.cache() 方法, Spark SQL可以使用一种内存列存储格式缓存表。接着Spark SQL只扫描必要的列，并且自动调整压缩比例，以最小化内存占用和GC压力。你可以调用 spark.uncacheTable(“tableName”) 方法删除内存中的表
五：spark Streaming:
5.1 StreamingContext介绍：
from pyspark.streaming import StreamingContext
'''
StreamingContext(sparkContext, batchDuration=None, jssc=None
'''
sparkContext 功能入口点， batchDuration：持续时间
例子：
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
conf = SparkConf()\
.setMaster("local[*]")\
.setAppName("stream app")\
.set("spark.streaming.backpressure.enabled", "True")\
.set("spark.streaming.receiver.maxRate", "100")
sc = SparkContext.getOrCreate(conf)
ssc = StreamingContext(sc, 3)
# 形成DStream
dstream = ssc.socketTextStream('127.0.0.1', 9999)
# 词频统计
result = dstream.flatMap(lambda line: line.split("\t")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a+b)
result.pprint()
# 开启程序
ssc.start()
ssc.awaitTermination()
注意： 运行这个方法前需要线运行： nc -lk 9999
5.2： SparkStreaming 和kafka集成
spark-streaming-kafka-0-10 和 spark-streaming-kafka-0-8
1：spark-streaming-kafka-0-8 ：
SparkStreaming 和kafka集成有2种模式：
1： receiver 模式（数据接收器模式）
2： direct模式（直接读取数据）
注意： sparkstreaming的被压机制： 为了解决数据进来的太快， 集群计算速度跟不上， 可以开启背压机制来进行限速：set("spark.streaming.backpressure.enabled", "True")
receiver:set("spark.streaming.receiver.maxRate", "100") 表示接收数据最大的速率
direct模式： set("spark.streaming.kafka.maxRatePerPartition") : 表示sparkstreaming每个批次每一个分区每秒钟接受的数据的条数， 默认是无限制
