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

Redshift

JDBC Redshift 源连接器

描述

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

支持这些引擎

Spark
Flink
Seatunnel Zeta

对于 Spark/Flink 引擎

  1. 您需要确保 jdbc 驱动程序 jar 包 已放置在目录 ${SEATUNNEL_HOME}/plugins/ 中。

对于 SeaTunnel Zeta 引擎

  1. 您需要确保 jdbc 驱动程序 jar 包 已放置在目录 ${SEATUNNEL_HOME}/lib/ 中。

关键特性

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

支持的数据源列表

数据源支持的版本驱动连接串Maven
redshift不同的依赖版本有不同的驱动类com.amazon.redshift.jdbc.Driverjdbc:redshift://localhost:5439/database下载

数据库依赖

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

数据类型映射

Redshift 数据类型SeaTunnel 数据类型
SMALLINT
INT2		SHORT
INTEGER
INT
INT4		INT
BIGINT
INT8
OID		LONG
DECIMAL
NUMERIC		DECIMAL
REAL
FLOAT4		FLOAT
DOUBLE_PRECISION
FLOAT8
FLOAT		DOUBLE
BOOLEAN
BOOL		BOOLEAN
CHAR
CHARACTER
NCHAR
BPCHAR
VARCHAR
CHARACTER_VARYING
NVARCHAR
TEXT
SUPER		STRING
VARBYTE
BINARY_VARYING		BYTES
TIME
TIME_WITH_TIME_ZONE
TIMETZ		LOCALTIME
TIMESTAMP
TIMESTAMP_WITH_OUT_TIME_ZONE
TIMESTAMPTZ		LOCALDATETIME

示例

简单

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

env {
  parallelism = 2
  job.mode = "BATCH"
}
source{
    Jdbc {
        url = "jdbc:redshift://localhost:5439/dev"
        driver = "com.amazon.redshift.jdbc.Driver"
        username = "root"
        password = "123456"
        
        table_path = "public.table2"
        # 使用查询过滤行和列
        query = "select id, name from public.table2 where id > 100"
        
        #split.size = 8096
        #split.even-distribution.factor.upper-bound = 100
        #split.even-distribution.factor.lower-bound = 0.05
        #split.sample-sharding.threshold = 1000
        #split.inverse-sampling.rate = 1000
    }
}

sink {
    Console {}
}

多表读取

配置 table_list 将打开自动分割，您可以配置 split.* 来调整分割策略

env {
  job.mode = "BATCH"
  parallelism = 2
}
source {
  Jdbc {
    url = "jdbc:redshift://localhost:5439/dev"
    driver = "com.amazon.redshift.jdbc.Driver"
    username = "root"
    password = "123456"

    table_list = [
      {
        table_path = "public.table1"
      },
      {
        table_path = "public.table2"
        # 使用查询过滤行和列
        query = "select id, name from public.table2 where id > 100"
      }
    ]
    #split.size = 8096
    #split.even-distribution.factor.upper-bound = 100
    #split.even-distribution.factor.lower-bound = 0.05
    #split.sample-sharding.threshold = 1000
    #split.inverse-sampling.rate = 1000
  }
}

sink {
  Console {}
}

变更日志