Как я могу создать таблицу внешнего каталога в Apache Flink

Question

Я попытался создать и ExternalCatalog для использования в Apache Flink Table. Я создал и добавил в среду Flink table (здесь официальная документация). По какой-то причине единственная внешняя таблица присутствует в "каталоге", она не найдена во время сканирования. Что я пропустил в коде выше?

  val catalogName = s"externalCatalog$fileNumber"
  val ec: ExternalCatalog = getExternalCatalog(catalogName, 1, tableEnv)
  tableEnv.registerExternalCatalog(catalogName, ec)
  val s1: Table = tableEnv.scan("S_EXT")

  def getExternalCatalog(catalogName: String, fileNumber: Int, tableEnv: BatchTableEnvironment): ExternalCatalog = {
    val cat = new InMemoryExternalCatalog(catalogName)
    // external Catalog table
    val externalCatalogTableS = getExternalCatalogTable("S")
    // add external Catalog table
    cat.createTable("S_EXT", externalCatalogTableS, ignoreIfExists = false)
    cat
  }

  private def getExternalCatalogTable(fileName: String): ExternalCatalogTable = {
    // connector descriptor
    val connectorDescriptor = new FileSystem()
    connectorDescriptor.path(getFilePath(fileNumber, fileName))
    // format
    val fd = new Csv()
    fd.field("X", Types.STRING)
    fd.field("Y", Types.STRING)
    fd.fieldDelimiter(",")
    // statistic
    val statistics = new Statistics()
    statistics.rowCount(0)
    // metadata
    val md = new Metadata()
    ExternalCatalogTable.builder(connectorDescriptor)
      .withFormat(fd)
      .withStatistics(statistics)
      .withMetadata(md)
      .asTableSource()
  }

Приведенный выше пример является частью этого тестового файла в git.

Answer 1

Вероятно, это проблема пространства имен. Таблицы во внешних каталогах идентифицируются списком имен каталога (возможно, схем) и, наконец, именем таблицы.

В вашем примере должно работать следующее:

val s1: Table = tableEnv.scan("externalCatalog1", "S_EXT")

Вы можете взглянуть на ExternalCatalogTest, чтобы увидеть, как можно использовать внешние каталоги.