一、次设需求背景
我们需要实现对存储在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"cant find the field with id ${ condition.fieldId}")) val values = condition.operator.toLowerCase() match { case "between" => { require(condition.values.size == 2, "the values of condition dont match between operator") s"BETWEEN ${ condition.values.head} AND ${ condition.values.tail.head}" } case _ => throw BadRequestException(s"cant 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天半的时间，新客户新项目的报表后端开发工作就完成了。要知道在如此短的开发周期内，大部分时间其实还是消耗在重构工作上，包括重新调整现有代码的包结构，提取重用代码。现在，我可以悠闲一点，喝喝茶，看看闲书，然后再重装待发，迎接下一个完全不同的新项目。
【本文为专栏作者“张逸”原创稿件，转载请联系原作者】
戳这里，看该作者更多好文