潍坊网站设计好处,域名服务器分为,北京冬奥会吉祥物设计制作,wordpress连接网盘插件一、sparkStreaming的不足
1.基于微批,延迟高不能做到真正的实时
2.DStream基于RDD,不直接支持SQL
3.流批处理的API应用层不统一,(流用的DStream-底层是RDD,批用的DF/DS/RDD)
4.不支持EventTime事件时间#xff08;一般流处理都会有两个时间#xff1a;事件发生的事件一般流处理都会有两个时间事件发生的事件一个是事件处理的时间
5.数据的Exactly-Once(恰好一次语义)需要手动实现
二、StructuredStreaming 的介绍 1、2016年Spark2.0版本中发布
2、基于SparkSQL引擎的可扩展、容错的全新的流处理引擎。
3、并不是对Spark Streaming的简单改进而是重新开发的全新流式引擎
准实时技术来一批处理一批 实时来一条处理一条 离线一般都是处理一些静止的数据
三、socketconsole
1、在虚拟机中下载nc
yum install -y nc2、启动 nc -lk 9999
案例wordcount
import osfrom pyspark.sql import SparkSession
from pyspark.sql.functions import explode
import pyspark.sql.functions as F
if __name__ __main__:os.environ[JAVA_HOME] E:/java-configuration/jdk-8# 配置Hadoop的路径就是前面解压的那个路径os.environ[HADOOP_HOME] E:/applications/bigdata_config/hadoop-3.3.1/hadoop-3.3.1# 配置base环境Python解析器的路径os.environ[PYSPARK_PYTHON] C:/Users/35741/miniconda3/python.exe# 配置base环境Python解析器的路径os.environ[PYSPARK_DRIVER_PYTHON] C:/Users/35741/miniconda3/python.exe# 创建一个sparkSession对象spark SparkSession.builder.appName(socketDemo).getOrCreate()socketDf spark.readStream.format(socket) \.option(host, bigdata01) \.option(port, 9999) \.load()# 处理# 方式一使用dsl语法splitDf socketDf.select(explode(F.split(socketDf.value, )).alias(word))resultDf1 splitDf.groupBy(word).count()# 方式二使用sqlsocketDf.createOrReplaceTempView(wordcount)resultDf2 spark.sql(with t1 as( select num from wordcount lateral view explode(split(value, )) c as num)select num,count(*) counts from t1 group by num;)# 下面的就是sink的写法 后续会写query1 resultDf1.writeStream \.outputMode(complete) \.format(console) \.start()query2 resultDf2.writeStream \.outputMode(complete) \.format(console) \.start() \.awaitTermination()spark.stop()
四、fileconsole
文件中的数据
1;yuwen;43
1;shuxue;55
2;yuwen;77
2;shuxue;88
3;yuwen;98
3;shuxue;65
3;yingyu;88
import osfrom pyspark.sql import SparkSessionfrom pyspark.sql.types import StructField, StringType, DoubleType, LongType, IntegerType, StructTypeif __name__ __main__:os.environ[JAVA_HOME] E:/java-configuration/jdk-8# 配置Hadoop的路径就是前面解压的那个路径os.environ[HADOOP_HOME] E:/applications/bigdata_config/hadoop-3.3.1/hadoop-3.3.1# 配置base环境Python解析器的路径os.environ[PYSPARK_PYTHON] C:/Users/35741/miniconda3/python.exe# 配置base环境Python解析器的路径os.environ[PYSPARK_DRIVER_PYTHON] C:/Users/35741/miniconda3/python.exe# 创建一个sparkSession对象spark SparkSession.builder.appName(socketDemo).getOrCreate()# score_schema StructType([# StructField(namestu_id, dataTypeIntegerType(), nullableFalse),# StructField(namesubject_name, dataTypeStringType(), nullableTrue),# StructField(namescore, dataTypeDoubleType(), nullableTrue)# ])score_schema StructType().add(stu_id, IntegerType()).add(subject_name, StringType()).add(score,DoubleType())socketDf spark.readStream.format(csv) \.option(sep, ;) \.schema(score_schema) \.load(../../resources/input1)socketDf.writeStream \.outputMode(append) \.format(console) \.option(truncate, False) \.start() \.awaitTermination()spark.stop()
