聊聊flink TableEnvironment的scan操做

序

本文主要研究一下flink TableEnvironment的scan操做

實例

//Scanning a directly registered table
val tab: Table = tableEnv.scan("tableName")

//Scanning a table from a registered catalog
val tab: Table = tableEnv.scan("catalogName", "dbName", "tableName")

• scan操做用於從schema讀取指定的table，也能夠傳入catalogName及dbName從指定的catalog及db讀取

TableEnvironment.scan

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/TableEnvironment.scala

abstract class TableEnvironment(val config: TableConfig) {

  private val internalSchema: CalciteSchema = CalciteSchema.createRootSchema(false, false)
  private val rootSchema: SchemaPlus = internalSchema.plus()

  //......

  @throws[TableException]
  @varargs
  def scan(tablePath: String*): Table = {
    scanInternal(tablePath.toArray) match {
      case Some(table) => table
      case None => throw new TableException(s"Table '${tablePath.mkString(".")}' was not found.")
    }
  }

  private[flink] def scanInternal(tablePath: Array[String]): Option[Table] = {
    require(tablePath != null && !tablePath.isEmpty, "tablePath must not be null or empty.")
    val schemaPaths = tablePath.slice(0, tablePath.length - 1)
    val schema = getSchema(schemaPaths)
    if (schema != null) {
      val tableName = tablePath(tablePath.length - 1)
      val table = schema.getTable(tableName)
      if (table != null) {
        return Some(new Table(this, CatalogNode(tablePath, table.getRowType(typeFactory))))
      }
    }
    None
  }

  private def getSchema(schemaPath: Array[String]): SchemaPlus = {
    var schema = rootSchema
    for (schemaName <- schemaPath) {
      schema = schema.getSubSchema(schemaName)
      if (schema == null) {
        return schema
      }
    }
    schema
  }

  //......
}

• scan方法內部調用的是scanInternal，scanInternal首先讀取catalog及db信息，而後調用getSchema方法來獲取schema
• getSchema是使用SchemaPlus的getSubSchema來按層次獲取SchemaPlus，若是沒有指定catalog及db，那麼這裏返回的是rootSchema
• 獲取到schema以後，就能夠從tablePath數組獲取tableName(數組最後一個元素)，調用SchemaPlus的getTable方法查找Table

小結

• TableEnvironment的scan操做就是從Schema中查找Table，可使用tableName，或者額外指定catalog及db來查找
• getSchema是使用SchemaPlus的getSubSchema來按層次獲取SchemaPlus，若是沒有指定catalog及db，那麼這裏返回的是rootSchema
• 獲取到schema以後，就能夠從tablePath數組獲取tableName(數組最後一個元素)，調用SchemaPlus的getTable方法查找Table

doc

• Table API