Flink 支持連接多種數據庫，如MySQL、PostgresSQL、Derby。Derby通常用於測試。從關系數據庫數據類型到Flink SQL數據類型的字段數據類型映射如下表所示，映射表可以幫助在Flink中輕松定義JDBC表。
| MySQL type | PostgreSQL type | Flink SQL type |
|---|---|---|
| TINYINT | TINYINT | |
| SMALLINT, TINYINT UNSIGNED |
SMALLINT, INT2, SMALLSERIAL, SERIAL2 |
SMALLINT |
| INT, MEDIUMINT, SMALLINT UNSIGNED |
INTEGER, SERIAL |
INT |
| BIGINT, INT UNSIGNED |
BIGINT, BIGSERIAL |
BIGINT |
| BIGINT UNSIGNED | DECIMAL(20,0) | |
| BIGINT | BIGINT | BIGINT |
| FLOAT | REAL, FLOAT4 |
FLOAT |
| DOUBLE, DOUBLE PRECISION |
FLOAT8, DOUBLE PRECISION |
DOUBLE |
| NUMERIC(p, s), DECIMAL(p, s) |
NUMERIC(p, s), DECIMAL(p, s) |
DECIMAL(p, s) |
| BOOLEAN, TINYINT(1) |
BOOLEAN | BOOLEAN |
| DATE | DATE | DATE |
| TIME [(p)] | TIME [(p)] [WITHOUT TIMEZONE] | TIME [(p)] [WITHOUT TIMEZONE] |
| DATETIME [(p)] | TIMESTAMP [(p)] [WITHOUT TIMEZONE] | TIMESTAMP [(p)] [WITHOUT TIMEZONE] |
| CHAR(n), VARCHAR(n), TEXT |
CHAR(n), CHARACTER(n), VARCHAR(n), CHARACTER VARYING(n), TEXT |
STRING |
| BINARY, VARBINARY, BLOB |
BYTEA | BYTES |
| ARRAY | ARRAY |
在API中，Flink嘗試使用反射從類信息中自動提取數據類型，以避免重復的手動模式工作。但是，反射性地提取數據類型並不總是成功的，因為可能會丟失邏輯信息。因此，可能需要在類或字段聲明附近添加其他信息以支持提取邏輯。下表列出了無需進一步信息即可隱式映射到數據類型的類。
注意：如果您打算在 Scala 中實現類，建議使用包裝類型（例如java.lang.Integer）而不是 Scala 的原語。Scala的原語（例如Intor Double）被編譯為JVM 原語（例如 int/double）並產生NOT NULL如下表所示的語義。此外，在泛型中使用的Scala原語（例如java.util.Map[Int, Double]）在編譯期間會被刪除，並導致類信息類似於java.util.Map[java.lang.Object, java.lang.Object]。
| Java Class | Data Type |
|---|---|
| java.lang.String | STRING |
| java.lang.Boolean | BOOLEAN |
| boolean | BOOLEAN NOT NULL |
| java.lang.Byte | TINYINT |
| byte | TINYINT NOT NULL |
| java.lang.Short | SMALLINT |
| short | SMALLINT NOT NULL |
| java.lang.Integer | INT |
| int | INT NOT NULL |
| java.lang.Long | BIGINT |
| long | BIGINT NOT NULL |
| java.lang.Float | FLOAT |
| float | FLOAT NOT NULL |
| java.lang.Double | DOUBLE |
| double | DOUBLE NOT NULL |
| java.sql.Date | DATE |
| java.time.LocalDate | DATE |
| java.sql.Time | TIME(0) |
| java.time.LocalTime | TIME(9) |
| java.sql.Timestamp | TIMESTAMP(9) |
| java.time.LocalDateTime | TIMESTAMP(9) |
| java.time.OffsetDateTime | TIMESTAMP(9) WITH TIME ZONE |
| java.time.Instant | TIMESTAMP_LTZ(9) |
| java.time.Duration | INVERVAL SECOND(9) |
| java.time.Period | INTERVAL YEAR(4) TO MONTH |
| byte[] | BYTES |
| T[] | ARRAY
|
| java.util.Map<K, V> | MAP<K, V> |
| structured type T | anonymous structured type T |
參考
https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/table/types/
https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/connectors/table/jdbc/