JDBC 连接器提供了对 MySQL、PostgreSQL 等常见的关系型数据库的读写能力，支持做数据源表、结果表和维表。

注意事项 #

如果您需要使用 JDBC 连接器连接云数据库 veDB MySQL 版，您的连接终端请按照以下要求配置，否则可能会因为自定义连接终端的限制而出现任务故障。
如需详细了解各参数含义，请参见编辑连接终端。

• 读写模式：配置为读写。
• 一致性级别：配置为最终一致性。
• 主节点接受读：关闭该选项。
• 事务拆分：打开该选项。
• 基于合规要求，如果需要使用 JDBC MySQL 相关功能，需要用户自行上传 MySQL Driver 。请前往 MySQL 官网下载 8.0.27 版本，并且上传到作业开发依赖文件中。具体操作可以参考 使用 JDBC 或者 MySQL-CDC 数据源 。

DDL 定义 #

用作数据源（Source） #

CREATE TABLE jdbc_source (
    name String,
    score INT
 ) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://192.*.*.*:3306/doc_db',  
    'table-name' = '<yourtablename>',
    'username' = 'admin',          
    'password' = 'Passord',      
    'scan.partition.column' = 'score',       
    'scan.partition.num' = '2',           
    'scan.partition.lower-bound' = '0',     
    'scan.partition.upper-bound' = '100',     
    'scan.fetch-size' = '1' 
);

用作数据目的（Sink） #

CREATE TABLE jdbc_sink (
    name String,
    score INT
 ) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://192.*.*.*:3306/doc_db',     
    'table-name' = '<yourtablename>',
    'username' = 'admin',          
    'password' = 'Pa$$w0rd',      
    'sink.buffer-flush.max-rows' = '100'
);

用作数据维表（Lookup） #

CREATE TABLE jdbc_lookup (
    name String,
    score INT
 ) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://192.*.*.*:3306/doc_db',     
    'table-name' = '<yourtablename>',
    'username' = 'admin',          
    'password' = 'MyPa$$w0rd',      
    'lookup.cache.max-rows' = '100',
    'lookup.max-retries' = '3'
);

WITH 参数 #

示例代码 #

• 结果表。

  CREATE TABLE datagen_source (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_update_time as localtimestamp
      )
  WITH (
       'connector' = 'datagen',
       'rows-per-second' = '5'            
  );
  
  
  CREATE TABLE jdbc_sink (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_update_time timestamp,
       PRIMARY KEY (order_id) NOT ENFORCED
   ) WITH (
      'connector' = 'jdbc',
      'url' = 'jdbc:mysql://mysq**3fb.rds.ivolces.com:3306/doc_db', 
      'username' = 'admin',      
      'password' = '***',  
      'table-name' = 'orders'
  );
  
  
  insert into jdbc_sink
  select * from datagen_source; 
  

• 源表。

  CREATE TABLE jdbc_source (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp,
       PRIMARY KEY (order_id) NOT ENFORCED
   ) WITH (
      'connector' = 'jdbc',
      'url' = 'jdbc:mysql://mysq**3fb.rds.ivolces.com:3306/doc_db', 
      'username' = 'admin',      
      'password' = 'Passord',  
      'table-name' = 'source_table'
  );
  
   CREATE TABLE print_sink (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp
      )
  WITH (
       'connector' = 'print'           
  );
  
  insert into print_sink
  select * from jdbc_source; 
  

FAQ #

部分列更新 #

JDBC Sink 支持部分列更新，只需要在 DDL 定义的时候，定义需要更新的列即可。
比如一张表有 id，a，b，c 四个列，如果只对 a 做部分列更新，则 DDL 只需要定义 id 和 a 两个字段。

CREATE TABLE jdbc_sink ( 
     `id` bigint, 
     a string
     PRIMARY KEY (`id`) NOT ENFORCED 
 ) WITH ( 
    'connector' = 'jdbc', 
    'url' = 'jdbc:mysql://mysq**3fb.rds.ivolces.com:3306/doc_db',  
    'username' = 'admin',       
    'password' = '***',   
    'table-name' = 'test' 
);