FunDA(11)- 数据库操作的并行运算:Parallel data processing

   FunDA最重要的设计目标之一就是能够实现数据库操作的并行运算。我们先重温一下fs2是如何实现并行运算的。我们用interleave、merge、either这几种方式来同时处理两个Stream里的元素。interleave保留了固定的交叉排列顺序,而merge和either则会产生不特定顺序,这个现象可以从下面的例子里看到:

implicit val strategy = Strategy.fromFixedDaemonPool(4)

implicit val scheduler = Scheduler.fromFixedDaemonPool(2)
 
//当前元素跟踪显示
def log[A](pre: String): Pipe[Task,A,A] = _.evalMap { row =>
  Task.delay {println(s"${pre}>${row}");row}
}                                                
def randomDelay[A](max: FiniteDuration): Pipe[Task,A,A] = _.evalMap { a => {
  val delay: Task[Int] = Task.delay {scala.util.Random.nextInt(max.toMillis.toInt)}
  delay.flatMap {d => Task.now(a).schedule(d.millis)}
 }
}  


val s1: Stream[Task,Int] = Stream(1,2,3,4,5).through(randomDelay(100.millis))
 
val s2 = Stream(11,22,33,44,55,66).through(randomDelay(30.millis))

val s3: Stream[Task,String] = Stream("a","b","c").through(randomDelay(200.millis))


(s1 interleave s2).through(log("")).run.unsafeRun //> >1
                                                  //| >11
                                                  //| >2
                                                  //| >22
                                                  //| >3
                                                  //| >33
                                                  //| >4
                                                  //| >44
                                                  //| >5
                                                  //| >55

(s1 merge s2).through(log("")).run.unsafeRun      //> >11
                                                  //| >1
                                                  //| >22
                                                  //| >2
                                                  //| >33
                                                  //| >44
                                                  //| >3
                                                  //| >55
                                                  //| >4
                                                  //| >5
                                                  //| >66
(s1 either s3).through(log("")).run.unsafeRun     //> >Left(1)
                                                  //| >Left(2)
                                                  //| >Right(a)
                                                  //| >Right(b)
                                                  //| >Left(3)
                                                  //| >Left(4)
                                                  //| >Left(5)
                                                  //| >Right(c)

从上面的例子我们可以看到merge产生的不规则顺序。fs2的nondeterministic算法可以保证两个队列元素处理顺序的合理分配最大化。如果我们需要对两个以上数据流进行并行处理的话,fs2提供了join(mergeN)函数:

def join[F[_],O](maxOpen: Int)(outer: Stream[F,Stream[F,O]])(implicit F: Async[F]): Stream[F,O] = {...}

从这个函数的款式我们看到它的入参数outer是个Stream[F,Stream[F,O]]类型,是个内外两层的流。现实场景如外层是多个数据库连接(connections),内层是多个客户端(clients)。在FunDA的功能描述里外层是多个数据源(sources),内层是多个读取函数(reader),又或者外层是多个数据行(元素),内层是数据处理函数。我们先看看如何实现多个数据源的并行产生:

val ss: Stream[Task,Stream[Task,Int]] = Stream(s1,s2,s1,s2)
                 //> ss  : fs2.Stream[fs2.Task,fs2.Stream[fs2.Task,Int]] = Segment(Emit(Chunk(Seg

从ss的类型款式来看,我们可以直接用Stream构建器来生成这个Stream[Task,Stream[Task,A]]类型。在前面我们已经掌握了用Slick来产生Stream[Task,FDAROW]的方法,例如:

  val albumStream1 = streamLoader.fda_typedStream(albumsInfo.result)(db)(10.minutes, 512, 128)()()

albumStream1是个Reactive-Stream数据源。这样我们可以在FunDA里增加一个并行Source构建函数:

  def fda_par_load(sources: FDAPipeLine[FDAROW]*)(maxOpen: Int) = {
     concurrent.join(maxOpen)(Stream(sources: _*))
  }

maxOpen代表最多可以同时运行的运算数,最好取小于机器内核数的一个数。用这个函数来并行构建数据源:

package com.bayakala.funda.fdapars.examples
import slick.driver.H2Driver.api._
import com.bayakala.funda.samples._
import com.bayakala.funda.fdarows.FDAROW
import com.bayakala.funda.fdasources.FDADataStream._
import scala.concurrent.duration._
import com.bayakala.funda.fdapipes._
import FDAValves._
import com.bayakala.funda.fdapars.FDAPars._
object Example1 extends App {
  val albums = SlickModels.albums
  val companies = SlickModels.companies

  //数据源query
  val albumsInfo = for {
    (a,c) <- albums join companies on (_.company === _.id)
  } yield (a.title,a.artist,a.year,c.name)

  //query结果强类型(用户提供)
  case class Album(title: String, artist: String, year: Int, publisher: String) extends FDAROW
  //强类型转换函数(用户提供)
  def toTypedRow(row: (String, String, Option[Int], String)): Album =
    Album(row._1, row._2, row._3.getOrElse(2000), row._4)

  val db = Database.forConfig("h2db")

  val streamLoader = FDAStreamLoader(slick.driver.H2Driver, toTypedRow _)
  val albumStream1 = streamLoader.fda_typedStream(albumsInfo.result)(db)(10.minutes, 512, 128)()()
  val albumStream2 = streamLoader.fda_typedStream(albumsInfo.result)(db)(10.minutes, 512, 128)()()
  val albumStream3 = streamLoader.fda_typedStream(albumsInfo.result)(db)(10.minutes, 512, 128)()()

  def printAlbums: FDATask[FDAROW] = row => {
    row match {
      case album: Album =>
        println("____________________")
        println(s"品名:${album.title}")
        println(s"演唱:${album.artist}")
        println(s"年份:${album.year}")
        println(s"发行:${album.publisher}")

        fda_skip
      //        fda_next(album)
      case r@_ => fda_next(r)
    }
  }

  fda_par_load(albumStream1,albumStream1,albumStream1)(3).appendTask(printAlbums).startRun

startRun后显示结果:

*** (c.z.hikari.HikariDataSource) HikariCP pool h2db is starting.
*** (s.jdbc.JdbcBackend.statement) Preparing statement: select x2."TITLE", x2."ARTIST", x2."YEAR", x3."NAME" from "ALBUMS" x2, "COMPANY" x3 where x2."COMPANY" = x3."ID"
*** (s.jdbc.JdbcBackend.statement) Preparing statement: select x2."TITLE", x2."ARTIST", x2."YEAR", x3."NAME" from "ALBUMS" x2, "COMPANY" x3 where x2."COMPANY" = x3."ID"
*** (s.jdbc.JdbcBackend.statement) Preparing statement: select x2."TITLE", x2."ARTIST", x2."YEAR", x3."NAME" from "ALBUMS" x2, "COMPANY" x3 where x2."COMPANY" = x3."ID"
____________________
品名:Keyboard Cat's Greatest Hits
演唱:Keyboard Cat
年份:2016
发行:Sony Music Inc
____________________
品名:Keyboard Cat's Greatest Hits
演唱:Keyboard Cat
年份:2016
发行:Sony Music Inc
____________________
品名:Keyboard Cat's Greatest Hits
演唱:Keyboard Cat
年份:2016
发行:Sony Music Inc
____________________
品名:Spice
演唱:Spice Girls
年份:2016
发行:Columbia Records
____________________
品名:Spice
演唱:Spice Girls
年份:2016
发行:Columbia Records
____________________
品名:Spice
演唱:Spice Girls
年份:2016
发行:Columbia Records
____________________
品名:Whenever You Need Somebody
演唱:Rick Astley
年份:2016
发行:Sony Music Inc
____________________
品名:Whenever You Need Somebody
演唱:Rick Astley
年份:2016
发行:Sony Music Inc
____________________
品名:Whenever You Need Somebody
演唱:Rick Astley
年份:2016
发行:Sony Music Inc
____________________
品名:The Triumph of Steel
演唱:Manowar
年份:2016
发行:The K-Pops Singers
____________________
品名:The Triumph of Steel
演唱:Manowar
年份:2016
发行:The K-Pops Singers
____________________
品名:The Triumph of Steel
演唱:Manowar
年份:2016
发行:The K-Pops Singers
____________________
品名:Believe
演唱:Justin Bieber
年份:2016
发行:Columbia Records
____________________
品名:Believe
演唱:Justin Bieber
年份:2016
发行:Columbia Records
____________________
品名:Believe
演唱:Justin Bieber
年份:2016
发行:Columbia Records

Process finished with exit code 0

FunDA的另一个并行运算需求是并行对一长串数据元素进行一个函数的施用。先看看这个函数的款式:

    //作业类型
    type FDATask[ROW] = ROW => Option[List[ROW]]

也就是我们前面使用过的,由用户提供的那个作业函数类型。但是再看看fda_runPar函数,只能对下面这种类型进行并行运算:

  def fda_runPar(parTask: FDAParTask)(maxOpen: Int) =
    concurrent.join(maxOpen)(parTask).through(fda_afterPar)

  //并行作业类型
  type FDAParTask = Stream[Task,Stream[Task,Option[List[FDAROW]]]]

我们首先必须把Stream[Task,A]转成Stream[Task,Stream[Task,A]]: 

    implicit class toFDAOps(fs2Stream: FDAPipeLine[FDAROW]) {
      def appendTask(t: FDATask[FDAROW]) = fs2Stream.through(fda_execUserTask(t))

      def startRun = fs2Stream.run.unsafeRun

      def startFuture = fs2Stream.run.unsafeRunAsyncFuture

      def toPar(st: FDATask[FDAROW]): Stream[Task, Stream[Task, Option[List[FDAROW]]]] =
        fs2Stream.map { row =>
          Stream.eval(Task {
            st(row)
          })
        }
    }

我们可以用toPar来实现并行运算类型转换。下面是一个调用例子:

  //并行作业函数
  def updateYear: FDATask[FDAROW] = row => {
    row match {
      case album: Album =>
        val action = albums.filter{r => r.title === album.title}.map(_.year).update(Some(2016))
 //把原数据和新构建的Action一起传下去
        fda_next(List(album,FDAActionRow(action)))
      case others@ _ => fda_next(others)
    }
  }

//并行读取
  val s1 = fda_par_load(albumStream1,albumStream1,albumStream1)(3)
//并行构建Action
  val s2 = fda_runPar(s1.toPar(updateYear))(3)

s1是并行构建的数据源,s2是对数据源产生的元素进行并行的函数updateYear施用。我们同样可以把产生的ActionRow用并行的方法来运算:

  val runner = FDAActionRunner(slick.driver.H2Driver)
  //并行运算函数
  def runActions: FDATask[FDAROW] = row => {
    row match {
      case FDAActionRow(action) =>
        runner.fda_execAction(action)(db)
        fda_skip
      case others@ _ => fda_next(others)
    }
  }

//并行运算Action
  val s3 = fda_runPar(s2.toPar(runActions))(3)
//开始运算
  s3.appendTask(printAlbums).startRun

从上面的例子里应该能够体会到函数式编程的灵活性:在startRun之前,我们可以任意进行函数组合,而且静态类型系统(static type system)会帮我们检查各组件的类型是否匹配。下面是具体运算结果显示:

*** (c.z.hikari.HikariDataSource) HikariCP pool h2db is starting.
*** (s.jdbc.JdbcBackend.statement) Preparing statement: select x2."TITLE", x2."ARTIST", x2."YEAR", x3."NAME" from "ALBUMS" x2, "COMPANY" x3 where x2."COMPANY" = x3."ID"
*** (s.jdbc.JdbcBackend.statement) Preparing statement: select x2."TITLE", x2."ARTIST", x2."YEAR", x3."NAME" from "ALBUMS" x2, "COMPANY" x3 where x2."COMPANY" = x3."ID"
*** (s.jdbc.JdbcBackend.statement) Preparing statement: select x2."TITLE", x2."ARTIST", x2."YEAR", x3."NAME" from "ALBUMS" x2, "COMPANY" x3 where x2."COMPANY" = x3."ID"
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Keyboard Cat''s Greatest Hits'
____________________
品名:Keyboard Cat's Greatest Hits
演唱:Keyboard Cat
年份:1999
发行:Sony Music Inc
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Keyboard Cat''s Greatest Hits'
____________________
品名:Keyboard Cat's Greatest Hits
演唱:Keyboard Cat
年份:1999
发行:Sony Music Inc
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Keyboard Cat''s Greatest Hits'
____________________
品名:Keyboard Cat's Greatest Hits
演唱:Keyboard Cat
年份:1999
发行:Sony Music Inc
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Spice'
____________________
品名:Spice
演唱:Spice Girls
年份:1999
发行:Columbia Records
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Spice'
____________________
品名:Spice
演唱:Spice Girls
年份:1999
发行:Columbia Records
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Spice'
____________________
品名:Spice
演唱:Spice Girls
年份:1999
发行:Columbia Records
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Whenever You Need Somebody'
____________________
品名:Whenever You Need Somebody
演唱:Rick Astley
年份:1999
发行:Sony Music Inc
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Whenever You Need Somebody'
____________________
品名:Whenever You Need Somebody
演唱:Rick Astley
年份:1999
发行:Sony Music Inc
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Whenever You Need Somebody'
____________________
品名:Whenever You Need Somebody
演唱:Rick Astley
年份:1999
发行:Sony Music Inc
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'The Triumph of Steel'
____________________
品名:The Triumph of Steel
演唱:Manowar
年份:1999
发行:The K-Pops Singers
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'The Triumph of Steel'
____________________
品名:The Triumph of Steel
演唱:Manowar
年份:1999
发行:The K-Pops Singers
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'The Triumph of Steel'
____________________
品名:The Triumph of Steel
演唱:Manowar
年份:1999
发行:The K-Pops Singers
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Believe'
____________________
品名:Believe
演唱:Justin Bieber
年份:1999
发行:Columbia Records
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Believe'
____________________
品名:Believe
演唱:Justin Bieber
年份:1999
发行:Columbia Records
*** (s.jdbc.JdbcBackend.statement) Preparing statement: update "ALBUMS" set "YEAR" = ? where "ALBUMS"."TITLE" = 'Believe'
____________________
品名:Believe
演唱:Justin Bieber
年份:1999
发行:Columbia Records

Process finished with exit code 0

注意:上面这个例子是存粹做出来作为函数调用示范的,不做任何逻辑和应用上的考虑。下面是本篇讨论的示范源代码:

package com.bayakala.funda.fdapars.examples
import slick.driver.H2Driver.api._
import com.bayakala.funda.samples._
import com.bayakala.funda.fdarows.FDARowTypes._
import com.bayakala.funda.fdarows.FDAROW
import com.bayakala.funda.fdasources.FDADataStream._

import scala.concurrent.duration._
import com.bayakala.funda.fdapipes._
import FDAValves._
import com.bayakala.funda.fdapars.FDAPars._
import com.bayakala.funda.fdarows.FDARowTypes.FDAActionRow
object Example1 extends App {
  val albums = SlickModels.albums
  val companies = SlickModels.companies

  //数据源query
  val albumsInfo = for {
    (a,c) <- albums join companies on (_.company === _.id)
  } yield (a.title,a.artist,a.year,c.name)

  //query结果强类型(用户提供)
  case class Album(title: String, artist: String, year: Int, publisher: String) extends FDAROW
  //转换函数(用户提供)
  def toTypedRow(row: (String, String, Option[Int], String)): Album =
    Album(row._1, row._2, row._3.getOrElse(2000), row._4)

  val db = Database.forConfig("h2db")

  val streamLoader = FDAStreamLoader(slick.driver.H2Driver, toTypedRow _)
  val albumStream1 = streamLoader.fda_typedStream(albumsInfo.result)(db)(10.minutes, 512, 128)()()
  val albumStream2 = streamLoader.fda_typedStream(albumsInfo.result)(db)(10.minutes, 512, 128)()()
  val albumStream3 = streamLoader.fda_typedStream(albumsInfo.result)(db)(10.minutes, 512, 128)()()

  def printAlbums: FDATask[FDAROW] = row => {
    row match {
      case album: Album =>
        println("____________________")
        println(s"品名:${album.title}")
        println(s"演唱:${album.artist}")
        println(s"年份:${album.year}")
        println(s"发行:${album.publisher}")

        fda_skip
      //        fda_next(album)
      case r@_ => fda_next(r)
    }
  }

 // fda_par_load(albumStream1,albumStream1,albumStream1)(3).appendTask(printAlbums).startRun

  //并行作业函数
  def updateYear: FDATask[FDAROW] = row => {
    row match {
      case album: Album =>
        val action = albums.filter{r => r.title === album.title}.map(_.year).update(Some(2016))
 //把原数据和新构建的Action一起传下去
        fda_next(List(album,FDAActionRow(action)))
      case others@ _ => fda_next(others)
    }
  }

  val runner = FDAActionRunner(slick.driver.H2Driver)
  //并行运算函数
  def runActions: FDATask[FDAROW] = row => {
    row match {
      case FDAActionRow(action) =>
        runner.fda_execAction(action)(db)
        fda_skip
      case others@ _ => fda_next(others)
    }
  }
//并行读取
  val s1 = fda_par_load(albumStream1,albumStream1,albumStream1)(3)
//并行构建Action
  val s2 = fda_runPar(s1.toPar(updateYear))(3)

//并行运算Action
  val s3 = fda_runPar(s2.toPar(runActions))(3)
//开始运算
  s3.appendTask(printAlbums).startRun

}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏菩提树下的杨过

发布一个轻量级的滑块控件

比系统自带的组件体积要小很多,而且支持进度条显示(在做播放器时,显示缓冲进度很有用哦),另外也支持三角形的音量调整显示 使用示例: package { imp...

3338
来自专栏Jerry的SAP技术分享

ABAP SICF服务和Java Servlet的比较

In my opinion ABAP ICF handler and Java Servlet play the same role in enhancemen...

350
来自专栏积累沉淀

多人聊天室

最近学完网络线程协议 ,因此写了一个用java编写的聊天室 话不多说 效果如图 ? 首先 创建服务器端 package com.yc.server...

2888
来自专栏landv

vb.net_一个半成品

1574
来自专栏拂晓风起

Stage3d AGAL GPU处理照片 旧照片效果 sepia || pixelbender

963
来自专栏码匠的流水账

聊聊spring cloud gateway的NettyConfiguration

本文主要研究下spring cloud gateway的NettyConfiguration

1291
来自专栏哲学驱动设计

P of EAA 总结

看了几遍了,就只在书上勾了下,也没什么总结。暂且先把关键内容摘抄如下: Domain Logic Patterns Transaction Script Org...

1865
来自专栏Netkiller

Java 处理 Json 实例

本文介绍Java中如果处理Json。例如json编码与解码。以上节选自《Netkiller Java 手札》 第 5 章 Json 目录 5.1. Json 编...

3267
来自专栏跟着阿笨一起玩NET

LINQ 从 CSV 文件生成 XML

本文参考:http://msdn.microsoft.com/zh-cn/library/bb387090.aspx

721
来自专栏码匠的流水账

聊聊spring cloud的EurekaServerInitializerConfiguration

本文主要研究一下spring cloud的EurekaServerInitializerConfiguration

1032

扫码关注云+社区