我尝试创建和 ExternalCatalog 以在 Apache Flink 表中使用。我创建并添加到 Flink 表环境(这里是官方文档)。由于某种原因,“目录”中存在的唯一外部表在扫描期间未找到。我在上面的代码中遗漏了什么?
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中这个测试文件的一部分。