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。

相关文章

Python在二进制文件中进行数据搜索的实战指南

Python在二进制文件中进行数据搜索的实战指南

《Python在二进制文件中进行数据搜索的实战指南》在二进制文件中搜索特定数据是编程中常见的任务,尤其在日志分析、程序调试和二进制数据处理中尤为重要,下面我们就来看看如何使用Python实现这一功能吧... 目录简介1. 二进制文件搜索概述2. python二进制模式文件读取（rb）2.1 二进制模式与文本
阅读更多...
C#实现将XML数据自动化地写入Excel文件

C#实现将XML数据自动化地写入Excel文件

《C#实现将XML数据自动化地写入Excel文件》在现代企业级应用中,数据处理与报表生成是核心环节,本文将深入探讨如何利用C#和一款优秀的库,将XML数据自动化地写入Excel文件,有需要的小伙伴可以... 目录理解XML数据结构与Excel的对应关系引入高效工具：使用Spire.XLS for .NETC
阅读更多...
MySQL游标和触发器的操作流程

MySQL游标和触发器的操作流程

《MySQL游标和触发器的操作流程》本文介绍了MySQL中的游标和触发器的使用方法,游标可以对查询结果集进行逐行处理,而触发器则可以在数据表发生更改时自动执行预定义的操作,感兴趣的朋友跟随小编一起看看... 目录游标游标的操作流程1. 定义游标2.打开游标3.利用游标检索数据4.关闭游标例题触发器触发器的基
阅读更多...
在C#中分离饼图的某个区域的操作指南

在C#中分离饼图的某个区域的操作指南

《在C#中分离饼图的某个区域的操作指南》在处理Excel饼图时,我们可能需要将饼图的各个部分分离出来,以使它们更加醒目,Spire.XLS提供了Series.DataFormat.Percent属性,... 目录引言如何设置饼图各分片之间分离宽度的代码示例：从整个饼图中分离单个分片的代码示例：引言在处理
阅读更多...
Python列表的创建与删除的操作指南

Python列表的创建与删除的操作指南

《Python列表的创建与删除的操作指南》列表（list）是Python中最常用、最灵活的内置数据结构之一,它支持动态扩容、混合类型、嵌套结构,几乎无处不在,但你真的会创建和删除列表吗,本文给大家介绍... 目录一、前言二、列表的创建方式1. 字面量语法（最常用）2. 使用list()构造器3. 列表推导式
阅读更多...
MySQL数据目录迁移的完整过程

MySQL数据目录迁移的完整过程

《MySQL数据目录迁移的完整过程》文章详细介绍了将MySQL数据目录迁移到新硬盘的整个过程,包括新硬盘挂载、创建新的数据目录、迁移数据（推荐使用两遍rsync方案）、修改MySQL配置文件和重启验证... 目录1，新硬盘挂载（如果有的话）2，创建新的 mysql 数据目录3，迁移 MySQL 数据（推荐两
阅读更多...
Python数据验证神器Pydantic库的使用和实践中的避坑指南

Python数据验证神器Pydantic库的使用和实践中的避坑指南

《Python数据验证神器Pydantic库的使用和实践中的避坑指南》Pydantic是一个用于数据验证和设置的库,可以显著简化API接口开发,文章通过一个实际案例,展示了Pydantic如何在生产环... 目录1️⃣ 崩溃时刻：当你的API接口又双叒崩了！2️⃣ 神兵天降：3行代码解决验证难题3️⃣ 深度
阅读更多...
MySQL快速复制一张表的四种核心方法(包括表结构和数据)

MySQL快速复制一张表的四种核心方法(包括表结构和数据)

《MySQL快速复制一张表的四种核心方法(包括表结构和数据)》本文详细介绍了四种复制MySQL表（结构+数据）的方法,并对每种方法进行了对比分析,适用于不同场景和数据量的复制需求,特别是针对超大表（1... 目录一、mysql 复制表（结构+数据）的 4 种核心方法（面试结构化回答）方法 1：CREATE
阅读更多...
Go异常处理、泛型和文件操作实例代码

Go异常处理、泛型和文件操作实例代码

《Go异常处理、泛型和文件操作实例代码》Go语言的异常处理机制与传统的面向对象语言（如Java、C#）所使用的try-catch结构有所不同,它采用了自己独特的设计理念和方法,：本文主要介绍Go异... 目录一：异常处理常见的异常处理向上抛中断程序恢复程序二：泛型泛型函数泛型结构体泛型切片泛型 map三：文
阅读更多...
详解C++ 存储二进制数据容器的几种方法

详解C++ 存储二进制数据容器的几种方法

《详解C++存储二进制数据容器的几种方法》本文主要介绍了详解C++存储二进制数据容器,包括std::vector、std::array、std::string、std::bitset和std::ve... 目录1.std::vector<uint8_t>（最常用）特点：适用场景：示例：2.std::arra
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2