Spark SQL 字段血缘在 vivo 互联网的实践

作者：vivo互联网服务器团队-Hao Guangshi

【Analyzer Rules】逻辑计划分析规则  
   【Check Analysis Rules】逻辑计划检查规则  
   【Optimizer Rules.】 逻辑计划优化规则  
   【Planning Strategies】形成物理计划的策略  
   【Customized Parser】自定义的sql解析器  
   【(External) Catalog listeners catalog】监听器

class ExtralSparkExtension extends (SparkSessionExtensions => Unit) {
  override def apply(spark: SparkSessionExtensions): Unit = {
  //字段血缘
  spark.injectCheckRule(FieldLineageCheckRuleV3)
  //sql解析器
  spark.injectParser { case (_, parser) => new ExtraSparkParser(parser) }
  }
}

class ExtraSparkParser(delegate: ParserInterface) extends ParserInterface with Logging{
  override def parsePlan(sqlText: String): LogicalPlan = {
  val lineAgeEnabled = SparkSession.getActiveSession
    .get.conf.getOption("spark.sql.xxx-xxx-xxx.enable").getOrElse("false").toBoolean
  logDebug(s"SqlText: $sqlText")
  if(sqlText.toLowerCase().contains("insert")){
    if(lineAgeEnabled){
    if(FIELD_LINE_AGE_SQL_COULD_SET.get()){
      //线程本地变量在这里
      FIELD_LINE_AGE_SQL.set(sqlText)
    }
    FIELD_LINE_AGE_SQL_COULD_SET.remove()
    }
  }
  delegate.parsePlan(sqlText)
  }
  //调用原始的sqlparser
  override def parseExpression(sqlText: String): Expression = {
  delegate.parseExpression(sqlText)
  }
  //调用原始的sqlparser
  override def parseTableIdentifier(sqlText: String): TableIdentifier = {
  delegate.parseTableIdentifier(sqlText)
  }
  //调用原始的sqlparser
  override def parseFunctionIdentifier(sqlText: String): FunctionIdentifier = {
  delegate.parseFunctionIdentifier(sqlText)
  }
  //调用原始的sqlparser
  override def parseTableSchema(sqlText: String): StructType = {
  delegate.parseTableSchema(sqlText)
  }
  //调用原始的sqlparser
  override def parseDataType(sqlText: String): DataType = {
  delegate.parseDataType(sqlText)
  }
}

case class FieldLineageCheckRuleV3(sparkSession:SparkSession) extends (LogicalPlan=>Unit ) {
  val executor: ThreadPoolExecutor =
  ThreadUtils.newDaemonCachedThreadPool("spark-field-line-age-collector",3,6)
  override def apply(plan: LogicalPlan): Unit = {
  val sql = FIELD_LINE_AGE_SQL.get
  FIELD_LINE_AGE_SQL.remove()
  if(sql != null){
    //这里我们拿到sql然后启动一个线程做剩余的解析任务
    val task = new FieldLineageRunnableV3(sparkSession,sql)
    executor.execute(task)
  }
  }
}

override def run(): Unit = {
  val parser = sparkSession.sessionState.sqlParser
  val analyzer = sparkSession.sessionState.analyzer
  val optimizer = sparkSession.sessionState.optimizer
  val planner = sparkSession.sessionState.planner
    ............
  val newPlan = parser.parsePlan(sql)
  PASS_TABLE_AUTH.set(true)
  val analyzedPlan = analyzer.executeAndCheck(newPlan)
  val optimizerPlan = optimizer.execute(analyzedPlan)
  //得到sparkPlan
  val sparkPlan = planner.plan(optimizerPlan).next()
  ...............
if(targetTable != null){
  val levelProject = new ArrayBuffer[ArrayBuffer[NameExpressionHolder]]()
  val predicates = new ArrayBuffer[(String,ArrayBuffer[NameExpressionHolder])]()
  //projection
  projectionLineAge(levelProject, sparkPlan.child)
  //predication
  predicationLineAge(predicates, sparkPlan.child)
  ...............

select item_id,TYPE,v_value,imei from t1
union all
select item_id,TYPE,v_value,imei from t2
union all
select item_id,TYPE,v_value,imei from t3

id ->tab1.id , 
 name->tab1.name,tabb2.name, 
 age→tabb2.age。

with A as (select id,name,age from tab1 where id > 100 ) ,
C as (select id,name,max(age) from A group by A.id,A.name) ,
B as (select id,name,age from tabb2 where age > 28)
insert into tab3
   select C.id,concat(C.name,B.name) as name, B.age from
   B,C where C.id = B.id

{
  "edges": [
  {
    "sources": [
    3
    ],
    "targets": [
    0
    ],
    "expression": "id",
    "edgeType": "PROJECTION"
  },
  {
    "sources": [
    4,
    7
    ],
    "targets": [
    1
    ],
    "expression": "name",
    "edgeType": "PROJECTION"
  },
  {
    "sources": [
    5
    ],
    "targets": [
    2
    ],
    "expression": "age",
    "edgeType": "PROJECTION"
  },
  {
    "sources": [
    6,
    3
    ],
    "targets": [
    0,
    1,
    2
    ],
    "expression": "INNER",
    "edgeType": "PREDICATE"
  },
  {
    "sources": [
    6,
    5
    ],
    "targets": [
    0,
    1,
    2
    ],
    "expression": "((((default.tabb2.`age` IS NOT NULL) AND (CAST(default.tabb2.`age` AS INT) > 28)) AND (B.`id` > 100)) AND (B.`id` IS NOT NULL))",
    "edgeType": "PREDICATE"
  },
  {
    "sources": [
    3
    ],
    "targets": [
    0,
    1,
    2
    ],
    "expression": "((default.tab1.`id` IS NOT NULL) AND (default.tab1.`id` > 100))",
    "edgeType": "PREDICATE"
  }
  ],
  "vertices": [
  {
    "id": 0,
    "vertexType": "COLUMN",
    "vertexId": "default.tab3.id"
  },
  {
    "id": 1,
    "vertexType": "COLUMN",
    "vertexId": "default.tab3.name"
  },
  {
    "id": 2,
    "vertexType": "COLUMN",
    "vertexId": "default.tab3.age"
  },
  {
    "id": 3,
    "vertexType": "COLUMN",
    "vertexId": "default.tab1.id"
  },
  {
    "id": 4,
    "vertexType": "COLUMN",
    "vertexId": "default.tab1.name"
  },
  {
    "id": 5,
    "vertexType": "COLUMN",
    "vertexId": "default.tabb2.age"
  },
  {
    "id": 6,
    "vertexType": "COLUMN",
    "vertexId": "default.tabb2.id"
  },
  {
    "id": 7,
    "vertexType": "COLUMN",
    "vertexId": "default.tabb2.name"
  }
  ]
}