FunDA(4)- 数据流内容控制:Stream data element control

    上节我们探讨了通过scalaz-stream-fs2来驱动一套数据处理流程,用fs2的Pipe类型来实现对数据流的逐行操作。本篇讨论准备在上节讨论的基础上对数据流的流动和元素操作进行优化完善。如数据流动中增加诸如next、skip、eof功能、内容控制中增加对行元素的append、insert、update、remove等操作方法。但是经过一番对fs2的再次解读,发现这些操作模式并不像我所想象那样的方式,实际上用fs2来实现数据行控制可能会更加简单和直接。这是因为与传统数据库行浏览方式不同的是fs2是一种拖式流(pull-model stream),它的数据行集合是一种泛函不可变集合。每一行一旦读取就等于直接消耗了断(consumed),所以只支持一种向前逐行读取模式。如果形象地描述的话,我们习惯的所谓数据集浏览可能是下面这样的场景:

读取一行数据 >>> (使用或更新行字段值)>>> 向下游发送新的一行数据。只有停止发送动作才代表终止运算。完成对上游的所有行数据读取并不代表终止操作,因为我们还可以不断向下游发送自定义产生的数据行。

我们用fs2模拟一套数据流管道FDAPipeLine,管道中间有不定数量的作业节点FDAWorkNode。作业方式包括从管道上游截取一个数据元素、对其进行处理、然后选择是否向下游的管道接口(FDAPipeJoint)发送。下面是这套模拟的类型:fdapipes/package.scala

 1 package com.bayakala.funda {
 2 
 3   import fs2._
 4 
 5   package object fdapipes {
 6     //数据行类型
 7     trait FDAROW
 8 
 9     //数据处理管道
10     type FDAPipeLine[ROW] = Stream[Task, ROW]
11     //数据作业节点
12     type FDAWorkNode[ROW] = Pipe[Task, ROW, ROW]
13     //数据管道开关阀门,从此处获得管道内数据
14     type FDAValve[ROW] = Handle[Task, ROW]
15     //管道连接器
16     type FDAPipeJoint[ROW] = Pull[Task, ROW, Unit]
17 
18     //作业类型
19     type FDATask[ROW] = ROW => Option[List[ROW]]
20 
21   }
22 
23 }

注意这个FDAROW类型:这是一种泛类型,因为在管道中流动的数据可能有多重类型,如数据行和QueryAction行。

流动控制方法:FDAValves.scala

 1 package com.bayakala.funda.fdapipes
 2 import fs2._
 3 object FDAValves {  //流动控制方法
 4 //跳过本行(不向下游发送)
 5   def fda_skip[ROW] = Some(List[ROW]())
 6 //将本行发送至下游连接管道
 7   def fda_next[ROW](r: ROW) = Some(List[ROW](r))
 8 //终止流动
 9   def fda_break = None
10 
11 }

数据发送方法:FDAPipes.scala

1 package com.bayakala.funda.fdapipes
2 import fs2._
3 object FDAJoints {  //数据发送方法
4 //write rows down the pipeline
5   def fda_pushRow[ROW](row: ROW) = Pull.output1(row)
6   def fda_pushRows[ROW](rows: List[ROW]) = Pull.output(Chunk.seq(rows))
7 }

作业节点工作方法:

 1 package com.bayakala.funda.fdapipes
 2 import FDAJoints._
 3 object FDANodes { //作业节点工作方法
 4  def fda_execUserTask[ROW](task: FDATask[ROW]): FDAWorkNode[ROW] = {
 5    def go: FDAValve[ROW] => FDAPipeJoint[ROW] = h => {
 6      h.receive1Option {
 7        case Some((r, h)) => task(r) match {
 8          case Some(xr) => xr match {
 9            case Nil => go(h)
10            case _ => fda_pushRows(xr) >> go(h)
11          }
12          case None => fda_halt
13        }
14        case None => fda_halt
15      }
16    }
17    in => in.pull(go)
18  }
19 
20 }

下面我们就示范这个工具库的具体使用方法:examples/Example1.scala 设置示范环境:

 1 package com.bayakala.funda.fdapipes.examples
 2 import fs2._
 3 import com.bayakala.funda.fdapipes._
 4 import FDANodes._
 5 import FDAValves._
 6 import Helpers._
 7 object Example1 extends App {
 8 
 9 
10   case class Employee(id: Int, name: String, age: Int, salary: BigDecimal) extends FDAROW
11 // test data set
12   val r1 = Employee(1, "John", 23, 100.00)
13   val r2 = Employee(2, "Peter", 25,100.00)
14   val r3 = Employee(3, "Kay", 35,100.00)
15   val r4 = Employee(4, "Cain", 45,100.00)
16   val r5 = Employee(5, "Catty", 35,100.00)
17   val r6 = Employee(6, "Little", 19,80.00)

注意Employee是一种行类型,因为它extends FDAROW。

我们再写一个跟踪显示当前流动数据行的函数:examples/Helpers.scala

1 package com.bayakala.funda.fdapipes.examples
2 import com.bayakala.funda.fdapipes._
3 import fs2.Task
4 object Helpers {
5   def log[ROW](prompt: String): FDAWorkNode[ROW] =
6     _.evalMap {row => Task.delay{ println(s"$prompt> $row"); row }}
7 }

下面我们就用几个有不同要求的例子来示范流动控制和数据处理功能,这些例子就是给最终用户的标准编程示范版本,然后由用户照版编写:

1、根据每条数据状态逐行进行处理:

 1 // 20 - 30岁加10%, 30岁> 加20%,其它加 5%
 2   def raisePay: FDATask[FDAROW] = row => {
 3     row match {
 4       case emp: Employee => {
 5         val cur = emp.age match {
 6           case a if ((a >= 20) && (a < 30)) => emp.copy(salary = emp.salary * 1.10)
 7           case a if ((a >= 30)) => emp.copy(salary = emp.salary * 1.20)
 8           case _ => emp.copy(salary = emp.salary * 1.05)
 9         }
10         fda_next(cur)
11       }
12       case _ => fda_skip
13     }
14   }

用户提供的功能函数类型必须是FDATask[FDAROW]。类型参数FDAROW代表数据行通用类型。如果用户指定了FDATask[Employee]函数类型,那么必须保证管道中流动的数据行只有Employee一种类型。完成对当前行数据的处理后用fda_next(emp)把它发送到下一节连接管道。我们用下面的组合函数来进行运算:

  Stream(r1,r2,r3,r4,r5,r6)
    .through(log("加薪前>"))
      .through(fda_execUserTask[FDAROW](raisePay))
      .through(log("加薪后>"))
    .run.unsafeRun
-----
运算结果:
加薪前>> Employee(1,John,23,100.0)
加薪后>> Employee(1,John,23,110.00)
加薪前>> Employee(2,Peter,25,100.0)
加薪后>> Employee(2,Peter,25,110.00)
加薪前>> Employee(3,Kay,35,100.0)
加薪后>> Employee(3,Kay,35,120.00)
加薪前>> Employee(4,Cain,45,100.0)
加薪后>> Employee(4,Cain,45,120.00)
加薪前>> Employee(5,Catty,35,100.0)
加薪后>> Employee(5,Catty,35,120.00)
加薪前>> Employee(6,Little,19,80.0)
加薪后>> Employee(6,Little,19,84.000)

2、在一组数据行内根据每条数据状态进行筛选:

  // 筛选40岁以上员工
  def filter40: FDATask[FDAROW] = row => {
    row match {
      case emp: Employee => {
        if (emp.age > 40)
          Some(List(emp))
        else fda_skip[Employee]
      }
      case _ => fda_break
    }
  }
  println("---------")
  Stream(r1,r2,r3,r4,r5,r6)
    .through(log("年龄>"))
    .through(fda_execUserTask[FDAROW](filter40))
    .through(log("合格>"))
    .run.unsafeRun
---
运算结果:
年龄>> Employee(1,John,23,100.0)
年龄>> Employee(2,Peter,25,100.0)
年龄>> Employee(3,Kay,35,100.0)
年龄>> Employee(4,Cain,45,100.0)
合格>> Employee(4,Cain,45,100.0)
年龄>> Employee(5,Catty,35,100.0)
年龄>> Employee(6,Little,19,80.0)
-

3、根据当前数据行状态终止作业:

 1   // 浏览至第一个30岁以上员工,跳出
 2   def stopOn30: FDATask[Employee] = emp => {
 3         if (emp.age > 30)
 4           fda_break
 5         else
 6           Some(List(emp))
 7   }
 8   println("---------")
 9   Stream(r1,r2,r3,r4,r5,r6)
10     .through(log("当前员工>"))
11     .through(fda_execUserTask[Employee](stopOn30))
12     .through(log("选入名单>"))
13     .run.unsafeRun
14 ---
15 运算结果:
16 当前员工>> Employee(1,John,23,100.0)
17 选入名单>> Employee(1,John,23,100.0)
18 当前员工>> Employee(2,Peter,25,100.0)
19 选入名单>> Employee(2,Peter,25,100.0)
20 当前员工>> Employee(3,Kay,35,100.0)

在这个例子里用户指定了行类型统一为Employee。

我们还可以把多个功能串接起来。像下面这样把1和2两个功能连起来:

  Stream(r1,r2,r3,r4,r5,r6)
    .through(log("加薪前>"))
    .through(fda_execUserTask[FDAROW](raisePay))
    .through(log("加薪后>"))
    .through(log("年龄>"))
    .through(fda_execUserTask[FDAROW](filter40))
    .through(log("合格>"))
    .run.unsafeRun
---
运算结果:
加薪前>> Employee(1,John,23,100.0)
加薪后>> Employee(1,John,23,110.00)
年龄>> Employee(1,John,23,110.00)
加薪前>> Employee(2,Peter,25,100.0)
加薪后>> Employee(2,Peter,25,110.00)
年龄>> Employee(2,Peter,25,110.00)
加薪前>> Employee(3,Kay,35,100.0)
加薪后>> Employee(3,Kay,35,120.00)
年龄>> Employee(3,Kay,35,120.00)
加薪前>> Employee(4,Cain,45,100.0)
加薪后>> Employee(4,Cain,45,120.00)
年龄>> Employee(4,Cain,45,120.00)
合格>> Employee(4,Cain,45,120.00)
加薪前>> Employee(5,Catty,35,100.0)
加薪后>> Employee(5,Catty,35,120.00)
年龄>> Employee(5,Catty,35,120.00)
加薪前>> Employee(6,Little,19,80.0)
加薪后>> Employee(6,Little,19,84.000)
年龄>> Employee(6,Little,19,84.000)

下面我把完整的示范代码提供给大家:

package com.bayakala.funda.fdapipes.examples
import fs2._
import com.bayakala.funda.fdapipes._
import FDANodes._
import FDAValves._
import Helpers._
object Example1 extends App {


  case class Employee(id: Int, name: String, age: Int, salary: BigDecimal) extends FDAROW
// test data set
  val r1 = Employee(1, "John", 23, 100.00)
  val r2 = Employee(2, "Peter", 25,100.00)
  val r3 = Employee(3, "Kay", 35,100.00)
  val r4 = Employee(4, "Cain", 45,100.00)
  val r5 = Employee(5, "Catty", 35,100.00)
  val r6 = Employee(6, "Little", 19,80.00)



// 20 - 30岁加10%, 30岁> 加20%,其它加 5%
  def raisePay: FDATask[FDAROW] = row => {
    row match {
      case emp: Employee => {
        val cur = emp.age match {
          case a if ((a >= 20) && (a < 30)) => emp.copy(salary = emp.salary * 1.10)
          case a if ((a >= 30)) => emp.copy(salary = emp.salary * 1.20)
          case _ => emp.copy(salary = emp.salary * 1.05)
        }
        fda_next(cur)
      }
      case _ => fda_skip
    }
  }

  Stream(r1,r2,r3,r4,r5,r6)
    .through(log("加薪前>"))
      .through(fda_execUserTask[FDAROW](raisePay))
      .through(log("加薪后>"))
    .run.unsafeRun


  // 筛选40岁以上员工
  def filter40: FDATask[FDAROW] = row => {
    row match {
      case emp: Employee => {
        if (emp.age > 40)
          Some(List(emp))
        else fda_skip[Employee]
      }
      case _ => fda_break
    }
  }
  println("---------")
  Stream(r1,r2,r3,r4,r5,r6)
    .through(log("年龄>"))
    .through(fda_execUserTask[FDAROW](filter40))
    .through(log("合格>"))
    .run.unsafeRun

  // 浏览至第一个30岁以上员工,跳出
  def stopOn30: FDATask[Employee] = emp => {
        if (emp.age > 30)
          fda_break
        else
          Some(List(emp))
  }
  println("---------")
  Stream(r1,r2,r3,r4,r5,r6)
    .through(log("当前员工>"))
    .through(fda_execUserTask[Employee](stopOn30))
    .through(log("选入名单>"))
    .run.unsafeRun


  println("---------")
  Stream(r1,r2,r3,r4,r5,r6)
    .through(log("加薪前>"))
    .through(fda_execUserTask[FDAROW](raisePay))
    .through(log("加薪后>"))
    .through(log("年龄>"))
    .through(fda_execUserTask[FDAROW](filter40))
    .through(log("合格>"))
    .run.unsafeRun

}

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏木宛城主

步步为营:三层架构+ASP.NET MVC 架构的快速搭建(入门篇)

上一篇博客中《两天完成一个小型工程报价系统》,许多朋友向我讨源码。其实我之后没发了,确实那种三层架构没什么意思,只是我个人的孤芳自赏,很多的处理都不是很成熟。...

4806
来自专栏杨建荣的学习笔记

Oracle 12c PDB浅析(r9笔记第10天)

不管怎么样,12c出来这么久,总是因为各种各样的原因没有开始学习,现在似乎还是有些晚了。总是耳闻PDB在12c是一种全新的架构模式,在各种技术聊 天也大概知道是...

2777
来自专栏圣杰的专栏

SQL递归查询知多少

最近工作中遇到了一个问题,需要根据保存的流程数据,构建流程图。数据库中保存的流程数据是树形结构的,表结构及数据如下图: ? 仔细观察表结构,会发现其树形结构的特...

2918
来自专栏me的随笔

.NET中数据访问方式(一):LINQ

语言集成查询(Language-Integrated Query),简称LINQ,.NET中的LINQ体系如下图所示:

763
来自专栏杨建荣的学习笔记

怎样突破表名30个字符的限制(r2笔记51天)

根据oracle的规范,对象的长度最大为30位,也就是说,在平时的使用中如果碰到表名长度大于30位,首先oracle是不答应的,它会提示idnetifier t...

2576
来自专栏Golang语言社区

[Go 语言社区]服务器游戏用户登陆数据读取函数

简单说下这个函数大家可以在玩家登陆的时候,返回玩家的结构体数据:玩家的基本的属性信息(玩家的名字,等级,性别等) 在返回值中我已经返回,主要在逻辑上稍...

3487
来自专栏青枫的专栏

day44_Oracle学习笔记_03

先去Oracle官网去下载最新版本的sqldeveloper,下载地址:https://www.oracle.com/technetwork/developer...

622
来自专栏Java3y

移动商城第七篇【购物车增删改查、提交订单】

把商品加入购物车 接下来我们要做的就是将商品加入到购物车中。我们这次使用的是Cookie来将用户的信息存储起来。那为什么要用cookie呢?? 如果将购物车存储...

81913
来自专栏微服务生态

用尽洪荒之力整理的Mysql数据库32条军规

2、控制单表数据量 int型不超过1000w,含char则不超过500w; 合理分表; 限制单库表数量在300以内;

653
来自专栏程序猿

系统设计之数据库范式

在设计数据库时,必须遵守一定的规则,在关系数据库中,就是范式。 字数虽少,信息量大。 什么是范式呢?(这个概念...

3506

扫码关注云+社区