在对上两篇讨论中我们介绍了并行运算的两种体现方式：并行构建数据源及并行运算用户自定义函数。我们分别对这两部分进行了示范。本篇我准备示范把这两种情况集成一体的并行运算模式。这次介绍的数据源并行构建方式也与前面描述的有所不同：在前面讨论里我们预知需要从三个独立流来并行构建数据源。但如果我们有一个不知长度的数据流，它的每个元素代表不同的数据流，应该如何处理。我们知道在AQMRPT表里有从1999年到2xxx年的空气质量测量数据，我们可以试着并行把按年份生成的数据流构建成一个数据源。直接使用上期示范中的铺垫代码包括NORMAQM表初始化和从STATES和COUNTIES里用名称搜索对应id的函数：

  val db = Database.forConfig("h2db") //drop original table schema
  val futVectorTables = db.run(MTable.getTables) val futDropTable = futVectorTables.flatMap{ tables => { val tableNames = tables.map(t => t.name.name) if (tableNames.contains(NORMAQMQuery.baseTableRow.tableName)) db.run(NORMAQMQuery.schema.drop) else Future() } }.andThen { case Success(_) => println(s"Table ${NORMAQMQuery.baseTableRow.tableName} dropped successfully! ") case Failure(e) => println(s"Failed to drop Table ${NORMAQMQuery.baseTableRow.tableName}, it may not exist! Error: ${e.getMessage}") } Await.ready(futDropTable,Duration.Inf) //create new table to refine AQMRawTable
  val actionCreateTable = Models.NORMAQMQuery.schema.create val futCreateTable = db.run(actionCreateTable).andThen { case Success(_) => println("Table created successfully!") case Failure(e) => println(s"Table may exist already! Error: ${e.getMessage}") } //would carry on even fail to create table
 Await.ready(futCreateTable,Duration.Inf) //truncate data, only available in slick 3.2.1
  val futTruncateTable = futVectorTables.flatMap{ tables => { val tableNames = tables.map(t => t.name.name) if (tableNames.contains(NORMAQMQuery.baseTableRow.tableName)) db.run(NORMAQMQuery.schema.truncate) else Future() } }.andThen { case Success(_) => println(s"Table ${NORMAQMQuery.baseTableRow.tableName} truncated successfully!") case Failure(e) => println(s"Failed to truncate Table ${NORMAQMQuery.baseTableRow.tableName}! Error: ${e.getMessage}") } Await.ready(futDropTable,Duration.Inf) //a conceived task for the purpose of resource consumption //getting id with corresponding name from STATES table
  def getStateID(state: String): Int = { //create a stream for state id with state name
    implicit def toState(row:  StateTable#TableElementType) = StateModel(row.id,row.name) val stateLoader = FDAViewLoader(slick.jdbc.H2Profile)(toState _) val stateSeq = stateLoader.fda_typedRows(StateQuery.result)(db).toSeq //constructed a Stream[Task,String]
    val stateStream = fda_staticSource(stateSeq)() var id  = -1 def getid: FDAUserTask[FDAROW] = row => { row match { case StateModel(stid,stname) =>   //target row type
          if (stname.contains(state)) { id = stid fda_break //exit
 } else fda_skip   //take next row
        case _ => fda_skip } } stateStream.appendTask(getid).startRun id } //another conceived task for the purpose of resource consumption //getting id with corresponding names from COUNTIES table
  def getCountyID(state: String, county: String): Int = { //create a stream for county id with state name and county name
    implicit def toCounty(row:  CountyTable#TableElementType) = CountyModel(row.id,row.name) val countyLoader = FDAViewLoader(slick.jdbc.H2Profile)(toCounty _) val countySeq = countyLoader.fda_typedRows(CountyQuery.result)(db).toSeq //constructed a Stream[Task,String]
    val countyStream = fda_staticSource(countySeq)() var id