import ChangeLog from ‘../changelog/connector-jdbc.md’;

Snowflake

JDBC Snowflake 源连接器

支持这些引擎

Spark
Flink
SeaTunnel Zeta

关键特性

支持查询 SQL 并可以实现投影效果。

描述

通过 JDBC 读取外部数据源数据。

支持的数据源列表

数据源支持的版本驱动连接串Maven
snowflake不同的依赖版本有不同的驱动类net.snowflake.client.jdbc.SnowflakeDriverjdbc:snowflake://<account_name>.snowflakecomputing.com下载

数据库依赖

请下载对应 ‘Maven’ 的支持列表，并将其复制到 ‘$SEATUNNEL_HOME/plugins/jdbc/lib/’ 工作目录
例如 Snowflake 数据源：cp snowflake-connector-java-xxx.jar $SEATUNNEL_HOME/plugins/jdbc/lib/

数据类型映射

Snowflake 数据类型SeaTunnel 数据类型
BOOLEANBOOLEAN
TINYINT
SMALLINT
BYTEINT		SHORT_TYPE
INT
INTEGER		INT
BIGINTLONG
DECIMAL
NUMERIC
NUMBER		DECIMAL(x,y)
DECIMAL(x,y)(>38)DECIMAL(38,18)
REAL
FLOAT4		FLOAT
DOUBLE
DOUBLE PRECISION
FLOAT8
FLOAT		DOUBLE
CHAR
CHARACTER
VARCHAR
STRING
TEXT
VARIANT
OBJECT		STRING
DATEDATE
TIMETIME
DATETIME
TIMESTAMP
TIMESTAMP_LTZ
TIMESTAMP_NTZ
TIMESTAMP_TZ		TIMESTAMP
BINARY
VARBINARY		BYTES
GEOGRAPHY (WKB or EWKB)
GEOMETRY (WKB or EWKB)		BYTES
GEOGRAPHY (GeoJSON, WKT or EWKT)
GEOMETRY (GeoJSON, WKB or EWKB)		STRING

选项

参数名类型必须默认值描述
urlString是-JDBC 连接的 URL。参考示例：jdbc:snowflake://<account_name>.snowflakecomputing.com
driverString是-用于连接到远程数据源的 jdbc 类名，如果您使用 Snowflake，值为 net.snowflake.client.jdbc.SnowflakeDriver。
usernameString否-连接实例用户名
passwordString否-连接实例密码
queryString是-查询语句
connection_check_timeout_secInt否30等待用于验证连接的数据库操作完成的时间（秒）
partition_columnString否-用于并行性分割的列名，仅支持数值类型，仅支持数值类型主键，只能配置一列。
partition_lower_boundBigDecimal否-partition_column 的最小值用于扫描，如果未设置，SeaTunnel 将查询数据库获取最小值。
partition_upper_boundBigDecimal否-partition_column 的最大值用于扫描，如果未设置，SeaTunnel 将查询数据库获取最大值。
partition_numInt否job parallelism分割数量，仅支持正整数。默认值是任务并行度。
fetch_sizeInt否0对于返回大量对象的查询，您可以配置查询中使用的行提取大小，以通过减少满足选择条件所需的数据库命中次数来提高性能。零表示使用 jdbc 默认值。
propertiesMap否-其他连接配置参数，当 properties 和 URL 具有相同参数时，优先级由驱动程序的具体实现确定。例如，在 MySQL 中，properties 优先于 URL。
common-options否-源插件通用参数，请参考 源通用选项 详见。

提示

如果未设置 partition_column，它将以单并发运行，如果设置了 partition_column，它将根据任务的并发度并行执行。

JDBC 驱动程序连接参数在 JDBC 连接字符串中受支持。例如，您可以添加 ?GEOGRAPHY_OUTPUT_FORMAT='EWKT' 来指定地理空间数据类型。有关可配置参数和地理空间数据类型的更多信息，请访问 Snowflake 官方文档

任务示例

简单

此示例在单个并行中查询您的测试“数据库”中的 type_bin 表的 16 条数据，并查询其所有字段。您也可以指定要查询的字段以最终输出到控制台。

# 定义运行时环境
env {
    parallelism = 2
   job.mode = "BATCH"
}
source {
    Jdbc {
        url = "jdbc:snowflake://<account_name>.snowflakecomputing.com"
        driver = "net.snowflake.client.jdbc.SnowflakeDriver"
        connection_check_timeout_sec = 100
        username = "root"
        password = "123456"
        query = "select * from type_bin limit 16"
    }
}
transform {
# 如果您想了解有关如何配置 seatunnel 的更多信息并查看完整的转换插件列表，
# 请访问 https://seatunnel.apache.org/docs/transforms/sql
}
sink {
   Console {}
}

并行

使用您配置的分片字段和分片数据并行读取查询表。如果您想读取整个表，可以这样做

Jdbc {
    url = "jdbc:snowflake://<account_name>.snowflakecomputing.com"
    driver = "net.snowflake.client.jdbc.SnowflakeDriver"
    connection_check_timeout_sec = 100
    username = "root"
    password = "123456"
    # 根据需要定义查询逻辑
    query = "select * from type_bin"
    # 并行分片读取字段
    partition_column = "id"
    # 分片数量
    partition_num = 10
}

并行边界

指定查询的上下边界内的数据更高效。根据您配置的上下边界读取数据源更高效

Jdbc {
    url = "jdbc:snowflake://<account_name>.snowflakecomputing.com"
    driver = "net.snowflake.client.jdbc.SnowflakeDriver"
    connection_check_timeout_sec = 100
    username = "root"
    password = "123456"
    # 根据需要定义查询逻辑
    query = "select * from type_bin"
    partition_column = "id"
    # 读取开始边界
    partition_lower_bound = 1
    # 读取结束边界
    partition_upper_bound = 500
    partition_num = 10
}

变更日志