曝光！Apache SeaTunnel Catalog 功能设计为何能大大简化用户启用步骤？

Catalog（目录）提供了关于数据库、表格和访问数据所需的信息的元数据，以及统一的 API 来管理元数据，验证连接，让元数据对 Sources（数据源）、Sinks（数据汇）和 Web 可访问。

初始化操作

Java
public interface CatalogFactory extends Factory { String factoryIdentifier(); OptionRule optionRule(); Catalog createCatalog(String catalogName, ReadonlyConfig options); } public interface Catalog extends AutoCloseable { void open() throws CatalogException; void close() throws CatalogException; }

数据库操作

java
public interface Catalog extends AutoCloseable { // -------------------------------------------------------------------------------------------- // 数据库 // -------------------------------------------------------------------------------------------- String getDefaultDatabase() throws CatalogException; boolean databaseExists(String databaseName) throws CatalogException; List<String> listDatabases() throws CatalogException; void createDatabase(String databaseName, boolean ignoreIfExists) throws DatabaseAlreadyExistException, CatalogException; void dropDatabase(String databaseName, boolean ignoreIfNotExists) throws DatabaseNotExistException, CatalogException; }

表格操作

java
public interface Catalog extends AutoCloseable { // -------------------------------------------------------------------------------------------- // 表格 // -------------------------------------------------------------------------------------------- List<String> listTables(String databaseName) throws CatalogException, DatabaseNotExistException; boolean tableExists(TablePath tablePath) throws CatalogException; CatalogTable getTable(TablePath tablePath) throws CatalogException, TableNotExistException; void createTable(TablePath tablePath, CatalogTable table, boolean ignoreIfExists) throws TableAlreadyExistException, DatabaseNotExistException, CatalogException; void dropTable(TablePath tablePath, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException; }

MySQL Catalog 的使用方式：

• username [String] 连接到数据库服务器时要使用的数据库名称。
• password [String] 连接到数据库服务器时要使用的密码。
• base-url [String] URL 必须包含数据库，例如 "jdbc:mysql://localhost:5432/db" 或 "jdbc:mysql://localhost:5432/db?useSSL=true"。
• table-names [List] 要捕获的数据库表格名称列表。表格名称需要包括数据库名称，例如：database_name.table_name。
• database-pattern [String] 要捕获的数据库名称的正则表达式。
• table-pattern [String] 要捕获的数据库表格名称的正则表达式。表格名称需要包括数据库名称，例如：database_..table_.。

配置文件配置

conf
[source/sink] { [connector-factory-id] { catalog { factory = "MySQL" username = "test" password = "123456" base-url = "jdbc:mysql://localhost:5432/db" table-names = [ "db.table" ] } } }

如何使用 Catalog

示例

sql
env { "job.mode"=STREAMING "job.name"="cdc_mysql_to_mysql" "checkpoint.interval"="2000" "custom_parameters"="" } source { MySQL-CDC { parallelism = 1 catalog { factory = "MySQL" # 默认情况下，Catalog 将使用与连接器同名的选项 } username = "mysqluser" password = "mysqlpw" database-names = ["seatunnel-test"] table-pattern = "seatunnel-test\.orders_\d+" base-url = "jdbc:mysql://localhost:54508/seatunnel-test" } } sink { jdbc { url = "jdbc:mysql://localhost:4000/test" driver = "com.mysql.cj.jdbc.Driver" catalog { factory = "MySQL" username = "root" password = "" base-url = "jdbc:mysql://localhost:4000/test" table-pattern = "seatunnel-test2\.orders_\d+" } user = "root" password = "" query = "insert into sink(age, name) values(?,?)" } }