当前位置： 首页 > news >正文

网站做支付需要准备什么条件wordpress 读取最新文章

news 2025/9/16 15:44:36
网站做支付需要准备什么条件,wordpress 读取最新文章,百捷网站建设工资,青岛建设工程管理信息网3. DStream SparkStreaming模块将流式数据封装的数据结构：DStream（Discretized Stream，离散化数据流，连续不断的数据流），代表持续性的数据流和经过各种Spark算子操作后的结果数据流。 3.1 DStream 是什么…

3. DStream

SparkStreaming模块将流式数据封装的数据结构：DStream（Discretized Stream，离散化数据流，连续不断的数据流），代表持续性的数据流和经过各种Spark算子操作后的结果数据流。

3.1 DStream 是什么

离散数据流（DStream）是Spark Streaming最基本的抽象。它代表了一种连续的数据流，要么从某种数据源提取数据，要么从其他数据流映射转换而来。DStream内部是由一系列连续的RDD组成的，每个RDD都包含了特定时间间隔内的一批数据，如下图所示：
在这里插入图片描述
DStream本质上是一个：一系列时间上连续的RDD（Seq[RDD]），DStream = Seq[RDD]。

DStream = Seq[RDD]
DStream相当于一个序列（集合），里面存储的数据类型为RDD（Streaming按照时间间隔划分流式数据）

对DStream的数据进行操作也是按照RDD为单位进行的。
在这里插入图片描述
通过WEB UI界面可知，对DStream调用函数操作，底层就是对RDD进行操作，发现很多时候DStream中函数与RDD中函数一样的。
在这里插入图片描述
DStream中每批次数据RDD在处理时，各个RDD之间存在依赖关系，DStream直接也有依赖关系，RDD具有容错性，那么DStream也具有容错性。
在这里插入图片描述
上图相关说明：
1）、每一个椭圆形表示一个RDD
2）、椭圆形中的每个圆形代表一个RDD中的一个Partition分区
3）、每一列的多个RDD表示一个DStream(图中有三列所以有三个DStream)
4）、每一行最后一个RDD则表示每一个Batch Size所产生的中间结果RDD

Spark Streaming将流式计算分解成多个Spark Job，对于每一时间段数据的处理都会经过Spark DAG图分解以及Spark的任务集的调度过程。

3.2 DStream Operations

DStream类似RDD，里面包含很多函数，进行数据处理和输出操作，主要分为两大类：

  • DStream#Transformations：将一个DStream转换为另一个DStream
    http://spark.apache.org/docs/2.4.5/streaming-programming-guide.html#transformations-on-dstreams
  • DStream#Output Operations：将DStream中每批次RDD处理结果resultRDD输出
    http://spark.apache.org/docs/2.4.5/streaming-programming-guide.html#output-operations-on-dstreams

函数概述
DStream中包含很多函数，大多数与RDD中函数类似，主要分为两种类型：

其一：转换函数【Transformation函数】
在这里插入图片描述
DStream中还有一些特殊函数，针对特定类型应用使用的函数，比如updateStateByKey状态函数、window窗口函数等，后续具体结合案例讲解。

其二：输出函数【Output函数】
在这里插入图片描述

DStream中每批次结果RDD输出使用foreachRDD函数，前面使用的print函数底层也是调用foreachRDD函数，截图如下所示：
在这里插入图片描述
在DStream中有两个重要的函数，都是针对每批次数据RDD进行操作的，更加接近底层，性能更好，强烈推荐使用：

  • 转换函数transform：将一个DStream转换为另外一个DStream；
  • 输出函数foreachRDD：将一个DStream输出到外部存储系统；

在SparkStreaming企业实际开发中，建议：能对RDD操作的就不要对DStream操作，当调用DStream中某个函数在RDD中也存在，使用针对RDD操作。

转换函数：transform
通过源码认识transform函数，有两个方法重载，声明如下：
在这里插入图片描述
接下来使用transform函数，修改词频统计程序，具体代码如下：

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
*/
object StreamingTransformRDD {
def main(args: Array[String]): Unit = {
// 1. 构建StreamingContext流式上下文实例对象
val ssc: StreamingContext = {
// a. 创建SparkConf对象，设置应用配置信息
val sparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[3]")
// b.创建流式上下文对象, 传递SparkConf对象，TODO: 时间间隔 -> 用于划分流式数据为很多批次Batch
val context = new StreamingContext(sparkConf, Seconds(5))
// c. 返回
context
}
// 2. 从数据源端读取数据，此处是TCP Socket读取数据
/*
def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String]
*/
val inputDStream: ReceiverInputDStream[String] = ssc.socketTextStream(
"node1.itcast.cn", //
9999, //
// TODO: 设置Block存储级别为先内存，不足磁盘，副本为1
storageLevel = StorageLevel.MEMORY_AND_DISK
)
// TODO: 3. 对每批次的数据进行词频统计
/*
transform表示对DStream中每批次数据RDD进行操作
def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U]
*/
// TODO: 在DStream中，能对RDD操作的不要对DStream操作。
val resultDStream: DStream[(String, Int)] = inputDStream.transform(rdd => {
val resultRDD: RDD[(String, Int)] = rdd
// 过滤不合格的数据
.filter(line => null != line && line.trim.length > 0)
// 按照分隔符划分单词
.flatMap(line => line.trim.split("\\s+"))
// 转换数据为二元组，表示每个单词出现一次
.map(word => (word, 1))
// 按照单词分组，聚合统计
.reduceByKey((tmp, item) => tmp + item)
resultRDD
})
// 4. 将结果数据输出 -> 将每批次的数据处理以后输出
resultDStream.print(10)
// 5. 对于流式应用来说，需要启动应用
ssc.start()
// 流式应用启动以后，正常情况一直运行（接收数据、处理数据和输出数据），除非人为终止程序或者程序异常停止
ssc.awaitTermination()
// 关闭流式应用(参数一：是否关闭SparkContext，参数二：是否优雅的关闭）
ssc.stop(stopSparkContext = true, stopGracefully = true)
}
}

查看WEB UI监控中每批次Batch数据执行Job的DAG图，直接显示针对RDD进行操作。
在这里插入图片描述

输出函数：foreachRDD
foreachRDD函数属于将DStream中结果数据RDD输出的操作，类似transform函数，针对每批次RDD数据操作，源码声明如下：
在这里插入图片描述
继续修改词频统计代码，自定义输出数据，具体代码如下：

import java.util.Date
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
/**
* 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
*/
object StreamingOutputRDD {
def main(args: Array[String]): Unit = {
// 1. 构建StreamingContext流式上下文实例对象
val ssc: StreamingContext = {
// a. 创建SparkConf对象，设置应用配置信息
val sparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[3]")
// TODO：设置数据输出文件系统的算法版本为2
.set("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")
// b.创建流式上下文对象, 传递SparkConf对象，TODO: 时间间隔 -> 用于划分流式数据为很多批次Batch
val context = new StreamingContext(sparkConf, Seconds(5))
// c. 返回
context
}
// 2. 从数据源端读取数据，此处是TCP Socket读取数据
/*
def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String]
*/
val inputDStream: ReceiverInputDStream[String] = ssc.socketTextStream(
"node1.itcast.cn", //
9999, //
// TODO: 设置Block存储级别为先内存，不足磁盘，副本为1
storageLevel = StorageLevel.MEMORY_AND_DISK
)
// 3. 对每批次的数据进行词频统计
/*
transform表示对DStream中每批次数据RDD进行操作
def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U]
*/
// TODO: 在DStream中，能对RDD操作的不要对DStream操作。
val resultDStream: DStream[(String, Int)] = inputDStream.transform(rdd => {
val resultRDD: RDD[(String, Int)] = rdd
// 过滤不合格的数据
.filter(line => null != line && line.trim.length > 0)
// 按照分隔符划分单词
.flatMap(line => line.trim.split("\\s+"))
// 转换数据为二元组，表示每个单词出现一次
.map(word => (word, 1))
// 按照单词分组，聚合统计
.reduceByKey((tmp, item) => tmp + item)
resultRDD
})
// TODO: 4. 将结果数据输出 -> 将每批次的数据处理以后输出
/*
对DStream中每批次结果RDD数据进行输出操作
def foreachRDD(foreachFunc: (RDD[T], Time) => Unit): Unit
其中Time就是每批次BatchTime，Long类型数据, 转换格式：2020/05/10 16:53:25
*/
resultDStream.foreachRDD{ (rdd, time) =>
// 使用lang3包下FastDateFormat日期格式类，属于线程安全的
val batchTime: String = FastDateFormat.getInstance("yyyyMMddHHmmss")
.format(new Date(time.milliseconds))
println("-------------------------------------------")
println(s"Time: $batchTime")
println("-------------------------------------------")
// TODO: 先判断RDD是否有数据，有数据在输出
if(!rdd.isEmpty()){
// 对于结果RDD输出，需要考虑降低分区数目
val resultRDD = rdd.coalesce(1)
// 对分区数据操作
resultRDD.foreachPartition{iter =>iter.foreach(item => println(item))}
// 保存数据至HDFS文件
resultRDD.saveAsTextFile(s"datas/streaming/wc-output-${batchTime}")
}
}
// 5. 对于流式应用来说，需要启动应用
ssc.start()
// 流式应用启动以后，正常情况一直运行（接收数据、处理数据和输出数据），除非人为终止程序或者程序异常停止
ssc.awaitTermination()
// 关闭流式应用(参数一：是否关闭SparkContext，参数二：是否优雅的关闭）
ssc.stop(stopSparkContext = true, stopGracefully = true)
}
}

将SparkStreaming处理结果RDD数据保存到MySQL数据库或者HBase表中，代码该如何编写呢？
http://spark.apache.org/docs/2.4.5/streaming-programming-guide.html#design-patterns-for-using-foreachrdd
伪代码如下所示：

// 数据输出，将分析处理结果数据输出到MySQL表
resultDStream.foreachRDD{(rdd, time) =>
// 将BatchTime转换：2019/10/10 14:59:35
val batchTime = FastDateFormat.getInstance("yyyy/MM/dd HH:mm:ss").format(time.milliseconds)
println("-------------------------------------------")
println(s"Time: $batchTime")
println("-------------------------------------------")
// TODO：首先判断每批次结果RDD是否有值，有值才输出, 必须判断，提升性能
if(!rdd.isEmpty()){
rdd.foreachPartition{iter =>
// 第一步、获取连接：从数据库连接池中获取连接
val conn: Connection = null
// 第二步、保存分区数据到MySQL表
iter.foreach{item =>// TODO: 使用conn将数据保存到MySQL表中
}
// 第三步、关闭连接：将连接放入到连接池中
if(null != conn) conn.close()
}
}
}

将每批次数据统计结果RDD保存到HDFS文件中，代码如下：

resultDStream.foreachRDD{(rdd, time) =>
// 将BatchTime转换：2019/10/10 14:59:35
val batchTime = FastDateFormat.getInstance("yyyy/MM/dd HH:mm:ss").format(time.milliseconds)
println("-------------------------------------------")
println(s"Time: $batchTime")
println("-------------------------------------------")
// TODO：首先判断每批次结果RDD是否有值，有值才输出, 必须判断，提升性能
if(!rdd.isEmpty()){
// 注意：将Streaming结果数据RDD保存文件中时，最好考虑降低分区数目
rdd.coalesce(1).saveAsTextFile(s"datas/spark/streaming/wc-${time.milliseconds}")
}
}
http://www.yayakq.cn/news/274533/

相关文章：

  • 电商视觉设计网站手机版网站设计风格
  • 重庆智慧团建网站登录平台做网站含营销
  • 铁岭 建筑公司网站 中企动力建设2017两学一做竞赛网站
  • 网站文化制度建设山东省建设厅网站首页
  • 网站标题更改wordpress表格代码
  • 中文html5网站模板wordpress文章无法中文
  • flash网站代码下载微信公众号运营策划
  • 网站备案没有固定电话男女生做恶心的网站
  • 哈尔滨网络科技公司做网站cnnic可信网站必须做吗?
  • t恤定制网站哪个好项目管理软件免费版
  • 广告公司推广软文长沙seo优化排名
  • wordpress有客户端吗网站seo排名优化工具
  • 网站域名使用怎么做待摊分录WordPress未设置密码用户
  • 外贸网站搭建推广江门网站
  • 网站主要栏目wordpress新用户默认
  • 手机网站搭建营销型网站建设目标
  • 室内设计高端网站wordpress+外网无法访问
  • 铆钉机 东莞网站建设100款软件免费下载
  • 网站后台怎么制作代写文章平台
  • 研究院网站建设的内容河南省新闻奖评选通知
  • WordPress 整个网站导出浏览器下载免费版
  • cad二次开发网站女孩做网站工作辛苦吗
  • 网站设计与实现wordpress 主题 google
  • 深圳最好的网站建设单位网站及政务新媒体建设管理
  • 大型门户网站建设哪家好东莞多语言网站建设
  • 男女做暖暖到网站北京高端网站建
  • 凡科建站和wordpress网页图片怎么下载
  • 做摄影和后期的兼职网站建站之星快速建站价格
  • 可以做设计兼职的网站有哪些工作网站设计存在的不足
  • 如何做试玩类网站editplus网站开发