- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在构建一个 Spark Structured Streaming 应用程序,我正在其中执行批处理流连接。批处理数据的源会定期更新。
因此,我计划定期对该批处理数据进行持久化/非持久化。
下面是我用来持久化和取消持久化批处理数据的示例代码。
流动:
var batchDF = handler.readBatchDF(sparkSession)
batchDF.persist(StorageLevel.MEMORY_AND_DISK)
var refreshedTime: Instant = Instant.now()
if (Duration.between(refreshedTime, Instant.now()).getSeconds > refreshTime) {
refreshedTime = Instant.now()
batchDF.unpersist(false)
batchDF = handler.readBatchDF(sparkSession)
.persist(StorageLevel.MEMORY_AND_DISK)
}
有没有更好的方法可以在 Spark 结构化流媒体作业中实现这种情况?
最佳答案
您可以通过使用 Structured Streaming 提供的流调度功能来实现这一点。
您可以通过创建一个定期刷新静态数据帧的人工“速率”流来触发静态数据帧的刷新(非持久化 -> 加载 -> 持久化)。这个想法是:
var
foreachBatch
调用在步骤 2 中创建的 refresher 方法的接收器。 // 1. Load the staticDataframe initially and keep as `var`
var staticDf = spark.read.format("delta").load(deltaPath)
staticDf.persist()
// 2. Define a method that refreshes the static Dataframe
def foreachBatchMethod[T](batchDf: Dataset[T], batchId: Long) = {
staticDf.unpersist()
staticDf = spark.read.format("delta").load(deltaPath)
staticDf.persist()
println(s"${Calendar.getInstance().getTime}: Refreshing static Dataframe from DeltaLake")
}
// 3. Use a "Rate" Stream that gets triggered at the required interval (e.g. 1 hour)
val staticRefreshStream = spark.readStream
.format("rate")
.option("rowsPerSecond", 1)
.option("numPartitions", 1)
.load()
.selectExpr("CAST(value as LONG) as trigger")
.as[Long]
// 4. Read actual streaming data and perform join operation with static Dataframe
// As an example I used Kafka as a streaming source
val streamingDf = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "test")
.option("startingOffsets", "earliest")
.option("failOnDataLoss", "false")
.load()
.selectExpr("CAST(value AS STRING) as id", "offset as streamingField")
val joinDf = streamingDf.join(staticDf, "id")
val query = joinDf.writeStream
.format("console")
.option("truncate", false)
.option("checkpointLocation", "/path/to/sparkCheckpoint")
.start()
// 5. Within that Rate Stream have a `foreachBatch` sink that calls refresher method
staticRefreshStream.writeStream
.outputMode("append")
.foreachBatch(foreachBatchMethod[Long] _)
.queryName("RefreshStream")
.trigger(Trigger.ProcessingTime("5 seconds")) // or e.g. 1 hour
.start()
举一个完整的例子,增量表被创建并更新为新值,如下所示:
val deltaPath = "file:///tmp/delta/table"
import spark.implicits._
val df = Seq(
(1L, "static1"),
(2L, "static2")
).toDF("id", "deltaField")
df.write
.mode(SaveMode.Overwrite)
.format("delta")
.save(deltaPath)
关于scala - 流静态加入 : How to refresh (unpersist/persist) static Dataframe periodically,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/66154867/
在 Spring-MVC 应用程序中,我尝试使用 validator 。我将注释 @NotEmpty 和 @Email 放在我的实体之一上,当我尝试验证它时,出现此错误: java.lang.NoSu
网络是我硕士学位的最后一门类(class)。我确实有一个关于如何计算非持久、持久和持久流水线的 http 往返时间的问题。 在花了无数小时阅读有关该问题、从其他大学下载笔记甚至搜索 youtube 视
让我们考虑典型的 订购 和 订单商品 例子。假设 订单商品 是 的一部分订购 聚合,只能通过订单添加。所以,添加一个新的 订单商品 到 订购,我们必须通过 Repository 加载整个 Aggreg
// lookup existing user & set a currently null child entity (ContactInfo) user.setContactInfo(contac
我正在尝试关注 this tutorial .我想我不是从使用可下载项目开始,而是从我之前做过的一个简单的“spring MVC - Maven - eclipse”项目开始。这个项目运行良好。 因此
我正在使用打开 Kubernetes 选项的 docker 应用程序运行 mac OSX Catalina。我使用以下 yaml 和命令创建了一个 PersistentVolume。 apiVersi
假设我有一个类 Employee和一个类Company其中包含 Employee 的 LinkedList对象,我想编写一个添加 Employee 的方法到特定的数据库Company 。我创建了一个新
我实际上正在将我们应用程序的所有组件更新到最新版本。因此,除其他外,我将从 eclipselink-2.5.0 升级到 eclipselink-2.7.3,并从 Tomcat 7 升级到 TomEE。
我试图遵循《用GlassFish 3开始Java EE 6平台》一书第2章中的示例。我正在Windows中使用cmd中的EclipseLink,Derby和Maven。我真的不知道这一点,将不胜感激!
我只是看看ClassGuard (虽然我知道一些 objection )。 但是我得到了 javax.persistence.PersistenceException: [PersistenceUni
我的印象是,如果我们使用持久字段,就不需要 getter 方法,因为实体管理器直接引用实例变量。但是,当我从实体中删除 getter 和 setter 方法以具有持久字段时,未从数据库中检索到相应实例
我正在 Eclipse 中使用 Servlet、JPA、EJB 和 JBoss 进行项目。正如您在我的主题标题中看到的,我的 persistence.xml 文件有错误,但我不知道是哪个:
我已经尝试了一个星期或更长时间来让我的状态在 react native Android 应用程序中持续存在,但在重新水化后状态始终具有初始值。如果我使用 Redux devtools 检查 ASync
什么时候应该坚持,什么时候应该补充水分?命名约定非常困惑,因为作者几乎没有提到它们在 redux-persist 的上下文中的含义。 . 最佳答案 在不真正了解图书馆的情况下: persist = 将
我正在尝试按照老师文档中提供的信息设置一个简单的 jpa 2.0 项目。我已经处理这个问题好几个小时了,但无论我做什么,当我尝试创建 EntityManagerFactory 时,我总是遇到这个异常:
我有一个 Maven 项目,我将其转换为现在可与 Maven 一起使用的 JPA 项目。我的persistence.xml如下: My Persistence Unit
我正在使用 Netbeans 6.8 并构建简单的 Maven Web 应用程序项目。 为持久实体创建实体和主文件 [也创建持久单元] 并使用 EclipsLink。 但是当我运行主文件时出现这个错误
我是 Kubernetes 的新手,我很难理解 Kubernetes 中持久存储背后的整个想法。 这就足够了吗,或者我必须创建持久卷,如果我只部署这两个对象而不创建 PV 会发生什么情况? 存储应该在
我正在尝试使用 JPA 为我目前参与的 Java-EE 项目设置持久性,并且我遇到了许多配置问题。目前,我已经在 persistence.xml 中定义了一个 RESOURCE_LOCAL 持久性单元
Akka 持久性查询通过提供一个通用的基于异步流的查询接口(interface)来补充 Persistence,各种日志插件可以实现该接口(interface)以公开它们的查询功能。 这是来自 akk
我是一名优秀的程序员,十分优秀!