# 1. gds-kafka

数据实时接入缓冲方案工具

[toc]

## 1.1. 约束

1. Oracle源端不支持的数据类型有：CLOB,BLOB,NCLOB,INTERVAL_YEAR_TO_MONTH,INTERVAL_DAY_TO_SECOND,UROWID,BFILE,XML,RAW,LONG_RAW,LONG

2. 目标端建表时分布方式不可设置为 Replication 方式；需采用默认 HASH 方式，不要修改分布方式

3. Oracle 精简日志模式下，dws目标表必须包含主键，否则update操作会同步失败

4. Oracle 精简日志模式下，dws目标表不允许将除扩展字段以外的其他列定义为一个主键

5. 源端不允许在一个表中存在忽略大小写后相同的列名

6. 必须保证源数据中 op_ts 和 current_ts 数据格式与配置文件中相应的格式设置相匹配，否则无法入库

7. 在使用前，应在环境中安装国密依赖包。详情见1.7.1密码使用保护方法第1节前置条件。 

## 1.2. 使用建议

1. 若源表为无主键表，建议建目标表设为行存表并将源表的逻辑主键设为唯一索引

2. 源端为 MySQL 时，两个 DRS 任务同时从 MySQL 同步数据的情况下，op_ts 列的值是不同的，是 DRS 任务处理这条数据的时间戳

3. 使用cdc.drs.avro数据类型，源端为 MySQL 时，需要设置 avro.timestamp.format 参数

4. 源端为 Oracle 时，需要设置 app.min_log_fill_symbol 参数

5. 目前只有 Oracle 链路支持 DDL 同步功能。由于 DRS 服务提供的 DDL 功能不完备，MySQL 链路不支持 DDL 同步，参数文件中 app.support.ddl=true 参数不生效

## 1.3. 部署方法

此方法是常规部署方法

1. 准备 ECS 环境

    在云上申请 ECS，并安装 JRE 1.8 或更高版本

    【可选】创建普通用户，允许用户执行crontab命令

2. 将 gds-kafka.tar.gz 解压缩到 /opt/gds-kafka 目录

    ```
    mkdir /opt/gds-kafka
    tar zxvf gds-kafka.tar.gz -C /opt/gds-kafka
    ```

3. 准备实例目录

    ```
    mkdir /opt/gds-kafka/instance1
    ```

4. 将配置文件拷贝到实例目录并填写必需的选项

    ```
    cp /opt/gds-kafka/GDS-Kafka.properties /opt/gds-kafka/instance1/
    vi /opt/gds-kafka/instance1/GDS-Kafka.properties
    ```

    填写必须选项，其他选项可参考 /opt/gds-kafka/GDS-Kafka.properties.example 文件中的注释说明

5. 使用 watcher 启动工具并挂载到监控中

    具体步骤参考下节《watcher使用方法》

6. JVM 参数属于实例级，需要在启动 gds-kafka 程序前将参数写入实例目录下的 gds-kafka.vmoptions 文件中

## 1.4. watcher使用方法

watcher 是用于将 GDS-Kafka 主程序注册/清理 crontab 定时命令的脚本，包括注册、清理、监控、展示。

1. 启动实例1并将此实例注册进 crontab，其中 /opt/gds-kafka/instance1 目录下应提前放置 GDS-Kafka.properties 文件，并设置正确内容

    ```
    ./watcher.sh -a /opt/gds-kafka/instance1
    ```

2. 将实例1从 crontab 中删除
    
    ```
    ./watcher.sh -d /opt/gds-kafka/instance1
    ```

3. 展示当前受监控的实例
    
    ```
    ./watcher.sh -l
    ```

4. 检查实例是否在运行，如果不再运行则拉起（此命令一般不需要手动执行）
    
    ```
    ./watcher.sh -m /opt/gds-kafka/instance1
    ```

## 1.5. jar包使用方法

此方法用于测试功能

gds-kafka.jar 是 GDS-Kafka 主程序，配置文件为 GDS-Kafka.properties

1. 修改 GDS-Kafka.properties 文件，将源端和目标端的信息填入（可以参考 GDS-Kafka.properties.example 文件）

2. 启动程序
    
    ```
    java -jar gds-kafka.jar -INSTANCEPATH /opt/gds-kafka/instance1 2>&1 &
    ```

## 1.6. 日志相关

1. watcher 日志位于 watcher.sh 同级目录下，名为 watcher.log

2. gds-kafka 工具日志默认位于每个实例目录下的 logs/GDS-Kafka.log 中，可以通过配置文件中的 app.log_file 选项进行指定

3. 每个日志文件为16MB，单个实例最多保存100个日志文件

## 1.7. DWS 密码保护

在程序启动前，需通过命令行参数或者配置文件配置DWS登录密码，使用明文存在密码泄漏风险。

### 1.7.1. 密码保护使用方法

1. 前置条件

    需要在linux操作系统下配置支持国密的环境。
    方法1 在系统中以rpm或yum方式安装seccomponent依赖包
   
    方法2 手动创建解压so与keytab文件目录：
          rm -rf ${解压so路径}                   // 防止文件夹下有文件残留
          mkdir -p ${解压so路径}                 // 创建空目录，并保证该目录权限为可读写
          export SC_LIBRARY_PATH=${解压so路径}                        // 将该空目录加载至环境变量                 
          export LD_LIBRARY_PATH=${解压so路径}:$LD_LIBRARY_PATH

2. 生成密文

    运行以下命令，根据提示，键盘填写明文密码，可以得到密文字符串。
    当不填写-ENCRYPT_TYPE参数时，采用默认加密方式generalCipher，使用ALG_AES256_CBC算法, SMcompatible使用ALG_SM4_CBC算法
    
    ```shell
    java -cp gds-kafka.jar com.huawei.util.StringCrypter -ENCRYPT_TYPE generalCipher/SMcompatible/WCC
    ```
   
    或，直接以接口形式调用该命令，获取json形式的加密结果：

    ```shell
    java -cp gds-kafka.jar com.huawei.util.StringCrypter -ENCRYPT_TYPE generalCipher/SMcompatible/WCC -PWD yourpassword
    ```
    得到输出：{"encrypt_result":"encrypted_password","encrypt_success":true}

3. 填写配置文件
    GDS-Kafka.properties 中，需要指定:
    
    ```
    dws.pwd.encrypt.type=SMcompatible
    dws.password=上一步骤生成的密文
    ```

## 1.8. 参数优先级

命令行参数优先级高于配置文件参数优先级

## 1.9. DWS 目标表扩展字段

目标表中，扩展字段用于记录同步关键信息。

1. 若目标表已存在，可以使用 add_expand_field.sql 脚本增加扩展字段。该工具建立出的扩展字段 op_ts 和 current_ts 类型为 text。
2. 若目标表需要新建，可以使用自动建表工具创建目标表。建立出的目标表带有的扩展字段 op_ts 和 current_ts 类型，可以通过其可选参数，配置为 text 或 bigint。

### 1.9.1. 扩展字段类型及作用

|   扩展字段   |      类型       |           作用           |
| ----------- | -------------- | ----------------------- |
| source      | text           | 源端表名（schema.table)  |
| op_type     | text           | 源端操作（I/U/D)         |
| op_ts       | text 或 bigint | 源端操作时间戳（毫秒）     |
| current_ts  | text 或 bigint | 目标端入库时间戳（毫秒）   |
| op_position | text           | 操作顺序（before/after） |
| pos         | bigint         | 数据在Kafka中的offset    |
| delflag     | text           | 数据删除标记位            |

> op_ts 和 current_ts 字段默认为 text 类型。

### 1.9.2. 扩展字段时间戳数据格式配置

对于 op_ts 和 current_ts 字段，由于输入的数据格式不同，需要针对输入数据指定具体的时间戳格式。
配置参数分别为: app.extend.field.op_ts.format 和 app.extend.field.curr_ts.format

1. 输入时间戳为整数格式（毫秒），可以不配置或配置为空
2. 输入时间戳为时间格式字符串（例：2021-01-01 08:00:00.123），需要按照Java时间格式化配置参数（yyyy-MM-dd HH:mm:ss.SSS)

### 1.9.3. DWS 目标表添加扩展字段

如果要将已经存在在 DWS 中的表作为目标表，可以使用如下方法添加扩展字段：

1. 连接到 DWS 数据库，并进入目标数据库
2. 执行 add_expand_field.sql 脚本中的 SQL 语句
3. 调用 add_expand_field 函数添加扩展字段
   函数原型为： add_expand_field(SCHEMA_NAME, TABLE_NAME, WithDelFlag)
   参数说明：  SCHEMA_NAME schema名称，TEXT类型，例：'schema1'
   TABLE_NAME  表名，TEXT类型，例：'table1'
   WithDelFlag 是否扩展 delflag 字段，BOOLEAN类型，例：TRUE
   
5. 举例：
    1. 为 schema1.table1 添加扩展字段，包含 delflag
        ```sql
        call add_expand_field('schema1', 'table1', TRUE);
        ```
    2. 为 schema1.table1 添加扩展字段，不包含 delflag
        ```sql
        call add_expand_field('schema1', 'table1', FALSE);
        ```
    3. 为 schema1 下的所有表添加扩展字段，包含 delflag
        ```sql
        call add_expand_field('schema1', null, TRUE);
        ```
    4. 在添加完毕后可以将函数删除：
        ```sql
        DROP FUNCTION add_expand_field( IN schemaName TEXT, IN tableName TEXT, IN delFlag BOOLEAN );
        DROP FUNCTION add_expand_field_operate( IN schemaName TEXT, IN tableName TEXT, IN delFlag BOOLEAN );
        ```

## 1.10. 表字段映射关系表

1. 表字段映射关系表保存在映射配置文件中，用于指定源端表和目标表的对应关系配置，以及过滤可处理的表数据。
2. 配置文件中参数为 app.table_mapping，默认为实例目录下的 table_mapping.txt 文件。

### 1.10.1. 表过滤信息

1. 文件为空，则不过滤任何表
2. 如果配置了白名单，未配置黑名单，则只处理白名单中的表
3. 如果配置了黑名单，未配置白名单，则不处理黑名单中的表
4. 若既配置了白名单又配置了黑名单，则处理在白名单中且不在黑名单中的表

### 1.10.2. 字段映射信息

1. 如果目标表和源端表字段信息不一致，则需要通过一一对应来设置：`orc_table1.col1=dws_table1.coll1`。注意，列名都应该是小写格式。
2. 若目标表和源端表字段信息完全一致，可通过 `orc_table1.*=dws_table1.*` 来进行配置
3. 黑名单只需要指定源端表信息
4. 目标表包含的扩展字段不需要在映射表中体现
5. 自动建表工具建立的表名、列名都是小写格式。

### 1.10.3. 映射表举例

如下所示：
1. 白名单共3个表。前两个表都是指定了只同步两列数据，第三个表指定同步所有的列
2. 黑名单共1个表。即使数据存在在 Kafka 中也不会被同步到 DWS 中
    
    ```
    /*whitelist*/
    # orc_schema1.orc_table1 => dws_schema1.dws_table1
    dws_table1.name=orc_table1.orc_name
    dws_table1.age=orc_table1.orc_age
    # orc_schema2.orc_table2 => dws_schema2.dws_table2
    dws_table2.name=orc_table2.orc_name
    dws_table2.age=orc_table2.orc_age
    # orc_schema3.orc_table3 => dws_schema3.dws_table3
    dws_table3.*=orc_table3.*
    /*blacklist*/
    orc_schema4.orc_table4
    ```

### 1.10.4. 目标端自动建表并生成 tablemapping 文件

1. 通过如下命令向目标端 DWS 自动创建 schema 和表并生成建表脚本文件和 table_mapping 文件。
2. table mapping 文件会自动生成到 gds-kafka.jar 文件所在的目录下，名称为 *table_mapping.txt-时间戳* 。
3. 在不指定 *-INSTANCEPATH* 参数与 *-SQL_FILE* 参数的情况下，table mapping 文件与建表脚本文件同级，
   如果指定 *-INSTANCEPATH* 参数，则以jar包下*-INSTANCEPATH* 参数指定的文件夹为父目录。
   默认名称为 *table_mapping-时间戳.txt-create_table.sql*，
   该文件路径及名称可以通过配置TABLE_MAPPING_FILE_NAME参数来指定。
   这个脚本可以直接在 DWS 上执行，方便源端数据库和 DWS 之间网络无法联通的情况下进行建表操作。
4. 如果指定 *-MAPPING_FILE_ONLY* 参数，则指挥创建建表脚本文件和 table mapping 文件，不会在 DWS 端真正创建 schema 和表结构。

```shell script
java -cp gds-kafka.jar com.huawei.util.createtable.CreateTableApp [参数列表]
```

#### 1.10.4.1. 必选参数
| 参数项  |                                                         描述                                                         |
| ------ | -------------------------------------------------------------------------------------------------------------------- |
| -SRC_URL   | 源端数据库连接URL，例如 100.85.116.150:1539 或 100.93.3.72:3306 |
| -SRC_TYPE  | 源端数据库类型，应填写oracle或mysql
| -USR   | 源端数据库用户名                                                                                                       |
| -PWD   | 源端数据库密码                                                                                                         |
| -SRC_D | 源端数据库名称                                                                                                         |
| -SRC_S | 源端数据库schema名称，支持配置单个或者多个，多个时用冒号隔开。                              |
| -DWS_H | DWS集群地址IP端口号，格式是IP:port，例如 10.185.178.129:48695                                                           |
| -DWS_U | DWS数据库的用户名                                                                                                      |
| -DWS_P | DWS数据库的密码                                                                                                       |
| -DWS_D | DWS数据库名称                                                                                                         |

>注：参数中包含特殊字符（如叹号!)，可以使用单引号('')将参数包围起来。

#### 1.10.4.2. 可选参数
|            参数项            |                                                                                                       描述                                                                                                       |
| --------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| -SRC_T                      | 源端数据库表名称，可以是单表或者多表，多表通过冒号分隔各表所属各schema，以逗号分隔同一schema下的不同table，如果配置此参数，一定要配置SRC_S参数，例如： test,test2:test3。  如果存在TABLE_MAPPING_INPUT参数，则此参数以T\F表示是否同步该schema下全部table,例如：T:F:T。该参数中冒号的数目应与SRC_S参数中的冒号数目相同。|
| -DWS_S                      | DWS数据库下schema名称，当配置了DWS_S名称且未配置白名单时，建表会在此schema下，当配置无白名单且未配置DWS_S时，在DWS数据库会建立和ORA_S同名的schema，建表到该schema下面，有白名单且无论是否配置DWS_S时，则建立到白名单中的schema下。 |
| -BLACK_PATH                 | 黑名单路径，定义了哪些schema下的哪些表不会被创建到DWS数据库中                                                                                                                                                         |
| -WHITE_PATH                 | 白名单路径，指定了哪些schema下的哪些表会被创建到DWS数据库中，如果配置了白名单，则以白名单为准，注意白名单和黑名单不能同时配置。                                                                                              |
| -PRE                        | DWS表名的前缀，默认在dws创建的表名称和源端表名称一样，如果要修改dws表名称，可以加上PRE参数指定dws表的名称前缀，一般用于批量建表时修改表名称。                                                                                 |
| -STORED_TYPE                | 批量建表默认存储格式为row(行存)，可通过STORED_TYPE参数指定存储类型，支持row(行存)和column(列存)两种。                                                                                                                   |
| -FLAG                       | 自定义扩展字段删除标志位名称，如果未设置，则不创建删除标记位字段。                                                                                                                                                     |
| -IS_DROP_TABLE              | 建表前是否先进行删表操作                                                                                                                                                                                           |
| -EXACT_COL_INFO             | table mapping 文件中是否详细记录列名                                                                                                                                                                               |
| -SQL_FILE                   | DWS数据库建表脚本文件路径                                                                                                                                                                                          |
| -MAPPING_FILE_ONLY          | 只创建 table mapping 文件和建表脚本，不在 DWS 端创建表结构                                                                                                                                                          |
| -SYNC_FIELD_NOT_NULL        | 是否同步列上的NOT NULL属性，默认为不同步                                                                                                                                                                            |
| -SYNC_FIELD_DEFAULTS        | 是否同步列上的DEFAULTS默认值属性，默认为不同步                                                                                                                                                                       |
| -CREATE_NO_PRIMARY_TABLE    | 如果源端表没有主键和唯一索引，是否在目标端建表                                                                                                                                                                        |
| -EXTEND_FIELD_OPTS_TYPE      | 扩展字段op_ts数据类型，默认为text                                                                                                                                                                                  |
| -EXTEND_FIELD_CURRENTTS_TYPE | 扩展字段current_ts数据类型，默认为text                                                                                                                                                                             |
| -TABLE_MAPPING_FILE_NAME     | 配置table-mapping文件的名称，默认为table-mapping-时间戳.txt                                                                                                                                                        |
| -INSTANCEPATH                | 实例目录，如果该目录不存在，则新建该目录。如果指定该参数，table-mapping以此文件夹为父目录。                                                                                                                                 |
| -JSON_OUTPUT_PATH            | 建表工具以json格式输出脚本操作结果，此参数配置脚本输出的路径和文件名。默认文件名为output.json，路径以执行脚本所在位置为父目录。                                                                                                |
| -TABLE_MAPPING_INPUT         | 建表工具以映射表形式获取哪些schema、table将被创建的信息。此项配置了该映射表读取的本地位置。如果配置了此项，则以此项与SRC_S、SRC_T配置作为同步哪些表的依据，并屏蔽黑名单中的表。如果未设置，则不使用此方式创建表，先检查是否以白名单方式创建表，如果不是则以常规方式确定哪些表将被创建。|
| -SOURCE_PWD_ENCRYPT_TYPE     | 源端密码的加密方式，支持选项generalCipher/SMcompatible/WCC/cleartext, cleartext表示不加密，默认值为cleartext|
| -DWS_PWD_ENCRYPT_TYPE        | 目标端密码的加密方式，支持选项generalCipher/SMcompatible/WCC/cleartext, cleartext表示不加密，默认值为cleartext|

#### 1.10.4.3. 白名单文件举例
```
ORCL-DWS_GHD:test3,TEST4;ORCL-DWS_GHD:DML1,DML2;ORCL-DRS:STRESSTESTTABLE,BLACKTABLE2;
```

#### 1.10.4.4. 黑名单举例
```
ORCL-DRS:BLACKTABLE1;
```

#### 1.10.4.5. 例外
1. 对于 timestamp 类型，存在无法识别默认值而导致建表失败的情况，需要手动建表。

例：

Oracle端：

```sql
CREATE TABLE DWS_GHD.TAB1 (
	COL_TIMESTAMP TIMESTAMP DEFAULT '10-8月-2021'
);
```

由于DWS端无法处理Oracle端的日期默认值，所以在建表时会报错：

```
ERROR: invalid input syntax for type timestamp: "10-8月-2021".
```

这种情况下，需要手动建表，或者通过 CreateTableApp 获取SQL语句，修改后再在 DWS 端执行。

## 1.11. 维护类工具
### 1.11.1. ONLY_CONSUME模式下消费并解析kafka中数据

此工具用于消费 Kafka 中的数据并解析为可读信息后保存在本地文件中。文件位于参数指定的 INSTANCEPATH 目录下的 .data 后缀文件中。

```shell script
java -jar gds-kafka.jar [参数列表]
```

|     参数项     |                          描述                          |
| ------------- | ------------------------------------------------------ |
| -INSTANCEPATH | 实例目录，此实例目录下需要有 GDS-Kafka.properties 配置文件 |
| -APP_MODE     | 当前只能填写 ONLY_CONSUME                               |
| -MAINTAIN_TOPIC        | Kafka Topic 名称                                       |
| -MAINTAIN_PARTITION    | Kafka 分区号                                            |
| -MAINTAIN_OFFSET_BEGIN | [可选]从此offset开始消费，默认从 beginning 开始消费        |
| -MAINTAIN_OFFSET_END   | [可选]消费到此位置为止，默认为消费到最后位置                |

### 1.11.2. 连通性测试

此工具用于对kafka端、目标数据库dws端、建表源数据库mysql端、建表源数据库oracle端、工业IOT端进行连通性测试。

对kafka端，支持以PLAINTEXT协议连接；以SASL_PLAINTEXT协议、PLAIN加密方式进行连接（通过用户名密码验证）。

```shell script
java -cp gds-kafka.jar com.huawei.util.checkconnect.CheckConnect [参数列表]
```

|     参数项     |                          描述                          |
| -------------- | ------------------------------------------------------ |
| -ST           | [必选] 测试类型，支持mysql/oracle/dws/kafka/iot        |
| -H            | [mysql/oracle/dws/iot必选] 数据库IP地址                       |
| -P            | [mysql/oracle/dws必选, iot可选] 数据库端口号                       |
| -U            | [mysql/oracle/dws/iot必选，kafka可选] 用户名，当kafka填写此项，将测试以SASL_PLAINTEXT协议、PLAIN加密方式进行认证|
| -PWD          | [mysql/oracle/dws/iot必选，kafka可选] 密码                   |
| -ENCRYPT_TYPE | [可选] 密码的加密方式，支持选项generalCipher/SMcompatible/WCC/cleartext, cleartext表示不加密，默认值为cleartext|
| -D            | [mysql/oracle/dws必选] 数据库名称                        |
| -KS           | [kafka必选] kafka broker连接地址，以逗号相隔，例如：192.168.0.1:9092,192.168.0.2:9092|
| -TIMEOUT      | [可选] 超时报错时间限制，默认值为2秒                        |
