您现在的位置是：首页 > 文章详情

Apache SeaTunnel基于JDBC连接器开发教程

日期：2025-02-19点击：66收藏

说明

以下内容旨在帮助开发人员，快速了解熟悉SeaTunnel2.3.8程序框架，并能够进行JDBC连接器开发内容。

必要知识

在进行开发前，你一定要仔细阅读如下框架文档，它们存在于SeaTunnel源码中的Docs目录，这些文档能够帮你快速上手和规范开发，如果你已经阅读过则无需关注。

docs/zh/concept/config.md
docs/zh/concept/connector-v2-features.md
docs/zh/concept/JobEnvConfig.md
docs/zh/connector-v2/sink/Jdbc.md
docs/zh/connector-v2/sink-common-options.md
docs/zh/connector-v2/source-common-options.md
连接器开发必看(https://www.modb.pro/db/447922)

SeaTunnel基础开发流程

拉取项目

 git clone https://github.com/apache/seatunnel.git

编译构建

选择Profiles
mvn打包安装本地包

mvn clean install -Dmaven.test.skip=true

运行工程样例

seatunnel-examples是SeaTunnel本地环境运行模块，运行SeaTunnelEngineLocalExample的样例配置文件fake_to_console.conf，可查看SeaTunnel运行环境是否成功。

打包发布

选择Profiles后，通过Dist模块可直接构建发布包。

JDBC连接器开发

包目录介绍

使用技巧

做JDBC连接器开发我们一般仅需要关注两部分，第一个是catalog包目录，第二个是internal中的dialect。

这两部分已经提供了区别不同数据库的差异性描述，其余大部分代码是公共使用，不建议轻易修改，否则可能会影响所有引用类。

catalog中的类介绍

MySqlCatalogFactory中使用factoryIdentifier()来标识数据库类型，optionRule()用于定制连接器参数效验规则，createCatalog()是工厂类用来创建实例的方法。

@AutoService(Factory.class) public class MySqlCatalogFactory implements CatalogFactory { [@Override](https://my.oschina.net/u/1162528) public String factoryIdentifier() { return DatabaseIdentifier.MYSQL; } [@Override](https://my.oschina.net/u/1162528) public Catalog createCatalog(String catalogName, ReadonlyConfig options) { String urlWithDatabase = options.get(JdbcCatalogOptions.BASE_URL); Preconditions.checkArgument( StringUtils.isNoneBlank(urlWithDatabase), "Miss config <base-url>! Please check your config."); JdbcUrlUtil.UrlInfo urlInfo = JdbcUrlUtil.getUrlInfo(urlWithDatabase); return new MySqlCatalog( catalogName, options.get(JdbcCatalogOptions.USERNAME), options.get(JdbcCatalogOptions.PASSWORD), urlInfo); } [@Override](https://my.oschina.net/u/1162528) public OptionRule optionRule() { return JdbcCatalogOptions.BASE_RULE.build(); } }

MySqlCatalog 中包含了对于数据库元数据的查询，例如SELECT\_DATABASE\_EXISTS（库信息查询）、SELECT\_TABLE\_EXISTS （表信息查询），还定义了一些库表生成语句，例如获取表DDL语句，并且使用getTable()方法还能直接获取到外部表。

[@Slf4j](https://my.oschina.net/slf4j) public class MySqlCatalog extends AbstractJdbcCatalog { private static final String SELECT_COLUMNS_SQL_TEMPLATE = "SELECT * FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME ='%s' ORDER BY ORDINAL_POSITION ASC"; private static final String SELECT_DATABASE_EXISTS = "SELECT SCHEMA_NAME FROM information_schema.schemata WHERE SCHEMA_NAME = '%s'"; private static final String SELECT_TABLE_EXISTS = "SELECT TABLE_SCHEMA,TABLE_NAME FROM information_schema.tables WHERE table_schema = '%s' AND table_name = '%s'"; private MySqlVersion version; private MySqlTypeConverter typeConverter; public MySqlCatalog( String catalogName, String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo) { super(catalogName, username, pwd, urlInfo, null); this.version = resolveVersion(); this.typeConverter = new MySqlTypeConverter(version); } @Override protected String getDatabaseWithConditionSql(String databaseName) { return String.format(SELECT_DATABASE_EXISTS, databaseName); } @Override protected String getTableWithConditionSql(TablePath tablePath) { return String.format( SELECT_TABLE_EXISTS, tablePath.getDatabaseName(), tablePath.getTableName()); } @Override protected String getListDatabaseSql() { return "SHOW DATABASES;"; } @Override protected String getListTableSql(String databaseName) { return "SHOW TABLES;"; } @Override protected String getTableName(ResultSet rs) throws SQLException { return rs.getString(1); } @Override protected String getTableName(TablePath tablePath) { return tablePath.getTableName(); } @Override protected String getSelectColumnsSql(TablePath tablePath) { return String.format( SELECT_COLUMNS_SQL_TEMPLATE, tablePath.getDatabaseName(), tablePath.getTableName()); } @Override protected TableIdentifier getTableIdentifier(TablePath tablePath) { return TableIdentifier.of( catalogName, tablePath.getDatabaseName(), tablePath.getTableName()); } @Override protected List<ConstraintKey> getConstraintKeys(DatabaseMetaData metaData, TablePath tablePath) throws SQLException { List<ConstraintKey> indexList = super.getConstraintKeys( metaData, tablePath.getDatabaseName(), tablePath.getSchemaName(), tablePath.getTableName()); for (Iterator<ConstraintKey> it = indexList.iterator(); it.hasNext(); ) { ConstraintKey index = it.next(); if (ConstraintKey.ConstraintType.UNIQUE_KEY.equals(index.getConstraintType()) && "PRIMARY".equals(index.getConstraintName())) { it.remove(); } } return indexList; } @Override protected Column buildColumn(ResultSet resultSet) throws SQLException { String columnName = resultSet.getString("COLUMN_NAME"); // e.g. tinyint(1) unsigned String columnType = resultSet.getString("COLUMN_TYPE"); // e.g. tinyint String dataType = resultSet.getString("DATA_TYPE").toUpperCase(); String comment = resultSet.getString("COLUMN_COMMENT"); Object defaultValue = resultSet.getObject("COLUMN_DEFAULT"); String isNullableStr = resultSet.getString("IS_NULLABLE"); boolean isNullable = isNullableStr.equals("YES"); // e.g. `decimal(10, 2)` is 10 long numberPrecision = resultSet.getInt("NUMERIC_PRECISION"); // e.g. `decimal(10, 2)` is 2 int numberScale = resultSet.getInt("NUMERIC_SCALE"); // e.g. `varchar(10)` is 40 long charOctetLength = resultSet.getLong("CHARACTER_OCTET_LENGTH"); // e.g. `timestamp(3)` is 3 int timePrecision = MySqlVersion.V_5_5.equals(version) ? 0 : resultSet.getInt("DATETIME_PRECISION"); Preconditions.checkArgument(!(numberPrecision > 0 && charOctetLength > 0)); Preconditions.checkArgument(!(numberScale > 0 && timePrecision > 0)); MysqlType mysqlType = MysqlType.getByName(columnType); boolean unsigned = columnType.toLowerCase(Locale.ROOT).contains("unsigned"); BasicTypeDefine<MysqlType> typeDefine = BasicTypeDefine.<MysqlType>builder() .name(columnName) .columnType(columnType) .dataType(dataType) .nativeType(mysqlType) .unsigned(unsigned) .length(Math.max(charOctetLength, numberPrecision)) .precision(numberPrecision) .scale(Math.max(numberScale, timePrecision)) .nullable(isNullable) .defaultValue(defaultValue) .comment(comment) .build(); return typeConverter.convert(typeDefine); } @Override protected String getCreateTableSql( TablePath tablePath, CatalogTable table, boolean createIndex) { return MysqlCreateTableSqlBuilder.builder(tablePath, table, typeConverter, createIndex) .build(table.getCatalogName()); } @Override protected String getDropTableSql(TablePath tablePath) { return String.format( "DROP TABLE `%s`.`%s`;", tablePath.getDatabaseName(), tablePath.getTableName()); } @Override protected String getCreateDatabaseSql(String databaseName) { return String.format("CREATE DATABASE `%s`;", databaseName); } @Override protected String getDropDatabaseSql(String databaseName) { return String.format("DROP DATABASE `%s`;", databaseName); } @Override public CatalogTable getTable(String sqlQuery) throws SQLException { Connection defaultConnection = getConnection(defaultUrl); return CatalogUtils.getCatalogTable( defaultConnection, sqlQuery, new MySqlTypeMapper(typeConverter)); } @Override protected String getTruncateTableSql(TablePath tablePath) throws CatalogException { return String.format( "TRUNCATE TABLE `%s`.`%s`;", tablePath.getDatabaseName(), tablePath.getTableName()); } public String getExistDataSql(TablePath tablePath) { return String.format( "SELECT * FROM `%s`.`%s` LIMIT 1;", tablePath.getDatabaseName(), tablePath.getTableName()); } private MySqlVersion resolveVersion() { try (Statement statement = getConnection(defaultUrl).createStatement(); ResultSet resultSet = statement.executeQuery("SELECT VERSION()")) { resultSet.next(); return MySqlVersion.parse(resultSet.getString(1)); } catch (Exception e) { log.info( "Failed to get mysql version, fallback to default version: {}", MySqlVersion.V_5_7, e); return MySqlVersion.V_5_7; } } }

MysqlCreateTableSqlBuilder和MysqlDataTypeConvertor是用于获取表DDL语句和类型转换器方法，较为简单这里就不一一赘述。

dialect中的类介绍

MySqlDialectFactory使用工厂方式创建MySqlDialect实例，使用acceptsURL()方法判断不同JDBC URL，从而识别不同数据库类型。

create()的重载方法，主要是用于数据库端兼容多个数据库方言的场景。

@AutoService(JdbcDialectFactory.class) public class MySqlDialectFactory implements JdbcDialectFactory { @Override public boolean acceptsURL(String url) { return url.startsWith("jdbc:mysql:"); } @Override public JdbcDialect create() { return new MysqlDialect(); } @Override public JdbcDialect create(@Nonnull String compatibleMode, String fieldIde) { if (DatabaseIdentifier.STARROCKS.equalsIgnoreCase(compatibleMode)) { return new StarRocksDialect(fieldIde); } return new MysqlDialect(fieldIde); } }

MySqlDialect 的getRowConverter()和getTypeConverter()用于获取行对象转换器和类型转换器。

行对象转换器主要提供JDBC数据对象 ⇔ \Leftrightarrow ⇔ SeaTunnelRow对象互转方法。

类型转换器主要提供JDBC数据类型 ⇔ \Leftrightarrow ⇔ SeaTunnel数据类型互转的方法，这两者共同作用于引擎侧内部对象和JDBC数据集的转入转出。

MySqlDialect中还定义了一些不通数据库操作的特性功能，例如Upsert功能实现和转义符号等。

MySqlDialect还能使用defaultParameter()方法为JDBC URL串定制一些默认的参数。

public class MysqlDialect implements JdbcDialect { private static final List NOT_SUPPORTED_DEFAULT_VALUES = Arrays.asList(MysqlType.BLOB, MysqlType.TEXT, MysqlType.JSON, MysqlType.GEOMETRY); public String fieldIde = FieldIdeEnum.ORIGINAL.getValue(); public MysqlDialect() { } public MysqlDialect(String fieldIde) { this.fieldIde = fieldIde; } @Override public String dialectName() { return DatabaseIdentifier.MYSQL; } @Override public JdbcRowConverter getRowConverter() { return new MysqlJdbcRowConverter(); } @Override public TypeConverter<BasicTypeDefine> getTypeConverter() { TypeConverter typeConverter = MySqlTypeConverter.DEFAULT_INSTANCE; return typeConverter; } @Override public JdbcDialectTypeMapper getJdbcDialectTypeMapper() { return new MySqlTypeMapper(); } @Override public String quoteIdentifier(String identifier) { return "`" + getFieldIde(identifier, fieldIde) + "`"; } @Override public String quoteDatabaseIdentifier(String identifier) { return "`" + identifier + "`"; } @Override public String tableIdentifier(TablePath tablePath) { return tableIdentifier(tablePath.getDatabaseName(), tablePath.getTableName()); } @Override public Optional<String> getUpsertStatement( String database, String tableName, String[] fieldNames, String[] uniqueKeyFields) { String updateClause = Arrays.stream(fieldNames) .map( fieldName -> quoteIdentifier(fieldName) + "=VALUES(" + quoteIdentifier(fieldName) + ")") .collect(Collectors.joining(", ")); String upsertSQL = getInsertIntoStatement(database, tableName, fieldNames) + " ON DUPLICATE KEY UPDATE " + updateClause; return Optional.of(upsertSQL); } @Override public PreparedStatement creatPreparedStatement( Connection connection, String queryTemplate, int fetchSize) throws SQLException { PreparedStatement statement = connection.prepareStatement( queryTemplate, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); statement.setFetchSize(Integer.MIN_VALUE); return statement; } @Override public String extractTableName(TablePath tablePath) { return tablePath.getTableName(); } @Override public Map<String, String> defaultParameter() { HashMap<String, String> map = new HashMap<>(); map.put("rewriteBatchedStatements", "true"); return map; } @Override public TablePath parse(String tablePath) { return TablePath.of(tablePath, false); } @Override public Object[] sampleDataFromColumn( Connection connection, JdbcSourceTable table, String columnName, int samplingRate, int fetchSize) throws Exception { String sampleQuery; if (StringUtils.isNotBlank(table.getQuery())) { sampleQuery = String.format( "SELECT %s FROM (%s) AS T", quoteIdentifier(columnName), table.getQuery()); } else { sampleQuery = String.format( "SELECT %s FROM %s", quoteIdentifier(columnName), tableIdentifier(table.getTablePath())); } try (Statement stmt = connection.createStatement( ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) { stmt.setFetchSize(Integer.MIN_VALUE); try (ResultSet rs = stmt.executeQuery(sampleQuery)) { int count = 0; List<Object> results = new ArrayList<>(); while (rs.next()) { count++; if (count % samplingRate == 0) { results.add(rs.getObject(1)); } if (Thread.currentThread().isInterrupted()) { throw new InterruptedException("Thread interrupted"); } } Object[] resultsArray = results.toArray(); Arrays.sort(resultsArray); return resultsArray; } } } @Override public Long approximateRowCntStatement(Connection connection, JdbcSourceTable table) throws SQLException { // 1. If no query is configured, use TABLE STATUS. // 2. If a query is configured but does not contain a WHERE clause and tablePath is // configured , use TABLE STATUS. // 3. If a query is configured with a WHERE clause, or a query statement is configured but // tablePath is TablePath.DEFAULT, use COUNT(*). boolean useTableStats = StringUtils.isBlank(table.getQuery()) || (!table.getQuery().toLowerCase().contains("where") && table.getTablePath() != null && !TablePath.DEFAULT .getFullName() .equals(table.getTablePath().getFullName())); if (useTableStats) { // The statement used to get approximate row count which is less // accurate than COUNT(*), but is more efficient for large table. TablePath tablePath = table.getTablePath(); String useDatabaseStatement = String.format("USE %s;", quoteDatabaseIdentifier(tablePath.getDatabaseName())); String rowCountQuery = String.format("SHOW TABLE STATUS LIKE '%s';", tablePath.getTableName()); try (Statement stmt = connection.createStatement()) { log.info("Split Chunk, approximateRowCntStatement: {}", useDatabaseStatement); stmt.execute(useDatabaseStatement); log.info("Split Chunk, approximateRowCntStatement: {}", rowCountQuery); try (ResultSet rs = stmt.executeQuery(rowCountQuery)) { if (!rs.next() || rs.getMetaData().getColumnCount() < 5) { throw new SQLException( String.format( "No result returned after running query [%s]", rowCountQuery)); } return rs.getLong(5); } } } return SQLUtils.countForSubquery(connection, table.getQuery()); } @Override public boolean supportDefaultValue(BasicTypeDefine typeBasicTypeDefine) { MysqlType nativeType = (MysqlType) typeBasicTypeDefine.getNativeType(); return !(NOT_SUPPORTED_DEFAULT_VALUES.contains(nativeType)); } @Override public boolean needsQuotesWithDefaultValue(BasicTypeDefine columnDefine) { MysqlType mysqlType = MysqlType.getByName(columnDefine.getColumnType()); switch (mysqlType) { case CHAR: case VARCHAR: case TEXT: case TINYTEXT: case MEDIUMTEXT: case LONGTEXT: case ENUM: case SET: case BLOB: case TINYBLOB: case MEDIUMBLOB: case LONGBLOB: case DATE: case DATETIME: case TIMESTAMP: case TIME: case YEAR: return true; default: return false; } } }

MySqlTypeConverter类只存在convert和reconvert，分别对应JDBC类型转换为SEATUNNEL 类型以及SEATUNNEL类型转换为JDBC类型，同时值得注意的是，在运行SEATUNNEL任务时，若存在该类中没有被定义的类型转换规则，则会抛出运行时异常UNSUPPORTED_DATA_TYPE，所以铁子们要想支持兼容更多的数据类型，应该从此类下手。

public class MySqlTypeConverter implements TypeConverter<BasicTypeDefine<MysqlType>> { // ============================data types===================== static final String MYSQL_NULL = "NULL"; static final String MYSQL_BIT = "BIT"; static final String MYSQL_BIT_UNSIGNED = "BIT UNSIGNED"; // -------------------------number---------------------------- static final String MYSQL_TINYINT = "TINYINT"; static final String MYSQL_TINYINT_UNSIGNED = "TINYINT UNSIGNED"; static final String MYSQL_SMALLINT = "SMALLINT"; static final String MYSQL_SMALLINT_UNSIGNED = "SMALLINT UNSIGNED"; static final String MYSQL_MEDIUMINT = "MEDIUMINT"; static final String MYSQL_MEDIUMINT_UNSIGNED = "MEDIUMINT UNSIGNED"; static final String MYSQL_INT = "INT"; static final String MYSQL_INT_UNSIGNED = "INT UNSIGNED"; static final String MYSQL_INTEGER = "INTEGER"; static final String MYSQL_INTEGER_UNSIGNED = "INTEGER UNSIGNED"; static final String MYSQL_BIGINT = "BIGINT"; static final String MYSQL_BIGINT_UNSIGNED = "BIGINT UNSIGNED"; static final String MYSQL_DECIMAL = "DECIMAL"; static final String MYSQL_DECIMAL_UNSIGNED = "DECIMAL UNSIGNED"; static final String MYSQL_FLOAT = "FLOAT"; static final String MYSQL_FLOAT_UNSIGNED = "FLOAT UNSIGNED"; static final String MYSQL_DOUBLE = "DOUBLE"; static final String MYSQL_DOUBLE_UNSIGNED = "DOUBLE UNSIGNED"; // -------------------------string---------------------------- public static final String MYSQL_CHAR = "CHAR"; public static final String MYSQL_VARCHAR = "VARCHAR"; static final String MYSQL_TINYTEXT = "TINYTEXT"; static final String MYSQL_MEDIUMTEXT = "MEDIUMTEXT"; static final String MYSQL_TEXT = "TEXT"; static final String MYSQL_LONGTEXT = "LONGTEXT"; static final String MYSQL_JSON = "JSON"; static final String MYSQL_ENUM = "ENUM"; // ------------------------------time------------------------- static final String MYSQL_DATE = "DATE"; public static final String MYSQL_DATETIME = "DATETIME"; public static final String MYSQL_TIME = "TIME"; public static final String MYSQL_TIMESTAMP = "TIMESTAMP"; static final String MYSQL_YEAR = "YEAR"; static final String MYSQL_YEAR_UNSIGNED = "YEAR UNSIGNED"; // ------------------------------blob------------------------- static final String MYSQL_TINYBLOB = "TINYBLOB"; static final String MYSQL_MEDIUMBLOB = "MEDIUMBLOB"; static final String MYSQL_BLOB = "BLOB"; static final String MYSQL_LONGBLOB = "LONGBLOB"; static final String MYSQL_BINARY = "BINARY"; static final String MYSQL_VARBINARY = "VARBINARY"; static final String MYSQL_GEOMETRY = "GEOMETRY"; public static final int DEFAULT_PRECISION = 38; public static final int MAX_PRECISION = 65; public static final int DEFAULT_SCALE = 18; public static final int MAX_SCALE = 30; public static final int MAX_TIME_SCALE = 6; public static final int MAX_TIMESTAMP_SCALE = 6; public static final long POWER_2_8 = (long) Math.pow(2, 8); public static final long POWER_2_16 = (long) Math.pow(2, 16); public static final long POWER_2_24 = (long) Math.pow(2, 24); public static final long POWER_2_32 = (long) Math.pow(2, 32); public static final long MAX_VARBINARY_LENGTH = POWER_2_16 - 4; public static final MySqlTypeConverter DEFAULT_INSTANCE = new MySqlTypeConverter(MySqlVersion.V_5_7); private final MySqlVersion version; public MySqlTypeConverter(MySqlVersion version) { this.version = version; } public MySqlTypeConverter() { this(MySqlVersion.V_5_7); } @Override public String identifier() { return DatabaseIdentifier.MYSQL; } @Override public Column convert(BasicTypeDefine typeDefine) { PhysicalColumn.PhysicalColumnBuilder builder = PhysicalColumn.builder() .name(typeDefine.getName()) .sourceType(typeDefine.getColumnType()) .nullable(typeDefine.isNullable()) .defaultValue(typeDefine.getDefaultValue()) .comment(typeDefine.getComment()); String mysqlDataType = typeDefine.getDataType().toUpperCase(); if (mysqlDataType.endsWith("ZEROFILL")) { mysqlDataType = mysqlDataType.substring(0, mysqlDataType.length() - "ZEROFILL".length()).trim(); } if (typeDefine.isUnsigned() && !(mysqlDataType.endsWith(" UNSIGNED"))) { mysqlDataType = mysqlDataType + " UNSIGNED"; } switch (mysqlDataType) { case MYSQL_NULL: builder.dataType(BasicType.VOID_TYPE); break; case MYSQL_BIT: case MYSQL_BIT_UNSIGNED: if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) { builder.dataType(BasicType.BOOLEAN_TYPE); } else if (typeDefine.getLength() == 1) { builder.dataType(BasicType.BOOLEAN_TYPE); } else { builder.dataType(PrimitiveByteArrayType.INSTANCE); // BIT(M) -> BYTE(M/8) long byteLength = typeDefine.getLength() / 8; byteLength += typeDefine.getLength() % 8 > 0 ? 1 : 0; builder.columnLength(byteLength); } break; case MYSQL_TINYINT: if (typeDefine.getColumnType().equalsIgnoreCase("tinyint(1)")) { builder.dataType(BasicType.BOOLEAN_TYPE); } else { builder.dataType(BasicType.BYTE_TYPE); } break; case MYSQL_TINYINT_UNSIGNED: case MYSQL_SMALLINT: builder.dataType(BasicType.SHORT_TYPE); break; case MYSQL_SMALLINT_UNSIGNED: case MYSQL_MEDIUMINT: case MYSQL_MEDIUMINT_UNSIGNED: case MYSQL_INT: case MYSQL_INTEGER: case MYSQL_YEAR: case MYSQL_YEAR_UNSIGNED: builder.dataType(BasicType.INT_TYPE); break; case MYSQL_INT_UNSIGNED: case MYSQL_INTEGER_UNSIGNED: case MYSQL_BIGINT: builder.dataType(BasicType.LONG_TYPE); break; case MYSQL_BIGINT_UNSIGNED: DecimalType intDecimalType = new DecimalType(20, 0); builder.dataType(intDecimalType); builder.columnLength(Long.valueOf(intDecimalType.getPrecision())); builder.scale(intDecimalType.getScale()); break; case MYSQL_FLOAT: builder.dataType(BasicType.FLOAT_TYPE); break; case MYSQL_FLOAT_UNSIGNED: log.warn("{} will probably cause value overflow.", MYSQL_FLOAT_UNSIGNED); builder.dataType(BasicType.FLOAT_TYPE); break; case MYSQL_DOUBLE: builder.dataType(BasicType.DOUBLE_TYPE); break; case MYSQL_DOUBLE_UNSIGNED: log.warn("{} will probably cause value overflow.", MYSQL_DOUBLE_UNSIGNED); builder.dataType(BasicType.DOUBLE_TYPE); break; case MYSQL_DECIMAL: Preconditions.checkArgument(typeDefine.getPrecision() > 0); DecimalType decimalType; if (typeDefine.getPrecision() > DEFAULT_PRECISION) { log.warn("{} will probably cause value overflow.", MYSQL_DECIMAL); decimalType = new DecimalType(DEFAULT_PRECISION, DEFAULT_SCALE); } else { decimalType = new DecimalType( typeDefine.getPrecision().intValue(), typeDefine.getScale() == null ? 0 : typeDefine.getScale().intValue()); } builder.dataType(decimalType); builder.columnLength(Long.valueOf(decimalType.getPrecision())); builder.scale(decimalType.getScale()); break; case MYSQL_DECIMAL_UNSIGNED: Preconditions.checkArgument(typeDefine.getPrecision() > 0); log.warn("{} will probably cause value overflow.", MYSQL_DECIMAL_UNSIGNED); DecimalType decimalUnsignedType = new DecimalType( typeDefine.getPrecision().intValue() + 1, typeDefine.getScale() == null ? 0 : typeDefine.getScale().intValue()); builder.dataType(decimalUnsignedType); builder.columnLength(Long.valueOf(decimalUnsignedType.getPrecision())); builder.scale(decimalUnsignedType.getScale()); break; case MYSQL_ENUM: builder.dataType(BasicType.STRING_TYPE); if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) { builder.columnLength(100L); } else { builder.columnLength(typeDefine.getLength()); } break; case MYSQL_CHAR: case MYSQL_VARCHAR: if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) { builder.columnLength(TypeDefineUtils.charTo4ByteLength(1L)); } else { builder.columnLength(typeDefine.getLength()); } builder.dataType(BasicType.STRING_TYPE); break; case MYSQL_TINYTEXT: builder.dataType(BasicType.STRING_TYPE); builder.columnLength(POWER_2_8 - 1); break; case MYSQL_TEXT: builder.dataType(BasicType.STRING_TYPE); builder.columnLength(POWER_2_16 - 1); break; case MYSQL_MEDIUMTEXT: builder.dataType(BasicType.STRING_TYPE); builder.columnLength(POWER_2_24 - 1); break; case MYSQL_LONGTEXT: builder.dataType(BasicType.STRING_TYPE); builder.columnLength(POWER_2_32 - 1); break; case MYSQL_JSON: builder.dataType(BasicType.STRING_TYPE); break; case MYSQL_BINARY: case MYSQL_VARBINARY: if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) { builder.columnLength(1L); } else { builder.columnLength(typeDefine.getLength()); } builder.dataType(PrimitiveByteArrayType.INSTANCE); break; case MYSQL_TINYBLOB: builder.dataType(PrimitiveByteArrayType.INSTANCE); builder.columnLength(POWER_2_8 - 1); break; case MYSQL_BLOB: builder.dataType(PrimitiveByteArrayType.INSTANCE); builder.columnLength(POWER_2_16 - 1); break; case MYSQL_MEDIUMBLOB: builder.dataType(PrimitiveByteArrayType.INSTANCE); builder.columnLength(POWER_2_24 - 1); break; case MYSQL_LONGBLOB: builder.dataType(PrimitiveByteArrayType.INSTANCE); builder.columnLength(POWER_2_32 - 1); break; case MYSQL_GEOMETRY: builder.dataType(PrimitiveByteArrayType.INSTANCE); break; case MYSQL_DATE: builder.dataType(LocalTimeType.LOCAL_DATE_TYPE); break; case MYSQL_TIME: builder.dataType(LocalTimeType.LOCAL_TIME_TYPE); builder.scale(typeDefine.getScale()); break; case MYSQL_DATETIME: case MYSQL_TIMESTAMP: builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE); builder.scale(typeDefine.getScale()); break; default: throw CommonError.convertToSeaTunnelTypeError( DatabaseIdentifier.MYSQL, mysqlDataType, typeDefine.getName()); } return builder.build(); } @Override public BasicTypeDefine<MysqlType> reconvert(Column column) { BasicTypeDefine.BasicTypeDefineBuilder builder = BasicTypeDefine.<MysqlType>builder() .name(column.getName()) .nullable(column.isNullable()) .comment(column.getComment()) .defaultValue(column.getDefaultValue()); switch (column.getDataType().getSqlType()) { case NULL: builder.nativeType(MysqlType.NULL); builder.columnType(MYSQL_NULL); builder.dataType(MYSQL_NULL); break; case BOOLEAN: builder.nativeType(MysqlType.BOOLEAN); builder.columnType(String.format("%s(%s)", MYSQL_TINYINT, 1)); builder.dataType(MYSQL_TINYINT); builder.length(1L); break; case TINYINT: builder.nativeType(MysqlType.TINYINT); builder.columnType(MYSQL_TINYINT); builder.dataType(MYSQL_TINYINT); break; case SMALLINT: builder.nativeType(MysqlType.SMALLINT); builder.columnType(MYSQL_SMALLINT); builder.dataType(MYSQL_SMALLINT); break; case INT: builder.nativeType(MysqlType.INT); builder.columnType(MYSQL_INT); builder.dataType(MYSQL_INT); break; case BIGINT: builder.nativeType(MysqlType.BIGINT); builder.columnType(MYSQL_BIGINT); builder.dataType(MYSQL_BIGINT); break; case FLOAT: builder.nativeType(MysqlType.FLOAT); builder.columnType(MYSQL_FLOAT); builder.dataType(MYSQL_FLOAT); break; case DOUBLE: builder.nativeType(MysqlType.DOUBLE); builder.columnType(MYSQL_DOUBLE); builder.dataType(MYSQL_DOUBLE); break; case DECIMAL: DecimalType decimalType = (DecimalType) column.getDataType(); long precision = decimalType.getPrecision(); int scale = decimalType.getScale(); if (precision <= 0) { precision = DEFAULT_PRECISION; scale = DEFAULT_SCALE; log.warn( "The decimal column {} type decimal({},{}) is out of range, " + "which is precision less than 0, " + "it will be converted to decimal({},{})", column.getName(), decimalType.getPrecision(), decimalType.getScale(), precision, scale); } else if (precision > MAX_PRECISION) { scale = (int) Math.max(0, scale - (precision - MAX_PRECISION)); precision = MAX_PRECISION; log.warn( "The decimal column {} type decimal({},{}) is out of range, " + "which exceeds the maximum precision of {}, " + "it will be converted to decimal({},{})", column.getName(), decimalType.getPrecision(), decimalType.getScale(), MAX_PRECISION, precision, scale); } if (scale < 0) { scale = 0; log.warn( "The decimal column {} type decimal({},{}) is out of range, " + "which is scale less than 0, " + "it will be converted to decimal({},{})", column.getName(), decimalType.getPrecision(), decimalType.getScale(), precision, scale); } else if (scale > MAX_SCALE) { scale = MAX_SCALE; log.warn( "The decimal column {} type decimal({},{}) is out of range, " + "which exceeds the maximum scale of {}, " + "it will be converted to decimal({},{})", column.getName(), decimalType.getPrecision(), decimalType.getScale(), MAX_SCALE, precision, scale); } builder.nativeType(MysqlType.DECIMAL); builder.columnType(String.format("%s(%s,%s)", MYSQL_DECIMAL, precision, scale)); builder.dataType(MYSQL_DECIMAL); builder.precision(precision); builder.scale(scale); break; case BYTES: if (column.getColumnLength() == null || column.getColumnLength() <= 0) { builder.nativeType(MysqlType.VARBINARY); builder.columnType( String.format("%s(%s)", MYSQL_VARBINARY, MAX_VARBINARY_LENGTH / 2)); builder.dataType(MYSQL_VARBINARY); } else if (column.getColumnLength() < MAX_VARBINARY_LENGTH) { builder.nativeType(MysqlType.VARBINARY); builder.columnType( String.format("%s(%s)", MYSQL_VARBINARY, column.getColumnLength())); builder.dataType(MYSQL_VARBINARY); } else if (column.getColumnLength() < POWER_2_24) { builder.nativeType(MysqlType.MEDIUMBLOB); builder.columnType(MYSQL_MEDIUMBLOB); builder.dataType(MYSQL_MEDIUMBLOB); } else { builder.nativeType(MysqlType.LONGBLOB); builder.columnType(MYSQL_LONGBLOB); builder.dataType(MYSQL_LONGBLOB); } break; case STRING: if (column.getColumnLength() == null || column.getColumnLength() <= 0) { builder.nativeType(MysqlType.LONGTEXT); builder.columnType(MYSQL_LONGTEXT); builder.dataType(MYSQL_LONGTEXT); } else if (column.getColumnLength() < POWER_2_8) { builder.nativeType(MysqlType.VARCHAR); builder.columnType( String.format("%s(%s)", MYSQL_VARCHAR, column.getColumnLength())); builder.dataType(MYSQL_VARCHAR); } else if (column.getColumnLength() < POWER_2_16) { builder.nativeType(MysqlType.TEXT); builder.columnType(MYSQL_TEXT); builder.dataType(MYSQL_TEXT); } else if (column.getColumnLength() < POWER_2_24) { builder.nativeType(MysqlType.MEDIUMTEXT); builder.columnType(MYSQL_MEDIUMTEXT); builder.dataType(MYSQL_MEDIUMTEXT); } else { builder.nativeType(MysqlType.LONGTEXT); builder.columnType(MYSQL_LONGTEXT); builder.dataType(MYSQL_LONGTEXT); } break; case DATE: builder.nativeType(MysqlType.DATE); builder.columnType(MYSQL_DATE); builder.dataType(MYSQL_DATE); break; case TIME: builder.nativeType(MysqlType.TIME); builder.dataType(MYSQL_TIME); if (version.isAtOrBefore(MySqlVersion.V_5_5)) { builder.columnType(MYSQL_TIME); } else if (column.getScale() != null && column.getScale() > 0) { int timeScale = column.getScale(); if (timeScale > MAX_TIME_SCALE) { timeScale = MAX_TIME_SCALE; log.warn( "The time column {} type time({}) is out of range, " + "which exceeds the maximum scale of {}, " + "it will be converted to time({})", column.getName(), column.getScale(), MAX_SCALE, timeScale); } builder.columnType(String.format("%s(%s)", MYSQL_TIME, timeScale)); builder.scale(timeScale); } else { builder.columnType(MYSQL_TIME); } break; case TIMESTAMP: builder.nativeType(MysqlType.DATETIME); builder.dataType(MYSQL_DATETIME); if (version.isAtOrBefore(MySqlVersion.V_5_5)) { builder.columnType(MYSQL_DATETIME); } else if (column.getScale() != null && column.getScale() > 0) { int timestampScale = column.getScale(); if (timestampScale > MAX_TIMESTAMP_SCALE) { timestampScale = MAX_TIMESTAMP_SCALE; log.warn( "The timestamp column {} type timestamp({}) is out of range, " + "which exceeds the maximum scale of {}, " + "it will be converted to timestamp({})", column.getName(), column.getScale(), MAX_TIMESTAMP_SCALE, timestampScale); } builder.columnType(String.format("%s(%s)", MYSQL_DATETIME, timestampScale)); builder.scale(timestampScale); } else { builder.columnType(MYSQL_DATETIME); } break; default: throw CommonError.convertToConnectorTypeError( DatabaseIdentifier.MYSQL, column.getDataType().getSqlType().name(), column.getName()); } return builder.build(); } }

MySqlTypeMapper 是老版本的TypeConverter，该类也是引用的MySqlTypeConverter。

MySqlVersion 是版本控制。