一次設計演進之旅

作者：張逸 2017-01-23 12:40:45

今天我給大家講一講關于設計演講的過程。

一、需求背景

我們需要實現對存儲在HDFS中的Parquet文件執行數據查詢，并通過REST API暴露給前端以供調用。由于查詢的結果可能數量較大，要求API接口能夠提供分頁查詢。在第一階段，需要支持的報表有5張，需要查詢的數據表與字段存在一定差異，查詢條件也有一定差異。

每個報表的查詢都牽涉到多張表的Join。每張表都被創建為數據集，對應為一個Parquet文件。Parquet文件夾名就是數據集名，名稱是系統自動生成的，所以我們需要建立業務數據表名、Join別名以及自動生成的數據集名的映射關系。數據集對應的各個字段信息都存儲在Field元數據表中，其中我們需要的三個主要屬性為：

CodeName：創建數據集時，由系統自動生成
FieldName：為客戶數據源對應數據表的字段名
DisplayName：為報表顯示的列名

說明：為了便于理解，我將要實現的五個報表分別按照序號命名。

二、解決方案

1. 前置條件

本需求是圍繞著我們已有的BI產品做定制開發。現有產品已經提供了如下功能：

通過Spark SQL讀取指定Parquet文件，但不支持同時讀取多個Parquet文件，并對獲得的DataFrame進行Join
獲取存儲在MySQL中的DataSet與Field元數據信息
基于AKKA Actor的異步查詢

2. 項目目標

交付日期非常緊急，尤其需要盡快提供最緊急的第一張報表：定期賬戶掛失后辦理支取。后續的報表也需要盡快交付，同時也應盡可能考慮到代碼的重用，因為報表查詢業務的相似度較高。

3. 整體方案

基于各個報表的具體需求，解析并生成查詢Parquet(事實上是讀取多個)的Spark SQL語句。生成的SQL語句會交給Actor，并由Actor請求Spark的SQLContext執行SQL語句，獲得DataFrame。利用take()結合zipWithIndex實現對DataFrame的分頁，轉換為前端需要的數據。

根據目前對報表的分析，生成的SQL語句包含join、where與order by子句。報表需要查詢的數據表是在系統中硬編碼的，然后通過數據表名到DataSet中查詢元數據信息，獲得真實的由系統生成的數據集名。查詢的字段名同樣通過硬編碼方式，并根據對應數據集的ID與字段名獲得Field的元數據信息。

三、設計演進

1. 引入模板方法模式

考慮到SQL語句具有一定的通用性(如select的字段、表名與join表名、on關鍵字、where條件、排序等)，差異在于不同報表需要的表名、字段以及查詢條件。通過共性與可變性分析，我把相同的實現邏輯放在一個模板方法中，而將差異的內容(也即各個報表特定的部分)交給子類去實現。這是一個典型的模板方法模式：

trait ReportTypeParser extends DataSetFetcher with ParcConfiguration { 
  def sqlFor(criteria: Option[List[Condition]]): String 
  def criteriaFields: Array[Field] 
 
  private[parc] def predefinedTables: List[TableName] 
  private[parc] def predefinedFields: List[TableField] 
 
  def generateHeaders: Array[Field] = { 
    predefinedFields.map(tf => tf.fieldName.field(tf.table.originalName)).toArray 
  }} 
 
class FirstReportTypeParser extends ReportTypeParser { 
  override def sqlFor(criteria: Option[List[Condition]]): String = { 
    s"""       
       select ${generateSelectFields}       
       from ${AccountDetailTable} a       
       left join ${AccountDebtDetailTable} b       
       left join ${AoucherJournalTable} c       
       on a.${AccountDetailTableSchema.Account.toString.codeName(AccountDetailTable)} = b.${AccountDebtDetailTableSchema.Account.toString.codeName(AccountDebtDetailTable)}       
       and a.${AccountDetailTableSchema.CustomerNo.toString.codeName(AccountDetailTable)} = c.${AoucherJournalTableSchema.CustomerNo.toString.codeName(AoucherJournalTable)}       
       where ${generateWhereClause}$       
       ${generateOrderBy}     
    """ 
  } 
 
  override private[parc] def predefinedTables: List[TableName] = ... 
  override private[parc] def predefinedFields: List[TableField] = ... 
 
  private[parc] def generateSelectFields: String = { 
    if (predefinedFields.isEmpty) "*" else predefinedFields.map(field => field.fullName).mkString(",") 
  } 
 
  private[parc] def generateWhereCluase(conditionsOpt: Option[List[Condition]]): String = { 
    def evaluate(condition: Condition): String = { 
      val aliasName = aliasNameFor(condition.originalTableName) 
 
      val codeName = fetchField(condition.fieldId) 
        .map(_.codeName) 
        .getOrElse(throw ResourceNotExistException(s"can't find the field with id ${condition.fieldId}")) 
 
      val values = condition.operator.toLowerCase() match { 
        case "between" => { 
          require(condition.values.size == 2, "the values of condition don't match between operator") 
          s"BETWEEN ${condition.values.head} AND ${condition.values.tail.head}" 
        } 
        case _ => throw BadRequestException(s"can't support operator ${condition.operator}") 
      } 
 
      s"${aliasName}.${codeName} ${values}" 
    } 
 
    conditionsOpt match { 
      case Some(conditions) if !conditions.isEmpty => s"where  ${conditions.map(c => evaluate(c)).mkString(" and ")}" 
      case _ => "" 
    } 
  }}

在ReportTypeParser中，我實現了部分可以重用的邏輯，例如generateHeaders()等方法。但是，還有部分實現邏輯放在了具體的實現類FirtReportTypeParser中，例如最主要的sqlFor方法，以及該方法調用的諸多方法，如generateSelectFields、generateWhereCluase等。

在這其中，TableName提供了表名與數據集名、別名之間的映射關系，而TableField則提供了TableName與Field之間的映射關系：

case class TableName(originalName: String,  
                     metaName: String,  
                     aliasName: String,  
                     generatedName: String = "") 
 
case class TableField(table: TableName,  
                      fieldName: String,  
                      orderType: Option[OrderType] = None)

仔細觀察sqlFor方法的實現，發現生成select的字段、生成Join的部分以及生成條件子句、排序子句都是有規律可循的。這個過程是在我不斷重構的過程中慢慢浮現出來的。我不斷找到了這些相似的方法，例如generateSelectFields、generateWhereClause這些方法。它們之間的差異只在于一些與具體報表有關的元數據上，例如表名、字段名、字段名與表名的映射、表名與別名的映射。

我首先通過pull member up重構，將這兩個方法提升到ReportTypeParser中：

trait ReportTypeParser extends ... { 
  private[parc] def generateSelectFields: String = ... 
  private[parc] def generateWhereCluase(conditionsOpt: Option[List[Condition]]): String

此外，還包括我尋找到共同規律的join部分：

trait ReportTypeParser extends ... { 
  private[parc] def generateJoinKeys: String = { 
    def joinKey(tableField: TableField): String = 
      s"${aliasNameFor(tableField.tableName)}.${tableField.fieldName.codeName(mapping.tableName)}" 
 
    predefinedJoinKeys.map{ 
      case (leftTable, rightTable) => s"${joinKey(leftTable)} = ${joinKey(rightTable)}" 
    }.mkString(" and ") 
  }}

現在sqlFor()方法就變成一個所有報表都通用的方法了，因此我也將它提升到ReportTypeParser中。

2. 元數據概念的浮現

我在最初定義諸如predefinedTables與predefinedFields等方法時，還沒有清晰地認識到所謂元數據(Metadata)的概念，然而這一系列重構后，我發現定義在FirstReportParser子類中的方法，其核心職責就是提供SQL解析所需要的元數據內容：

class FirstReportTypeParser extends ReportTypeParser { 
  private[parc] def predefinedJoinKeys: List[(TableField, TableField)] = ... 
  override private[parc] def predefinedAliasNames: Map[TableName, AliasName] = ... 
  override private[parc] def predefinedCriteriaFields: List[TableField] = ... 
  override private[parc] def predefinedOrderByFields: List[TableField] = ... 
  override private[parc] def predefinedTables: List[TableName] = ... 
  override private[parc] def predefinedFields: List[TableFieldMapping] = ... 
}

3. 以委派取代繼承

元數據的概念給了我啟發。針對報表的SQL語句解析，邏輯是完全相同的，不同之處僅在于解析的元數據而已。這就浮現出兩個不同的職責：

提供元數據
元數據解析

在變化方向上，引起這兩個職責發生變化的原因是完全不同的。不同的報表需要提供的元數據是不同的，而對于元數據的解析，則取決于Spark SQL的訪問方式(在后面我們會看到這種變化)。根據單一職責原則，我們需要將這兩個具有不同變化方向的職責分離，因此它們之間正確的依賴關系不應該是繼承，而應該是委派。

我首先引入了ReportMetadata，并將原來的FirstReportTypeParser更名為FirstReportMetadata，在實現了ReportMetadata的同時，對相關元數據的方法進行了重命名：

trait ReportMetadata extends ParcConfiguration { 
  def joinKeys: List[(TableField, TableField)] 
  def tables: List[TableName] 
  def fields: List[TableField] 
  def criteriaFields: List[TableField] 
  def orderByFields: List[TableField]}trait FirstReportMetadata extends ReportMetadata

至于原有的ReportTypeParser則被更名為ReportMetadataParser。

4. 引入Cake Pattern

如果仍然沿用之前的繼承關系，我們可以根據reportType分別創建不同報表的Parser實例。但是現在，我們需要將具體的ReportMetadata實例傳給ReportMetadataParser。至于具體傳遞什么樣的ReportMetadata實例，則取決于reportType。

這事實上是一種依賴注入。在Scala中，實現依賴注入通常是通過self type實現所謂Cake Pattern：

class ReportMetadataParser extends DataSetFetcher with ParcConfiguration { 
  self: ReportMetadata => 
 
  def evaluateSql(criteria: Option[List[Condition]]): String = { 
    s"""       
        select ${evaluateSelectFields}       
        from ${evaluateJoinTables}       
        where ${evaluateJoinKeys}       
        ${evaluateCriteria(criteria)}       
        ${evaluateOrderBy}     
    """ 
  }}

為了更清晰地表達解析的含義，我將相關方法都更名為以evaluate為前綴。通過self type，ReportMetadataParser可以訪問ReportMetadata的方法，至于具體是什么樣的實現，則取決于創建ReportMetadataParser對象時傳遞的具體類型。

通過將Metadata從Parser中分離出來，實際上是差異化編程的體現。這是我們在建立繼承體系時需要注意的。我們要學會觀察差異的部分，然后僅僅將差異的部分剝離出來，然后為其進行更通用的抽象，由此再針對實現上的差異去建立繼承體系，如分離出來的ReportMetadata。當我們要實現其他報表時，其實只需要定義ReportMetadata的實現類，提供不同的元數據，就可以滿足要求。這就使得我們能夠有效地避免代碼的重復，職責也更清晰。

5. 建立測試樁

引入Cake Pattern實現依賴注入還有利于我們編寫單元測試。例如在前面的實現中，我們通過Cake Pattern實際上注入了實現了DataSetFetcher的ReportMetadata類型。之所以需要實現DataSetFetcher，是因為我想通過它訪問數據庫中的數據集相關元數據。但是，在測試時我只想驗證sql解析的邏輯是否正確，并不希望真正去訪問數據庫。這時，我們可以建立一個DataSetFetcher的測試樁。

trait StubDataSetFetcher extends DataSetFetcher { 
    override def fetchField(dataSetId: ID, fieldName: String): Option[Field] = ... 
    override def fetchDataSetByName(dataSetName: String): Option[DataSetFetched] = ... 
    override def fetchDataSet(dataSetId: ID): Option[DataSetFetched] = ... 
}

StubDataSetFetcher通過繼承DataSetFetcher重寫了三個本來要訪問數據庫的方法，直接返回了需要的對象。然后，我再將這個trait定義在測試類中，并將其注入到ReportMetadataParser中：

class ReportMetadataParserSpec extends FlatSpec with ShouldMatchers { 
  it should "evaluate to sql for first report" in { 
    val parser = new ReportMetadataParser() with FirstReportMetadata with StubDataSetFetcher 
    val sql = parser.evaluateSql(None) 
    sql should be(expectedSql) 
  } 
}

6. 引入表達式樹

針對第一個報表，我們還有一個問題沒有解決，就是能夠支持相對復雜的where子句。例如條件：

extractDate(a.TransactionDate) < extractDate(b.DueDate) and b.LoanFlag = 'D'

不同的報表，可能會有不同的where子句。其中，extractDate函數是我自己定義的UDF。

前面提到的元數據，主要都牽涉到表名、字段名，而這里的元數據是復雜的表達式。所以，我借鑒表達式樹的概念，建立了如下的表達式元數據結構：

object ExpressionMetadata { 
  trait Expression { 
    def accept(parser: ExpressionParser): String = parser.evaluateExpression(this) 
  } 
  case class ConditionField(tableName:String, fieldName: String, funName: Option[String] = None) extends Expression 
  case class IntValue(value: Int) extends Expression 
  abstract class SingleExpression(expr: Expression) extends Expression { 
    override def accept(evaluate: Expression => String): String = 
      s"(${expr.accept(evaluate)} ${operator})" 
    def operator: String 
  } 
 
  case class IsNotNull(expr: Expression) extends SingleExpression(expr) { 
    override def operator: String = "is not null" 
  } 
 
  abstract class BinaryExpression(left: Expression, right: Expression) extends Expression { 
    override def accept(parser: ExpressionParser): String = 
      s"${left.accept(parser)} ${operator} ${right.accept(parser)}" 
    def operator: String 
  } 
  case class Equal(left: Expression, right: Expression) extends BinaryExpression(left, right) { 
    override def operator: String = "=" 
  } 
}

7. 利用模式匹配實現訪問者模式

一開始，我為各個Expression對象定義的其實是evaluate方法，而非現在的accept方法。我認為各個Expression對象都是自我完備的對象，它所擁有的知識(數據或屬性)使得它能夠自我實現解析，并利用類似合成模式的方式實現遞歸的解析。

然而在實現時我遇到了一個問題：在解析字段名時，我們不能直接用字段名來組成where子句，因為在我們產品的Parquet數據集中，字段的名字其實是系統自動生成的。我們需要獲得：

該字段對應的表的別名
該字段名在數據集中真正存儲的名稱，即code_name，例如C01。

換言之，真正要生成的條件子句應該形如：

extractDate(a.c1) < extractDate(b.c1) and b.c2 = 'D'

然而，關于表名與別名的映射則是配置在ReportMetadata中，獲得別名與codeName的方法則被定義在ReportMetadataParser的內部。如果將解析的實現邏輯放在Expression中，就需要依賴ReportMetadata與ReportMetadataParser。與之相比，我更傾向于將Expression傳給它們，讓它們完成對Expression的解析。換言之，Expression樹結構只提供數據，真正的解析職責則被委派給另外的對象，我將其定義為ExpressionParser：

trait ExpressionParser { 
  def evaluateExpression(expression: Expression): String}

這種雙重委派與樹結構的場景不正是訪問者模式最適宜的嗎?至于ExpressionParser的實現，則可以交給ReportMetadataParser：

class ReportMetadataParser extends DataSetFetcher with ParcConfiguration with ExpressionParser {override def evaluateExpression(expression: Expression): String = { 
    expression match { 
      case ConditionField(tableName, fieldName, funName) => 
         val fullName = s"${table.aliasName}.${fieldName.codeName(table.originalName)}${orderType.getOrElse("")}" 
         funName match { 
            case Some(fun) => s"${funName}(${fullName})" 
            case None => fullName 
      case IntValue(v) => s"${v}" 
      case StringValue(v) => s"'${v}'" 
    } 
  } 
 
  def evaluateWhereClause: String = { 
    if (whereClause.isEmpty) return "" 
    val clause = whereClause.map(c => c.accept(this)).mkString(" and ") 
    s"where ${clause}" 
  }}

這里的evaluateExpression方法相當于Visitor模式的visit方法。與傳統的Visitor模式不同，我不需要定義多個visit方法的重載，而是直接運用Scala的模式匹配。

evaluateWhereClause方法會對Expression的元數據whereClause進行解析，真正的實現是對每個Expression對象，執行accept(this)方法，在其內部又委派給this即ReportMetadataParser的evaluateExpression方法。

代碼中的whereClause是新增加的Metadata，具體的實現放到了FirstReportMetadata中：

override def whereClause: List[Expression] = { 
   List( 
         LessThan( 
                    ConditionField(AccountDetailTable, AccountDetailTableSchema.TransactionDate.toString, Some("extractDate")), 
                    ConditionField(AoucherJournalTable, AoucherJournalTableSchema.DueDate.toString, Some("extractDate")) 
                  ), 
         Equal( 
                ConditionField(AccountDetailTable, AccountDetailTableSchema.LoanFlag.toString), 
                StringValue("D") 
              ) 
       ) 
 }

8. 用函數取代trait定義

在Scala中，我們完全可以用函數來替代trait：

trait Expression { 
  def accept(evaluate: Expression => String): String = evaluate(this) 
} 
 
class ReportMetadataParser extends DataSetFetcher with ParcConfiguration { 
  self: ReportMetadata with DataSetFetcher => 
 
  def evaluateExpr(expression: Expression): String = { 
    expression match { 
      case ConditionField(tableName, fieldName) => 
        s"${aliasNameFor(tableName)}.${fieldName.codeName(tableName)}" 
      case IntValue(v) => s"${v}" 
      case StringValue(v) => s"'${v}'" 
    } 
  } 
 
  def evaluateWhereClause: String = { 
    if (whereClause.isEmpty) return " true " 
    whereClause.map(c => c.accept(evaluateExpr)).mkString(" and ") 
  }}

9. 演進過程的提交記錄

這個設計的過程并非事先明確進行針對性的設計，而是隨著功能的逐步實現，伴隨著對代碼的重構而逐漸浮現出來的。

整個過程的提交記錄如下圖所示(從上至下由最近到最遠)：

四、當變化發生

通過前面一系列的設計演進，代碼結構與質量已經得到了相當程度的改進與提高。關鍵是這樣的設計演進是有價值回報的。在走出分離元數據關鍵步驟之后，設計就向著好的方向在發展。

在實現了第一張報表之后，后面四張報表的開發就變得非常容易了，只需要為這四張報表提供必需的元數據信息即可。

令人欣慰的是，這個設計還經受了解決方案變化與需求變化的考驗。

1. 解決方案變化

在前面的實現中，我采用了Spark SQL的SQL方式執行查詢。查詢時通過join關聯了多張表。在生產環境上部署后，發現查詢數據集的性能不盡如人意，必須改進性能(關于性能的調優，則是另一個故事了，我會在另外的文章中講解)。由于join的表有大小表的區別，改進性能的方式是引入broadcast。雖然可以通過設置spark.sql.autoBroadcastJoinThreshold來告知Spark滿足條件時啟用broadcast，但更容易控制的方法是調用DataFrame提供的API。

于是，實現方案就需要進行調整：解析SQL的過程 ---> 組裝DataFrame API的過程

從代碼看，從原來的：

def evaluateSql(criteria: Option[List[Condition]]): String = { 
    logging { 
      s""" 
      select ${evaluateSelectFields} 
      from ${evaluateJoinTables} 
      on ${evaluateJoinKeys} 
      where ${evaluateWhereClause}${evaluateCriteria(criteria)} 
      ${evaluateOrderBy} 
      """ 
    } 
  }

變為解析各個API的參數，然后在加載DataFrame的地方調用API：

val dataFrames = tableNames.map { table => 
      load(table.generatedName).as(table.aliasName) 
    } 
    sqlContext.udf.register("extractDate", new ExtractDate) 
 
    val (joinedDF, _) = dataFrames.zipWithIndex.reduce { 
      (dfToIndex, accumulatorToIndex) => 
        val (df, index) = dfToIndex 
        val (acc, _) = accumulatorToIndex 
        (df.join(broadcast(acc), keyColumnPairs(index)._1 === keyColumnPairs(index)._2), index) 
    } 
 
    joinedDF.where(queryConditions) 
      .orderBy(orderColumns: _*) 
      .select(selectColumns: _*)

解析方式雖然有變化，但需要的元數據還是基本相似，差別在于需要將之前我自己定義的字段類型轉換為Column類型。我們僅僅只需要修改 ReportMetadataParser類，在原有基礎上，增加部分獨有的元數據解析功能：

class ReportMetadataParser extends ParcConfiguration with MortLogger { 
  def evaluateKeyPairs: List[(Column, Column)] = { 
    joinKeys.map { 
      case (leftKey, rightKey) => (leftKey.toColumn, rightKey.toColumn) 
    } 
  } 
  def evaluateSelectColumns: List[Column] = { 
    fields.map(tf => tf.toColumn) 
  } 
  def evaluateOrderColumns: List[Column] = { 
    orderByFields.map(f => f.toColumn) 
  } 
}

2. 需求變化

我們的另一個客戶同樣需要類似的需求，區別在于他們的數據治理更好，我們只需要對已經治理好的視圖數據執行查詢即可，而無需跨表Join。在對現有代碼的包結構做出調整，并定義了更為通用的Spark SQL查詢方法后，要做的工作其實就是定義對應報表的元數據罷了。

僅僅花費了1天半的時間，新客戶新項目的報表后端開發工作就完成了。要知道在如此短的開發周期內，大部分時間其實還是消耗在重構工作上，包括重新調整現有代碼的包結構，提取重用代碼。現在，我可以悠閑一點，喝喝茶，看看閑書，然后再重裝待發，迎接下一個完全不同的新項目。

【本文為51CTO專欄作者“張逸”原創稿件，轉載請聯系原作者】

戳這里，看該作者更多好文

責任編輯：趙寧寧來源： 51CTO專欄

設計演講報表數據

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

一次設計演進之旅