Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add jdbc options and support oracle mode
Browse files Browse the repository at this point in the history
@whhe
whhe committed Jun 5, 2023

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature.
GPG key ID: B5690EEEBB952194
Verified
Learn about vigilant mode
1 parent c2a6456 commit edc376a
Showing 13 changed files with 467 additions and 40 deletions.
2 changes: 1 addition & 1 deletion README.md
View file
Original file line number Diff line number Diff line change
@@ -11,7 +11,7 @@ This README is meant as a brief walkthrough on the core features of CDC Connecto
|------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------|
| [mongodb-cdc](docs/content/connectors/mongodb-cdc.md) | <li> [MongoDB](https://www.mongodb.com): 3.6, 4.x, 5.0 | MongoDB Driver: 4.3.1 |
| [mysql-cdc](docs/content/connectors/mysql-cdc.md) | <li> [MySQL](https://dev.mysql.com/doc): 5.6, 5.7, 8.0.x <li> [RDS MySQL](https://www.aliyun.com/product/rds/mysql): 5.6, 5.7, 8.0.x <li> [PolarDB MySQL](https://www.aliyun.com/product/polardb): 5.6, 5.7, 8.0.x <li> [Aurora MySQL](https://aws.amazon.com/cn/rds/aurora): 5.6, 5.7, 8.0.x <li> [MariaDB](https://mariadb.org): 10.x <li> [PolarDB X](https://github.com/ApsaraDB/galaxysql): 2.0.1 | JDBC Driver: 8.0.27 |
| [oceanbase-cdc](/docs/content/connectors/oceanbase-cdc.md) | <li> [OceanBase CE](https://open.oceanbase.com): 3.1.x <li> [OceanBase EE](https://www.oceanbase.com/product/oceanbase) (MySQL mode): 2.x, 3.x, 4.x | JDBC Driver: 5.1.4x |
| [oceanbase-cdc](/docs/content/connectors/oceanbase-cdc.md) | <li> [OceanBase CE](https://open.oceanbase.com): 3.1.x, 4.x <li> [OceanBase EE](https://www.oceanbase.com/product/oceanbase): 2.x, 3.x, 4.x | OceanBase Driver: 2.4.x |
| [oracle-cdc](docs/content/connectors/oracle-cdc.md) | <li> [Oracle](https://www.oracle.com/index.html): 11, 12, 19 | Oracle Driver: 19.3.0.0 |
| [postgres-cdc](docs/content/connectors/postgres-cdc.md) | <li> [PostgreSQL](https://www.postgresql.org): 9.6, 10, 11, 12 | JDBC Driver: 42.2.27 |
| [sqlserver-cdc](docs/content/connectors/sqlserver-cdc.md) | <li> [Sqlserver](https://www.microsoft.com/sql-server): 2012, 2014, 2016, 2017, 2019 | JDBC Driver: 7.2.2.jre8 |
2 changes: 1 addition & 1 deletion docs/content/about.md
View file
Original file line number Diff line number Diff line change
@@ -11,7 +11,7 @@ The CDC Connectors for Apache Flink<sup>®</sup> integrate Debezium as the engin
|----------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------|
| [mongodb-cdc](connectors/mongodb-cdc.md) | <li> [MongoDB](https://www.mongodb.com): 3.6, 4.x, 5.0 | MongoDB Driver: 4.3.1 |
| [mysql-cdc](connectors/mysql-cdc.md) | <li> [MySQL](https://dev.mysql.com/doc): 5.6, 5.7, 8.0.x <li> [RDS MySQL](https://www.aliyun.com/product/rds/mysql): 5.6, 5.7, 8.0.x <li> [PolarDB MySQL](https://www.aliyun.com/product/polardb): 5.6, 5.7, 8.0.x <li> [Aurora MySQL](https://aws.amazon.com/cn/rds/aurora): 5.6, 5.7, 8.0.x <li> [MariaDB](https://mariadb.org): 10.x <li> [PolarDB X](https://github.com/ApsaraDB/galaxysql): 2.0.1 | JDBC Driver: 8.0.27 |
| [oceanbase-cdc](connectors/oceanbase-cdc.md) | <li> [OceanBase CE](https://open.oceanbase.com): 3.1.x <li> [OceanBase EE](https://www.oceanbase.com/product/oceanbase) (MySQL mode): 2.x, 3.x, 4.x | JDBC Driver: 5.1.4x |
| [oceanbase-cdc](connectors/oceanbase-cdc.md) | <li> [OceanBase CE](https://open.oceanbase.com): 3.1.x, 4.x <li> [OceanBase EE](https://www.oceanbase.com/product/oceanbase): 2.x, 3.x, 4.x | OceanBase Driver: 2.4.x |
| [oracle-cdc](connectors/oracle-cdc.md) | <li> [Oracle](https://www.oracle.com/index.html): 11, 12, 19 | Oracle Driver: 19.3.0.0 |
| [postgres-cdc](connectors/postgres-cdc.md) | <li> [PostgreSQL](https://www.postgresql.org): 9.6, 10, 11, 12 | JDBC Driver: 42.2.12 |
| [sqlserver-cdc](connectors/sqlserver-cdc.md) | <li> [Sqlserver](https://www.microsoft.com/sql-server): 2012, 2014, 2016, 2017, 2019 | JDBC Driver: 7.2.2.jre8 |
124 changes: 124 additions & 0 deletions docs/content/connectors/oceanbase-cdc(ZH).md
View file
Original file line number Diff line number Diff line change
@@ -15,6 +15,16 @@ OceanBase CDC 连接器允许从 OceanBase 读取快照数据和增量数据。
</dependency>
```

如果您是要连接企业版的 OceanBase，您可能需要使用 OceanBase 官方的 JDBC 驱动，这时需要引入如下依赖。

```xml
<dependency>
<groupId>com.oceanbase</groupId>
<artifactId>oceanbase-client</artifactId>
<version>2.4.2</version>
</dependency>
```

## 下载 SQL 客户端 JAR 包

```下载链接仅在已发布版本可用，请在文档网站左下角选择浏览已发布的版本。```
@@ -23,6 +33,8 @@ OceanBase CDC 连接器允许从 OceanBase 读取快照数据和增量数据。

**注意:** flink-sql-connector-oceanbase-cdc-XXX-SNAPSHOT 版本是开发分支`release-XXX`对应的快照版本，快照版本用户需要下载源代码并编译相应的 jar。用户应使用已经发布的版本，例如 [flink-sql-connector-oceanbase-cdc-2.3.0.jar](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-oceanbase-cdc) 当前已发布的所有版本都可以在 Maven 中央仓库获取。

对于 JDBC 驱动，上述的 cdc jar 文件中已经包含了我们推荐的 MySQL 驱动版本 5.1.47。由于开源许可证的原因，我们不能在上述 cdc jar 文件中包含 OceanBase 的官方 JDBC 驱动，如果您需要使用它，可以从[这里](https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.2/oceanbase-client-2.4.2.jar)下载，然后放到 `<FLINK_HOME>/lib/` 目录下，同时需要将配置项 `jdbc.driver` 设为 `com.oceanbase.jdbc.Driver`。

### 配置 OceanBase 数据库和 oblogproxy 服务

1. 按照 [文档](https://github.com/oceanbase/oceanbase#quick-start) 配置 OceanBase 集群。
@@ -248,6 +260,20 @@ OceanBase CDC 连接器包括用于 SQL 和 DataStream API 的选项，如下表
<td>String</td>
<td>日志代理中 `libobcdc` 的工作模式 , 可以是 `storage` 或 `memory`。</td>
</tr>
<tr>
<td>jdbc.driver</td>
<td>否</td>
<td style="word-wrap: break-word;">com.mysql.jdbc.Driver</td>
<td>String</td>
<td>全量读取时使用的 jdbc 驱动类名。</td>
</tr>
<tr>
<td>jdbc.properties.*</td>
<td>否</td>
<td style="word-wrap: break-word;">无</td>
<td>String</td>
<td>传递自定义 JDBC URL 属性的选项。用户可以传递自定义属性，如 'jdbc.properties.useSSL' = 'false'。</td>
</tr>
</tbody>
</table>
</div>
@@ -396,6 +422,7 @@ public class OceanBaseSourceExample {
.tableName("^test_table$")
.hostname("127.0.0.1")
.port(2881)
.jdbcDriver("com.mysql.jdbc.Driver")
.logProxyHost("127.0.0.1")
.logProxyPort(2983)
.serverTimeZone(serverTimeZone)
@@ -597,3 +624,100 @@ public class OceanBaseSourceExample {
</tbody>
</table>
</div>
### Oracle 模式
<div class="wy-table-responsive">
<table class="colwidths-auto docutils">
<thead>
<tr>
<th class="text-left">OceanBase type</th>
<th class="text-left">Flink SQL type</th>
<th class="text-left">NOTE</th>
</tr>
</thead>
<tbody>
<tr>
<td>NUMBER(1)</td>
<td>BOOLEAN</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 3 </td>
<td>TINYINT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 5 </td>
<td>SMALLINT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 10 </td>
<td>INT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 19 </td>
<td>BIGINT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), 19 <=p - s <=38</td>
<td>DECIMAL(p - s, 0)</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s > 0)</td>
<td>DECIMAL(p, s)</td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s> 38 </td>
<td>STRING</td>
<td></td>
</tr>
<tr>
<td>
FLOAT<br>
BINARY_FLOAT
</td>
<td>FLOAT</td>
<td></td>
</tr>
<tr>
<td>BINARY_DOUBLE</td>
<td>DOUBLE</td>
<td></td>
</tr>
<tr>
<td>
DATE<br>
TIMESTAMP [(p)]
</td>
<td>TIMESTAMP [(p)]</td>
<td></td>
</tr>
<tr>
<td>
CHAR(n)<br>
NCHAR(n)<br>
VARCHAR(n)<br>
VARCHAR2(n)<br>
NVARCHAR2(n)<br>
CLOB<br>
</td>
<td>STRING</td>
<td></td>
</tr>
<tr>
<td>
RAW<br>
BLOB<br>
ROWID
</td>
<td>BYTES</td>
<td></td>
</tr>
</tbody>
</table>
</div>
124 changes: 124 additions & 0 deletions docs/content/connectors/oceanbase-cdc.md
View file
Original file line number Diff line number Diff line change
@@ -16,6 +16,16 @@ In order to set up the OceanBase CDC connector, the following table provides dep
</dependency>
```

If you want to use OceanBase JDBC driver to connect to the enterprise edition database, you should also include the following dependency in your class path.

```xml
<dependency>
<groupId>com.oceanbase</groupId>
<artifactId>oceanbase-client</artifactId>
<version>2.4.2</version>
</dependency>
```

### SQL Client JAR

```Download link is available only for stable releases.```
@@ -24,6 +34,8 @@ Download [flink-sql-connector-oceanbase-cdc-2.4-SNAPSHOT.jar](https://repo1.mave

**Note:** flink-sql-connector-oceanbase-cdc-XXX-SNAPSHOT version is the code corresponding to the development branch. Users need to download the source code and compile the corresponding jar. Users should use the released version, such as [flink-sql-connector-oceanbase-cdc-2.2.1.jar](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-oceanbase-cdc), the released version will be available in the Maven central warehouse.

For JDBC driver, the cdc jar above already contains MySQL JDBC driver 5.1.47, which is our recommended version. Due to the license issue, we can not include the OceanBase JDBC driver in the cdc jar. If you need to use it, you can download it from [here](https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.2/oceanbase-client-2.4.2.jar) and put it under `<FLINK_HOME>/lib/`, you also need to set the start option `jdbc.driver` to `com.oceanbase.jdbc.Driver`.

Setup OceanBase and LogProxy Server
----------------------

@@ -251,6 +263,20 @@ The OceanBase CDC Connector contains some options for both sql and stream api as
<td>String</td>
<td>Working mode of `obcdc` in LogProxy, can be `storage` or `memory`.</td>
</tr>
<tr>
<td>jdbc.driver</td>
<td>optional</td>
<td style="word-wrap: break-word;">com.mysql.jdbc.Driver</td>
<td>String</td>
<td>JDBC driver class for snapshot reading.</td>
</tr>
<tr>
<td>jdbc.properties.*</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Option to pass custom JDBC URL properties. User can pass custom properties like 'jdbc.properties.useSSL' = 'false'.</td>
</tr>
</tbody>
</table>
</div>
@@ -404,6 +430,7 @@ public class OceanBaseSourceExample {
.tableName("^test_table$")
.hostname("127.0.0.1")
.port(2881)
.jdbcDriver("com.mysql.jdbc.Driver")
.logProxyHost("127.0.0.1")
.logProxyPort(2983)
.serverTimeZone(serverTimeZone)
@@ -598,3 +625,100 @@ Data Type Mapping
</tbody>
</table>
</div>
### Oracle Mode
<div class="wy-table-responsive">
<table class="colwidths-auto docutils">
<thead>
<tr>
<th class="text-left">OceanBase type</th>
<th class="text-left">Flink SQL type</th>
<th class="text-left">NOTE</th>
</tr>
</thead>
<tbody>
<tr>
<td>NUMBER(1)</td>
<td>BOOLEAN</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 3 </td>
<td>TINYINT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 5 </td>
<td>SMALLINT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 10 </td>
<td>INT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s < 19 </td>
<td>BIGINT</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), 19 <=p - s <=38</td>
<td>DECIMAL(p - s, 0)</td>
<td></td>
</tr>
<tr>
<td>NUMBER(p, s > 0)</td>
<td>DECIMAL(p, s)</td>
</tr>
<tr>
<td>NUMBER(p, s <= 0), p - s> 38 </td>
<td>STRING</td>
<td></td>
</tr>
<tr>
<td>
FLOAT<br>
BINARY_FLOAT
</td>
<td>FLOAT</td>
<td></td>
</tr>
<tr>
<td>BINARY_DOUBLE</td>
<td>DOUBLE</td>
<td></td>
</tr>
<tr>
<td>
DATE<br>
TIMESTAMP [(p)]
</td>
<td>TIMESTAMP [(p)]</td>
<td></td>
</tr>
<tr>
<td>
CHAR(n)<br>
NCHAR(n)<br>
VARCHAR(n)<br>
VARCHAR2(n)<br>
NVARCHAR2(n)<br>
CLOB<br>
</td>
<td>STRING</td>
<td></td>
</tr>
<tr>
<td>
RAW<br>
BLOB<br>
ROWID
</td>
<td>BYTES</td>
<td></td>
</tr>
</tbody>
</table>
</div>
2 changes: 1 addition & 1 deletion .../connectors/mysql/table/JdbcUrlUtils.java → ...rica/cdc/debezium/utils/JdbcUrlUtils.java
View file
Original file line number Diff line number Diff line change
@@ -14,7 +14,7 @@
* limitations under the License.
*/

package com.ververica.cdc.connectors.mysql.table;
package com.ververica.cdc.debezium.utils;

import java.util.Map;
import java.util.Properties;
1 change: 1 addition & 0 deletions ...l-cdc/src/main/java/com/ververica/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
View file
Original file line number Diff line number Diff line change
@@ -29,6 +29,7 @@
import com.ververica.cdc.connectors.mysql.source.offset.BinlogOffset;
import com.ververica.cdc.connectors.mysql.source.offset.BinlogOffsetBuilder;
import com.ververica.cdc.debezium.table.DebeziumOptions;
import com.ververica.cdc.debezium.utils.JdbcUrlUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Loading

0 comments on commit edc376a

Please sign in to comment.