FunDA（2）－ Streaming Data Operation：流式数据操作

2024-04-09 04:58
文章标签 数据 操作 data operation streaming 流式 funda

本文主要是介绍FunDA（2）－ Streaming Data Operation：流式数据操作，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

   在上一集的讨论里我们介绍并实现了强类型返回结果行。使用强类型主要的目的是当我们把后端数据库SQL批次操作搬到内存里转变成数据流式按行操作时能更方便、准确、高效地选定数据字段。在上集讨论示范里我们用集合的foreach方式模拟了一个最简单的数据流，并把从数据库里批次读取的数据集转换成一串连续的数据行来逐行使用。一般来说完整的流式数据处理流程包括了从数据库中读取数据、根据读取的每行数据状态再对后台数据库进行更新，包括：插入新数据、更新、删除等。那么在上篇中实现的流式操作基础上再添加一种指令行类型就可以完善整个数据处理流程了，就像下面这个图示：

Database => Query -> Collection => Streaming -> DataRow => QueryAction(DataRow) -> ActionRow => execAction(ActionRow) -> Database

如果我们还是以Slick为目标FRM，那么这个ActionRow的类型就是Slick的DBIO[T]了：

package com.bayakala.funda.rowtypes
import slick.dbio._
object ActionType {type FDAAction[T] = DBIO[T]
}


记得有一次在一个Scala讨论区里遇到这样一个问题：如何把a表里的status字段更新成b表的status字段值，转化成SQL语句如下： 

update a,b set a.status=b.status where a.id=b.id


那位哥们的问题是如何用Slick来实现对a表的更新，不能用sql"???" interpolation 直接调用SQL语句，可能因为要求compile time语法check保障吧。这个问题用Slick Query还真的不太容易解决（能不能解决就不想费功夫去想了），这是因为FRM的SQL批次处理弱点。如果用FunDA的流式操作思路就会很容易解决了，只要用join Query把b.status读出来再用b.id=a.id逐个更新a.status。刚好，下面我们就示范通过ActionRow来解决这个问题。先用下面这段代码来设置测试数据：

import slick.dbio.DBIO
import slick.driver.H2Driver.api._import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.util.{Failure, Success}
import scala.concurrent.ExecutionContext.Implicits.global
import slick.jdbc.meta.MTable
object ActionRowTest extends App {class ATable(tag: Tag) extends Table[(Int,String,Int)](tag,"TA")  {def id = column[Int]("id",O.PrimaryKey)def flds = column[String]("aflds")def status = column[Int]("status")def * = (id,flds,status)}val tableA = TableQuery[ATable]class BTable(tag: Tag) extends Table[(Int,String,Int)](tag,"TB")  {def id = column[Int]("id",O.PrimaryKey)def flds = column[String]("bflds")def status = column[Int]("status")def * = (id,flds,status)}val tableB = TableQuery[BTable]val insertAAction =tableA ++= Seq ((1,"aaa",0),(2,"bbb",3),(3,"ccc",1),(4,"ddd",0),(16,"kkk",16))val insertBAction =tableB ++= Seq ((1,"aaa",1),(2,"bbb",2),(3,"ccc",3),(4,"ddd",4),(5,"kkk",5))val db = Database.forConfig("h2db")def tableExists(tables: Vector[MTable], tblname: String) =tables.exists {t =>t.name.toString.contains(tblname)}def createSchemaIfNotExists(): Future[Unit] = {db.run(MTable.getTables).flatMap {case tables if !tableExists(tables,".TA") && !tableExists(tables,".TB") =>println("Creating schemas for TA and TB...")db.run((tableA.schema ++ tableB.schema).create)case tables if !tableExists(tables,".TA") =>println("Creating schema for TA ...")db.run(tableA.schema.create)case tables if !tableExists(tables,".TB") =>println("Creating schema for TB ...")db.run(tableB.schema.create)case _ =>println("Schema for TA, TB already created.")Future.successful()}}def insertInitialData(): Future[Unit] = {val cleanInsert = DBIO.seq(tableA.delete, tableB.delete,insertAAction,insertBAction)db.run(cleanInsert).andThen {case Success(_) => println("Data insert completed.")case Failure(e) => println(s"Data insert failed [${e.getMessage}]")}}Await.ready(db.run(sql"DROP TABLE TA; DROP TABLE TB".as[String]),Duration.Inf)val initResult = createSchemaIfNotExists().flatMap {_ => insertInitialData()}Await.ready(initResult,Duration.Inf)}

用join query先把这两个表相关的字段值搬到内存转成强类型行FDADataRow：

 val selectAB = for {a <- tableAb <- tableBif (a.id === b.id)} yield (a.id,b.id,a.status,b.status)case class ABRow (id: Int, asts: Int, bsts: Int)def toABRow(raw: (Int,Int,Int,Int)) = ABRow(raw._1,raw._3,raw._4)import com.bayakala.funda.rowtypes.DataRowTypeval loader = FDADataRow(slick.driver.H2Driver, toABRow _)loader.getTypedRows(selectAB.result)(db).foreach {dataRow =>println(s"ID:${dataRow.id} Status A = ${dataRow.asts}, B = ${dataRow.bsts}")}


初始结果如下： 

ID:1 Status A = 0, B = 1
ID:2 Status A = 3, B = 2
ID:3 Status A = 1, B = 3
ID:4 Status A = 0, B = 4


现在我们把每条数据行DataRow转成动作行ActionRow。然后把每条DataRow的asts字段值替换成bsts的字段值： 

import com.bayakala.funda.rowtypes.ActionType.FDAActiondef updateAStatus(row: ABRow): FDAAction[Int] = {tableA.filter{r => r.id === row.id}.map(_.status).update(row.asts)}loader.getTypedRows(selectAB.result)(db).map(updateAStatus(_)).foreach {actionRow =>println(s"${actionRow.toString}")}


显示结果如下：

slick.driver.JdbcActionComponent$UpdateActionExtensionMethodsImpl$$anon$7@492691d7
slick.driver.JdbcActionComponent$UpdateActionExtensionMethodsImpl$$anon$7@27216cd
slick.driver.JdbcActionComponent$UpdateActionExtensionMethodsImpl$$anon$7@558bdf1f
slick.driver.JdbcActionComponent$UpdateActionExtensionMethodsImpl$$anon$7@8576fa0

现在每条DataRow已经被转化成jdbc action类型了。

下一步我们只需要运行这些ActionRow就可以完成任务了： 

  def execAction(act: FDAAction[Int]) = db.run(act)loader.getTypedRows(selectAB.result)(db).map(updateAStatus(_)).map(execAction(_))


现在再看看数据库中的TA表状态： 

  loader.getTypedRows(selectAB.result)(db).foreach {dataRow =>println(s"ID:${dataRow.id} Status A = ${dataRow.asts}, B = ${dataRow.bsts}")}结果：
ID:1 Status A = 1, B = 1
ID:2 Status A = 2, B = 2
ID:3 Status A = 3, B = 3
ID:4 Status A = 4, B = 4


我们看到已经正确更新了TA的status字段值。

在这个示范中明显有很多不足之处：如果a.status=b.status应该省略更新步骤。这是因为foreach只能模拟最基本的数据流动。如果我们使用了具备强大功能的Stream工具库如scalaz-stream-fs2，就可以更好控制数据元素的流动。更重要的是scalaz-stream-fs2支持并行运算，那么上面所描述的流程：

Database => Query -> Collection => Streaming -> DataRow => QueryAction(DataRow) -> ActionRow => execAction(ActionRow) -> Database
几个 => 环节：Query、Streaming、QueryAction、execAction将可以并行运算，从而实现充分利用多核CPU硬件资源，提高运算效率的目的。

下面是这次讨论涉及的源代码：
 
package com.bayakala.funda.rowtypesimport scala.concurrent.duration._
import scala.concurrent.Await
import slick.driver.JdbcProfileobject DataRowType {class FDADataRow[SOURCE, TARGET](slickProfile: JdbcProfile,convert: SOURCE  => TARGET){import slickProfile.api._def getTypedRows(slickAction: DBIO[Iterable[SOURCE]])(slickDB: Database): Iterable[TARGET] =Await.result(slickDB.run(slickAction), Duration.Inf).map(raw => convert(raw))}object FDADataRow {def apply[SOURCE, TARGET](slickProfile: JdbcProfile, converter: SOURCE => TARGET): FDADataRow[SOURCE, TARGET] =new FDADataRow[SOURCE, TARGET](slickProfile, converter)}}

package com.bayakala.funda.rowtypes
import slick.dbio._
object ActionType {type FDAAction[T] = DBIO[T]
}


import slick.dbio.DBIO
import slick.driver.H2Driver.api._import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.util.{Failure, Success}
import scala.concurrent.ExecutionContext.Implicits.global
import slick.jdbc.meta.MTable
object ActionRowTest extends App {class ATable(tag: Tag) extends Table[(Int,String,Int)](tag,"TA")  {def id = column[Int]("id",O.PrimaryKey)def flds = column[String]("aflds")def status = column[Int]("status")def * = (id,flds,status)}val tableA = TableQuery[ATable]class BTable(tag: Tag) extends Table[(Int,String,Int)](tag,"TB")  {def id = column[Int]("id",O.PrimaryKey)def flds = column[String]("bflds")def status = column[Int]("status")def * = (id,flds,status)}val tableB = TableQuery[BTable]val insertAAction =tableA ++= Seq ((1,"aaa",0),(2,"bbb",3),(3,"ccc",1),(4,"ddd",0),(16,"kkk",16))val insertBAction =tableB ++= Seq ((1,"aaa",1),(2,"bbb",2),(3,"ccc",3),(4,"ddd",4),(5,"kkk",5))val db = Database.forConfig("h2db")def tableExists(tables: Vector[MTable], tblname: String) =tables.exists {t =>t.name.toString.contains(tblname)}def createSchemaIfNotExists(): Future[Unit] = {db.run(MTable.getTables).flatMap {case tables if !tableExists(tables,".TA") && !tableExists(tables,".TB") =>println("Creating schemas for TA and TB...")db.run((tableA.schema ++ tableB.schema).create)case tables if !tableExists(tables,".TA") =>println("Creating schema for TA ...")db.run(tableA.schema.create)case tables if !tableExists(tables,".TB") =>println("Creating schema for TB ...")db.run(tableB.schema.create)case _ =>println("Schema for TA, TB already created.")Future.successful()}}def insertInitialData(): Future[Unit] = {val cleanInsert = DBIO.seq(tableA.delete, tableB.delete,insertAAction,insertBAction)db.run(cleanInsert).andThen {case Success(_) => println("Data insert completed.")case Failure(e) => println(s"Data insert failed [${e.getMessage}]")}}Await.ready(db.run(sql"DROP TABLE TA; DROP TABLE TB".as[String]),Duration.Inf)val initResult = createSchemaIfNotExists().flatMap {_ => insertInitialData()}Await.ready(initResult,Duration.Inf)val selectAB = for {a <- tableAb <- tableBif (a.id === b.id)} yield (a.id,b.id,a.status,b.status)case class ABRow (id: Int, asts: Int, bsts: Int)def toABRow(raw: (Int,Int,Int,Int)) = ABRow(raw._1,raw._3,raw._4)import com.bayakala.funda.rowtypes.DataRowType.FDADataRowval loader = FDADataRow(slick.driver.H2Driver, toABRow _)loader.getTypedRows(selectAB.result)(db).foreach {dataRow =>println(s"ID:${dataRow.id} Status A = ${dataRow.asts}, B = ${dataRow.bsts}")}import com.bayakala.funda.rowtypes.ActionType.FDAActiondef updateAStatus(row: ABRow): FDAAction[Int] = {tableA.filter{r => r.id === row.id}.map(_.status).update(row.bsts)}loader.getTypedRows(selectAB.result)(db).map(updateAStatus(_)).foreach {actionRow =>println(s"${actionRow.toString}")}def execAction(act: FDAAction[Int]) = db.run(act)loader.getTypedRows(selectAB.result)(db).map(updateAStatus(_)).map(execAction(_))loader.getTypedRows(selectAB.result)(db).foreach {dataRow =>println(s"ID:${dataRow.id} Status A = ${dataRow.asts}, B = ${dataRow.bsts}")}}



 


这篇关于FunDA（2）－ Streaming Data Operation：流式数据操作的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/887231。

相关文章

使用Folium在Python中进行地图可视化的操作指南

使用Folium在Python中进行地图可视化的操作指南

《使用Folium在Python中进行地图可视化的操作指南》在数据分析和可视化领域,地图可视化是一项非常重要的技能,它能够帮助我们更直观地理解和展示地理空间数据,Folium是一个基于Python的地... 目录引言一、Folium简介与安装1. Folium简介2. 安装Folium二、基础使用1. 创建
阅读更多...
使用EasyExcel实现简单的Excel表格解析操作

使用EasyExcel实现简单的Excel表格解析操作

《使用EasyExcel实现简单的Excel表格解析操作》：本文主要介绍如何使用EasyExcel完成简单的表格解析操作,同时实现了大量数据情况下数据的分次批量入库,并记录每条数据入库的状态,感兴... 目录前言固定模板及表数据格式的解析实现Excel模板内容对应的实体类实现AnalysisEventLis
阅读更多...
如何使用C#串口通讯实现数据的发送和接收

如何使用C#串口通讯实现数据的发送和接收

《如何使用C#串口通讯实现数据的发送和接收》本文详细介绍了如何使用C#实现基于串口通讯的数据发送和接收,通过SerialPort类,我们可以轻松实现串口通讯,并结合事件机制实现数据的传递和处理,感兴趣... 目录1. 概述2. 关键技术点2.1 SerialPort类2.2 异步接收数据2.3 数据解析2.
阅读更多...
大数据spark3.5安装部署之local模式详解

大数据spark3.5安装部署之local模式详解

《大数据spark3.5安装部署之local模式详解》本文介绍了如何在本地模式下安装和配置Spark,并展示了如何使用SparkShell进行基本的数据处理操作,同时,还介绍了如何通过Spark-su... 目录下载上传解压配置jdk解压配置环境变量启动查看交互操作命令行提交应用spark，一个数据处理框架
阅读更多...
通过ibd文件恢复MySql数据的操作方法

通过ibd文件恢复MySql数据的操作方法

《通过ibd文件恢复MySql数据的操作方法》文章介绍通过.ibd文件恢复MySQL数据的过程,包括知道表结构和不知道表结构两种情况,对于知道表结构的情况,可以直接将.ibd文件复制到新的数据库目录并... 目录第一种情况：知道表结构第二种情况：不知道表结构总结今天干了一件大事，安装1Panel导致原来服务
阅读更多...
Jmeter如何向数据库批量插入数据

Jmeter如何向数据库批量插入数据

《Jmeter如何向数据库批量插入数据》：本文主要介绍Jmeter如何向数据库批量插入数据方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Jmeter向数据库批量插入数据Jmeter向mysql数据库中插入数据的入门操作接下来做一下各个元件的配置总结Jmete
阅读更多...
SpringBoot操作MaxComputer方式(保姆级教程)

SpringBoot操作MaxComputer方式(保姆级教程)

《SpringBoot操作MaxComputer方式(保姆级教程)》：本文主要介绍SpringBoot操作MaxComputer方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的... 目录引言uqNqjoe一、引入依赖二、配置文件 application.properties（信息用自己
阅读更多...
HTML5 data-*自定义数据属性的示例代码

HTML5 data-*自定义数据属性的示例代码

《HTML5data-*自定义数据属性的示例代码》HTML5的自定义数据属性（data-*）提供了一种标准化的方法在HTML元素上存储额外信息,可以通过JavaScript访问、修改和在CSS中使用... 目录引言基本概念使用自定义数据属性1. 在 html 中定义2. 通过 JavaScript 访问3.
阅读更多...
C#中的 Dictionary常用操作

C#中的 Dictionary常用操作

《C#中的Dictionary常用操作》C#中的DictionaryTKey,TValue是用于存储键值对集合的泛型类,允许通过键快速检索值,并且具有唯一键、动态大小和无序集合的特性,常用操作包括添... 目录基本概念Dictionary的基本结构Dictionary的主要特性Dictionary的常用操作
阅读更多...
Spring AI集成DeepSeek实现流式输出的操作方法

Spring AI集成DeepSeek实现流式输出的操作方法

《SpringAI集成DeepSeek实现流式输出的操作方法》本文介绍了如何在SpringBoot中使用Sse（Server-SentEvents）技术实现流式输出,后端使用SpringMVC中的S... 目录一、后端代码二、前端代码三、运行项目小天有话说题外话参考资料前面一篇文章我们实现了《Spring
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2