Spark SQL 欄位血緣在 vivo 網際網路的實踐

語言: CN / TW / HK

時間 2022-04-20 21:52:19 vivo網際網路技術

作者：

vivo網際網路伺服器團隊-Hao Guangshi

一、背景

欄位血緣是在表處理的過程中將欄位的處理過程保留下來。為什麼會需要欄位血緣呢？

有了欄位間的血緣關係，便可以知道資料的來源去處，以及欄位之間的轉換關係，這樣對資料的質量，治理有很大的幫助。

Spark SQL 相對於 Hive 來說通常情況下效率會比較高，對於執行時間、資源的使用上面等都會有較大的收益。

平臺計劃將 Hive 任務遷移到 Spark SQL 上，同時也需要實現欄位血緣的功能。

二、前期調研

開發前我們做了很多相關調研，從中得知 Spark 是支援擴充套件的：允許使用者對 Spark SQL 的 SQL 解析、邏輯計劃的分析和檢查、邏輯計劃的優化、物理計劃的形成等進行擴充套件。

該方案可行，且對 Spark 的原始碼沒有改動，代價也比較小，確定使用該方案。

三、Spark SQL 擴充套件

3.1 Spark 可擴充套件的內容

SparkSessionExtensions是比較重要的一個類，其中定義了注入規則的方法，現在支援以下內容：

【Analyzer Rules】邏輯計劃分析規則
【Check Analysis Rules】邏輯計劃檢查規則
【Optimizer Rules.】邏輯計劃優化規則
【Planning Strategies】形成物理計劃的策略
【Customized Parser】自定義的sql解析器
【(External) Catalog listeners catalog】監聽器

在以上六種可以使用者自定義的地方，我們選擇了【Check Analysis Rules】。因為該檢查規則在方法呼叫的時候是不需要有返回值的，也就意味著不需要對當前遍歷的邏輯計劃樹進行修改，這正是我們需要的。

而【Analyzer Rules】、【Optimizer Rules】則需要對當前的邏輯計劃進行修改，使得我們難以迭代整個樹，難以得到我們想要的結果。

3.2 實現自己的擴充套件

class ExtralSparkExtension extends (SparkSessionExtensions => Unit) {
  override def apply(spark: SparkSessionExtensions): Unit = {


    //欄位血緣
    spark.injectCheckRule(FieldLineageCheckRuleV3)


    //sql解析器
    spark.injectParser { case (_, parser) => new ExtraSparkParser(parser) }


  }
}

上面按照這種方式實現擴充套件，並在 apply 方法中把自己需要的規則注入到 SparkSessionExtensions 即可，除了以上四種可以注入的以外還有其他的規則。要讓 ExtralSparkExtension 起到作用的話我們需要在spark-default.conf下配置

spark.sql.extensions=org.apache.spark.sql.hive.ExtralSparkExtension

在啟動 Spark 任務的時候即可生效。

注意到我們也實現了一個自定義的SQL解析器，其實該解析器並沒有做太多的事情。只是在判斷如果該語句包含insert的時候就將 SQLText（SQL語句）設定到一個為 FIELD_LINE_AGE_SQL ，之所以將SQLText放到 FIELD_LINE_AGE_SQL裡面。因為在 DheckRule 裡面是拿不到SparkPlan的我們需要對SQL再次解析拿到 SprkPlan，而FieldLineageCheckRuleV3的實現也特別簡單，重要的在另一個執行緒實現裡面。

這裡我們只關注了insert語句，因為插入語句裡面有從某些個表裡面輸入然後寫入到某個表。

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)
  }
}

3.3 擴充套件的規則類

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)
    }


  }
}

很簡單，我們只是拿到了 SQL 然後便啟動了一個執行緒去得到 SparkPlan，實際邏輯在

FieldLineageRunnableV3。

3.4 具體的實現方法

3.4.1 得到 SparkPlan

我們在 run 方法中得到 SparkPlan：

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)
  ...............

為什麼要使用 SparkPlan 呢？當初我們考慮的時候，物理計劃拿取欄位關係的時候是比較準的，且鏈路比較短也更直接。

在這裡補充一下 Spark SQL 解析的過程如下：

經過SqlParser後會得到邏輯計劃，此時表名、函式等都沒有解析，還不能執行；經過Analyzer會分析一些繫結資訊，例如表驗證、欄位資訊、函式資訊；經過Optimizer 後邏輯計劃會根據既定規則被優化，這裡的規則是RBO，當然 Spark 還支援CBO的優化；經過SparkPlanner後就成了可執行的物理計劃。

我們看一個邏輯計劃與物理計劃對比的例子：

一個 SQL 語句：

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

邏輯計劃是這樣的：

物理計劃是這樣的：

顯然簡化了很多。

得到 SparkPlan 後，我們就可以根據不同的SparkPlan節點做迭代處理。

我們將欄位血緣分為兩種型別：projection（select查詢欄位）、predication(wehre查詢條件)。

這兩種是一種點對點的關係,即從原始表的欄位生成目標表的欄位的對應關係。

想象一個查詢是一棵樹，那麼迭代關係會如下從樹的頂端開始迭代，直到樹的葉子節點，葉子節點即為原始表：

那麼我們迭代查詢的結果應該為

id ->tab1.id ,

name->tab1.name,tabb2.name,

age→tabb2.age。

注意到有該變數

val levelProject = new ArrayBuffer

[ArrayBuffer[NameExpressionHolder]]()，通過projecti-onLineAge 迭代後 levelProject 儲存了頂層id，name,age對應的（tab1.id），（tab1.name,tabb2.name），（tabb2.age）。

當然也不是簡單的遞迴迭代，還需要考慮特殊情況例如：Join、ExplandExec、Aggregate、Explode、GenerateExec等都需要特殊考慮。

例子及效果：

SQL:

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"
    }
  ]
}

四、總結

在 Spark SQL 的欄位血緣實現中，我們通過其自擴充套件，首先拿到了 insert 語句，在我們自己的檢查規則中拿到

SQL 語句，通過SparkSqlParser、Analyzer、Optimizer、SparkPlanner，最終得到了物理計劃。

我們通過迭代物理計劃，根據不同執行計劃做對應的轉換，然後就得到了欄位之間的對應關係。當前的實現是比較簡單的，欄位之間是直線的對應關係，中間過程被忽略，如果想實現欄位的轉換的整個過程也是沒有問題的。

「其他文章」