FunDA（4）－ 數據流內容控制：Stream data element control

来源:http://www.cnblogs.com/tiger-xc/archive/2017/01/19/6305731.html
-Advertisement-
Play Games

上節我們探討了通過scalaz-stream-fs2來驅動一套數據處理流程，用fs2的Pipe類型來實現對數據流的逐行操作。本篇討論準備在上節討論的基礎上對數據流的流動和元素操作進行優化完善。如數據流動中增加諸如next、skip、eof功能、內容控制中增加對行元素的append、insert、up ...


    上節我們探討了通過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

}

 

 

 

 

 

 

 

 


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 溫習SQL語句
    作為一名使用C#語言開發人員，就很難逃脫與SQLSERVER打交道，雖說我們是開發人員，但我想說的是，對資料庫的操作還是應該時不時的拿出來溫習一番。下麵那就是我見過的一道有趣的SQL題目，與你們一起分享！ 題目：在我的TeamPK資料庫中有一個叫department的表，裡面只有一個欄位name,一 ...
  • .NET程式集引用COM組件MSScriptControl所遇到的問題
    問題描述：為了在C#中執行js腳本，在一個目標平臺編譯為Any Cpu的.NET程式集中引用了MSScriptControl組件，在winform程式中，調用這個程式集中的執行js的方法，沒有任何問題。但是在windows服務中調用卻報錯，報錯信息如下： 在嘗試過很多方法之後仍然沒有解決，包括將MS ...
  • 獲取類的屬性併排除特定屬性（getType().GetProperties()）
    當獲取一個類型（class）的所有屬性時，想排除指定屬性，該如何操作？ 比如：EF中一個實體類型UserEntity，通過反射獲取這個類的屬性時，想排除這個為映射的欄位ID 使用以下方法即可！ 參考：http://stackoverflow.com/questions/2051834/exclude ...
  • Java WEB開發環境搭建以及創建Maven Web項目
    根據此鏈接博文學習配置: http://www.cnblogs.com/zyw-205520/p/4767633.html 1.JDK的安裝 自行百度,(最好是jdk1.7版本的) 測試如下圖,即完成jdk的安裝 2.MyEclipse安裝 自行下載安裝即可,(我使用的是2013版的) 3.Tomc ...
  • tomcat相關知識
    1、隱藏tomcat版本： ①執行命令：cd /usr/local/tomcat_web/lib/ && ll 我們需要對catalina.jar進行解壓（最好提前先備份一下） ②執行命令：unzip catalina.jar 這時候會多出META-INF和org兩個目錄，找到org/apache/ ...
  • 如何做好程式猿？--個人心得
    摘要 在寫這篇文章之前，自己思索了很多，也是因為一些事情觸發了自己，使得自己想寫這麼一篇文章。也算是對2016年自己的一個總結吧。 正文 先說說我自己吧！本人16年於一所二本學校畢業,考過研，夢想著上北航，結果卻因政治考的太差沒能上，說起來很慚愧。英雄都不提往事，何況我這一個失敗的狗熊呢！不，是失敗 ...
  • C++程式設計與語言(特別版) -- 基本功能(上)
    第4章 類型和聲明 4.2 布爾量 按照定義，true具有值1，而false具有值0.整數可以隱式地轉換到bool值。指針也可以隱式地轉換到bool，非零指針轉為true，具有零值的指針轉為false。 4.3 字元類型 一個char類型幾乎都包含8個二進位位，這裡還提供了另一個類型wchar_t用 ...
  • java util - 時間工具包 PrettyTime
    需要 prettytime-3.2.3.Final.jar 包 代碼例子 輸出結果 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類