# dws-client #### 简介 dws-client是一款基于DWS JDBC实现的高性能、便捷入库工具，用户在使用时必须保证JDBC可以连接。其中使用dws-client入库具备如下优势： 1. dws-client提供对缓存的空间、时间维度的限制，支持攒批提升入库性能，以满足业务在高峰低谷期的入库实时性。 ![](https://support.huaweicloud.com/tg-dws/public_sys-resources/note_3.0-zh-cn.png) 攒批：在实时性要求不严格的场景，对单条数据操作进行缓存，待缓存至多条后批量操作，以提升写入性能。 2. 支持并发入库。 3. 内部实现多种高性能入库方式、主键冲突策略，满足各种场景入库需求。 4. API方式交互，低门槛使用。 #### 引入依赖 dws-client已经发布至maven仓库，可在仓库中选择最新版本使用，链接请参见：。 ``` com.huaweicloud.dws dws-client ${version} ``` #### 核心功能整体主要以2.x版本为例，说明为兼容方式时表示1.x版本的用法。 **初始化client** 初始化client用于创建一个client实例，以便后续的入库等操作使用。 dws-client支持参数全部在com.huaweicloud.dws.client.config.DwsClientConfigs中记录，其中每一个常量ConfigOp均表示一个参数，底层最终采用map存储，其中ConfigOp中的key表示存储时的key，如果采用配置文件方式也是配置文件的key。 1. 以下举例为最简单的方式，只需要数据库的连接配置即可，其余配置均采用默认配置。 ``` public DwsClient getClient() throws Exception { DwsConfig config = DwsConfig.of() .with(DwsClientConfigs.JDBC_URL, System.getenv("db_url")) .with(DwsClientConfigs.JDBC_PASSWORD, System.getenv("db_pwd")) .with(DwsClientConfigs.JDBC_USERNAME, System.getenv("db_username")); return new DwsClient(config); } ``` 2. 使用配置文件新建配置文件client.properties ``` dws.client.jdbc.url=jdbc:gaussdb://xxxx:8000/gaussdb dws.client.jdbc.password=**** dws.client.jdbc.username=dbadmin ``` 配置文件初始化 ``` public DwsClient getClientByProperties() throws Exception { URL resource = this.getClass().getClassLoader().getResource("client.properties"); DwsConfig config = new DwsConfig(resource.getFile()); return new DwsClient(config); } ``` 3. 通过map参数 ``` public DwsClient getClientByMap() throws Exception { Map config = new HashMap<>(); config.put(DwsClientConfigs.JDBC_URL.key(), System.getenv("db_url")); config.put(DwsClientConfigs.JDBC_PASSWORD.key(), System.getenv("db_pwd")); config.put(DwsClientConfigs.JDBC_USERNAME.key(), System.getenv("db_username")); return new DwsClient(new DwsConfig(config)); } ``` 4. 兼容1.x版本方式 ``` public DwsClient getClient(){ DwsConfig config = DwsConfig .builder() .withUrl("jdbc:gaussdb://***/gaussdb") .withUsername("***") .withPassword("****") .build(); return new DwsClient(config); } ``` 5. 配置表级参数 ``` 当client入库多张表时，需要设置不同表的参数配置时可通过表级参数配置，设置时可通过对全局配置withTable("xxx")获取到表级参数构造器，此时会根据此刻的全局参数初始化表级参数，新设置的将覆盖原有参数，build时完成设置并添加到全局参数中，此时接口返回的是全局参数，如下可实现链式调用。 public DwsClient getClientTable() throws Exception { DwsConfig config = DwsConfig.of() .with(DwsClientConfigs.JDBC_URL, System.getenv("db_url")) .with(DwsClientConfigs.JDBC_PASSWORD, System.getenv("db_pwd")) .with(DwsClientConfigs.JDBC_USERNAME, System.getenv("db_username")) .with(DwsClientConfigs.WRITE_AUTO_FLUSH_BATCH_SIZE, 10000) .withTable("test") .with(DwsClientConfigs.WRITE_CONFLICT_STRATEGY, ConflictStrategy.INSERT_OR_IGNORE) .build() .withTable("test1") .with(DwsClientConfigs.WRITE_AUTO_FLUSH_BATCH_SIZE, 200) .build(); return new DwsClient(config); } ``` 6. 1.x兼容 ``` return DwsConfig.builder() .withUrl(System.getenv("db_url")) .withPassword(System.getenv("db_pwd")) .withUsername(System.getenv("db_username")) .withAutoFlushBatchSize(1000) // 默认攒批1000 .withTableConfig("test.t_c_batch_size_2", new TableConfig() .withAutoFlushBatchSize(500)); // 对于表test.t_c_batch_size_2 攒批500; ``` **使用数据库连接执行SQL** 该接口主要用于一些特殊业务，在目前支持的功能中无法满足时使用。例如：数据查询，可以直接使用原生JDBC连接操作数据库。 API参数为一个函数式接口，接口会提供一个数据库连接，返回值可以是任意类型，由业务返回类型决定。 ``` public void sql() throws DwsClientException { Integer id = getClient().sql(connection -> { try (ResultSet resultSet = connection.createStatement().executeQuery("select id from test.user where name = 'zhangsan'")) { if (resultSet.next()) { return resultSet.getInt("id"); } } return null; }); System.out.println("zhangsan id = " + id); } ``` **获取表信息** API可以根据一个带schema的表名获取到表结构（会存在缓存），表结构的定义包括：所有列、主键。 ``` public void getTableSchema() throws DwsClientException { TableSchema tableSchema = getClient().getTableSchema(TableName.valueOf("test.test")); } ``` **数据入库** 数据入库操作由client提供的write API承载，接口会返回一个Operate接口，可以通过接口操作表字段，提交代表操作完成，交由client执行入库，提交时可以选择同步或者异步，设置字段时可以选择是否在主键冲突时忽略该设置。 ``` public void write() throws DwsClientException { getClient().write("test.test") .setObject("id", 1) .setObject("name", "test") // 只有在insert时会生效该设置，主键冲突时不更新 .setObject("age", 38, true) // 异步入库，放入后台缓存即可返回 //.commit() // 同步入库会等待入库成功返回 .syncCommit(); } ``` 对于入库时表结构不会发生变化的场景，通过全局schema减少数据库元数据的查询，以减少DWS的压力并提升入库性能。 ``` public void testWrite() throws Exception { try (DwsClient client = getClient()) { client.sql((conn) -> { conn.createStatement().execute("DROP Table IF EXISTS test.dws_client_test;" + "create table test.dws_client_test (id integer, name varchar(10), age int);"); return null; }); TableSchema tableSchema = client.getTableSchema(TableName.valueOf("test.dws_client_test")); log.info("table schema {}", tableSchema); for (int i = 0; i < 100; i++) { Operate operate = client.write(tableSchema) .setObject("id", i) .setObject("name", "name_" + i) .setObject("age", i); operate.commit(); } } } ``` 通过字段索引写client，减少hash计算以缓解客户端cpu压力提升写client吞吐。字段的下标即为数据库字段的顺序，也是tableSchema.getColumns()集合中的顺序。 ``` public void testWrite() throws Exception { try (DwsClient client = getClient()) { client.sql((conn) -> { conn.createStatement().execute("DROP Table IF EXISTS test.dws_client_test;" + "create table test.dws_client_test (id integer, name varchar(10), age int);"); return null; }); TableSchema tableSchema = client.getTableSchema(TableName.valueOf("test.dws_client_test")); log.info("table schema {}", tableSchema); for (int i = 0; i < 100; i++) { Operate operate = client.write(tableSchema) .setObject(0, i) .setObject(1, "name_" + i) .setObject(2, i); operate.commit(); } } } ``` **数据删除** 删除API和入库API都由Operate承载，但删除时只能且必须设置主键字段，同时忽略更新字段不生效。 ``` public void delete() throws DwsClientException { getClient().delete("test.test") .setObject("id", 1) // 异步入库，放入后台缓存即可返回 //.commit() // 同步入库会等待入库成功返回 .syncCommit(); } ``` **强制刷新缓存到数据库** ``` public void flush() throws DwsClientException { getClient().flush(); } ``` **关闭资源** close时会将缓存刷库，同时close后无法再次执行入库、删除、sql等API接口。 ``` public void close() throws IOException { getClient().close(); } ``` #### 监听数据入库成功的事件在异步入库的场景，想要知道哪些数据已经入库，则可通过绑定flushSuccess函数接口实现，该接口会在数据库事务提交完成后回调，回调时会将该批入库数据传递给接口。 ``` public DwsClient getClient() throws Exception { DwsConfig config = DwsConfig.of() .with(DwsClientConfigs.JDBC_URL, System.getenv("db_url")) .with(DwsClientConfigs.JDBC_PASSWORD, System.getenv("db_pwd")) .with(DwsClientConfigs.JDBC_USERNAME, System.getenv("db_username")) .onFlushSuccess(records -> { for (Record record : records) { log.info("flush success. value = {}, pk = {}", RecordUtil.toMap(record), RecordUtil.getRecordPrimaryKeyValue(record)); } }); return new DwsClient(config); } ``` 兼容1.x ``` public DwsClient getClient() { DwsConfig config = DwsConfig .builder() .withUrl("jdbc:postgresql://***/gaussdb") .withUsername("***") .withPassword("****") .onFlushSuccess(records -> { for (Record record : records) { log.info("flush success. value = {}, pk = {}", RecordUtil.toMap(record), RecordUtil.getRecordPrimaryKeyValue(record)); } }) .build(); return new DwsClient(config); } ``` #### 监听后台异常任务异步入库数据会由后台任务执行，后台任务执行失败想要感知可通过绑定error函数接口实现，否则只能在下次提交时发生异常错误，由业务感知，如果绑定接口不上报异常，那么该异常将忽略，不会在下次提交时报出，否则会在下次提交时以接口异常的形式报出给业务。 ``` public DwsClient getClient() throws Exception { DwsConfig config = DwsConfig.of() .with(DwsClientConfigs.JDBC_URL, System.getenv("db_url")) .with(DwsClientConfigs.JDBC_PASSWORD, System.getenv("db_pwd")) .with(DwsClientConfigs.JDBC_USERNAME, System.getenv("db_username")) .onError((clientException, client) -> { if (clientException instanceof DwsClientRecordException) { DwsClientRecordException recordException = (DwsClientRecordException) clientException; List records = recordException.getRecords(); List exceptions = recordException.getExceptions(); for (int i = 0; i < records.size(); i++) { log.error("pk = {} . error = {}", RecordUtil.getRecordPrimaryKeyValue(records.get(i)), exceptions.get(i)); } } if (clientException.getCode() != ExceptionCode.CONNECTION_ERROR && clientException.getCode() != ExceptionCode.LOCK_ERROR) { throw clientException; } log.error("code = {}", clientException.getCode(), clientException.getOriginal()); return null; }); return new DwsClient(config); } ``` 兼容1.x ``` public DwsClient getClient() { DwsConfig config = DwsConfig .builder() .withUrl("jdbc:postgresql://***/gaussdb") .withUsername("***") .withPassword("****") .onError((clientException, client) -> { if (clientException instanceof DwsClientRecordException) { DwsClientRecordException recordException = (DwsClientRecordException) clientException; List records = recordException.getRecords(); List exceptions = recordException.getExceptions(); for (int i = 0; i < records.size(); i++) { log.error("pk = {} . error = {}", RecordUtil.getRecordPrimaryKeyValue(records.get(i)), exceptions.get(i)); } } if (clientException.getCode() != ExceptionCode.CONNECTION_ERROR && clientException.getCode() != ExceptionCode.LOCK_ERROR) { throw clientException; } log.error("code = {}", clientException.getCode(), clientException.getOriginal()); return null; }) .build(); return new DwsClient(config); } ``` #### 异常处理异常可分为三类： 1. InvalidException运行时异常不显示抛出，触发在请求参数无效时。 2. DwsClientException对所有异常的封装，包含被解析的code以及原始异常。 3. DwsClientRecordException对DwsClientException的扩展，包含写入异常的数据集合以及对应DwsClientException异常。异常code对照： ``` public enum ExceptionCode { /** * 无效参数 */ INVALID_CONFIG(1), /** * 连接异常 */ CONNECTION_ERROR(100), /** * 只读 */ READ_ONLY(101), /** * 超时 */ TIMEOUT(102), /** * 连接数过多 */ TOO_MANY_CONNECTIONS(103), /** * 加锁异常 */ LOCK_ERROR(104), /** * 认证失败 */ AUTH_FAIL(201), /** * 已经关闭 */ ALREADY_CLOSE(202), /** * 无权限 */ PERMISSION_DENY(203), SYNTAX_ERROR(204), /** * 内部异常 */ INTERNAL_ERROR(205), /** * 中断异常 */ INTERRUPTED(206), /** * 表未发现 */ TABLE_NOT_FOUND(207), CONSTRAINT_VIOLATION(208), DATA_TYPE_ERROR(209), DATA_VALUE_ERROR(210), /** * 解析不到的异常 */ UNKNOWN_ERROR(500); private final int code; } ``` #### 详细配置说明列表只包含公开使用的参数，部分参数为内部运行需要，列表中不包含的参数不建议自行配置。在使用配置文件方式配置时通过key配置参数，value为时间单位，支持：天：d、day(s) 小时：h、hour(s) 分钟：min(s)、m、minute(s) 秒：s、sec(s)、second(s) 毫秒：ms、milli(s)、millisecond(s) 内存参数： byte： b kb：k、kb mb： m、mb gb：g、gb | 参数 | key | 1.x参数 | 说明 | 默认值 | |:---|:---|:---|:---|:---| | JDBC_URL | dws.client.jdbc.url | url | dws数据库JDBC连接地址，client强制使用jdbc:gaussdb://\*\*\*,配置 jdbc:postgresql也会被替换为jdbc:gaussdb，因此使用上两者均可配置且功能一致。 | - | | JDBC_USERNAME | dws.client.jdbc.username | username | dws数据库用户名。 | - | | JDBC_PASSWORD | dws.client.jdbc.password | password | dws数据库用户密码。 | - | | JDBC_DRIVER | dws.client.jdbc.driver-name | -- | dws数据库驱动。 | com.huawei.gauss200.jdbc.Driver | | JDBC_IS_DN | dws.client.jdbc.dn | -- | JDBC是否为DN连接，不需要主动配置，程序中会自动检查。 | false | | JDBC_CONNECTION_CHECK_TIMEOUT | dws.client.jdbc.check-timeout | -- | 连接检查超时时间。 | 60s | | JDBC_CONNECTION_MAX_USE_TIME | dws.client.jdbc.max.use-time | connectionMaxUseTimeSeconds（**已废弃**） connectionMaxUseTimeThreshold | 连接最大使用时间，超过该时间会强制关闭当前连接并重新获取；使用COPY_MERGE/COPY_UPSERT，并且有主键时会使用临时表，临时表的schema在连接断开时才会清除，主要用于清除该部分数据。新参数connectionMaxUseTimeThreshold从**2.1.0.2**版本开始支持。 | 3600 | | JDBC_CONNECTION_MAX_IDLE | dws.client.jdbc.max.idle | connectionMaxIdleMs | 连接最大空闲时间：当无数据时连接一直处于空闲状态，当达到该值时连接被释放 | 60s | | CLIENT_NAME | dws.client.name | -- | dws-client名称。 | dwsClient | | WRITE_PARTITION_POLICY | dws.client.write.partition-policy | -- | partition缓存（表下面的多份缓存）分区策略： DYNAMIC：动态调整，根据资源使用情况动态调整partition数量在WRITE_PARTITION_MIN和WRITE_PARTITION_MAX之间（当前未实现动态调整仅按WRITE_PARTITION_MIN初始化） DN：按照dws集群DN分布策略分布，分布后每一份缓存均在同一个DN中，入库时也会使用内部协议直接将数据通过DN入库，当前策略需要集群额外配置，只有在指导下配置使用，同时表的分布列仅支持int\*、text类字段作为分布列，切勿自行使用。 | DYNAMIC | | CACHE_TABLE_METADATA | dws.client.cache.table-metadata | metadataCacheSeconds | 表元数据缓存时间，为提升性能，会对理论上不怎么变更的数据，例如表结构，该参数用于设置缓存过期时间；如果系统不考虑支持在线表结构变更，无需配置该参数，以减少对dws集群系统表查询导致的压力，小于等于0表示永不过期。 | -1 1.x为180s | | RETRY_SLEEP_BASE_TIME | dws.client.retry.sleep-base-time | retryBaseTime | 重试时sleep时间 = RETRY_SLEEP_BASE_TIME \* 次数 + （0\~RETRY_SLEEP_RANDOM_TIME），该参数设置时间基数。 | 1000ms | | RETRY_SLEEP_RANDOM_TIME | dws.client.retry.sleep-random-time | retryRandomTime | 重试时sleep时间 = RETRY_SLEEP_BASE_TIME \* 次数 +（0\~RETRY_SLEEP_RANDOM_TIME，该参数设置重试时的随机数范围，该参数主要用于在死锁场景将两个task执行时间错开。 | 300ms | | RETRY_MAX_TIMES | dws.client.retry.max-times | maxFlushRetryTimes | 执行刷库任务时，最大尝试执行次数。 | 3 | | WRITE_AUTO_FLUSH_BATCH_SIZE | dws.client.write.auto-flush-size | autoFlushBatchSize | 后台任务刷库策略：缓存条数大于等于WRITE_AUTO_FLUSH_BATCH_SIZE或者当前时间 - 缓存开始时间大于等于WRITE_AUTO_FLUSH_MAX_INTERVAL，该参数配置缓存最大条数。 | 30000 | | WRITE_FORCE_FLUSH_BATCH_SIZE | dws.client.write.force-flush-size | -- | 强制刷库数据条数，为提高整体吞吐，当业务线程buffer中提交数据时buffer中数据符合自动刷库条件，但是未达到当前参数值，则业务线程依旧写缓存，期望由后台线程负责提交入库以减少业务线程阻塞，若达到当前值则业务线程会将缓存提交执行入库才可返回，此时如果线程池无空闲资源则需等到有空闲资源方可提交成功，通常该参数设置比WRITE_AUTO_FLUSH_BATCH_SIZE略大比较合理。 | 40000 | | WRITE_FORCE_FLUSH_UPSERT_BATCH_SIZE | dws.client.write.force-flush-upsert-size | -- | 当writeMode使用upsert时，修改使用upsert入库模式下最大攒批大小WRITE_FORCE_FLUSH_BATCH_SIZE。 | 10000 | | WRITE_AUTO_FLUSH_MAX_INTERVAL | dws.client.write.auto-flush-max-interval | autoFlushMaxIntervalMs | 后台任务刷库策略：缓存条数大于等于WRITE_AUTO_FLUSH_BATCH_SIZE或者当前时间 - 缓存开始时间大于等于WRITE_AUTO_FLUSH_MAX_INTERVAL，该参数配置缓存最大时间。 | 3s | | WRITE_FIXED_COPY_CACHE_SIZE | dws.client.write.fixed-copy.cache-size | -- | 当使用fixed copy模式时，用于流式写入的缓冲队列大小。 | 1000 | | WRITE_USE_COPY_BATCH_SIZE | dws.client.write.use-copy-size | copyWriteBatchSize | 在writeMode设置为AUTO时，在数据量低于copyWriteBatchSize时会使用upsert方式入库，否则根据是否有主键选择copy/copy+ upsert方式入库。 | 1000（2.1.0.1版本及以上修改为1000，此前为5000） | | WRITE_BUFFER_ALL_MAX_BYTES | dws.client.write.buffer.all-max-bytes | -- | 当整个client实例中攒批数据内存大小达到该参数设置大小时为避免OOM会执行强制刷库，该参数如果不配置会根据JVM最大堆大小设置，大小为JVM最大内存 \* WRITE_BUFFER_JVM_PROCESSORS。说明：攒批时根据对象类型预估，并非精确值。 | 1G | | WRITE_BUFFER_TABLE_MAX_BYTES | dws.client.write.buffer.table-max-bytes | -- | 当一张表最大攒批大小达到该值时为避免oom会将整个表缓存刷库，默认取值WRITE_BUFFER_ALL_MAX_BYTES，如果涉及入库多张表建议合理设置此参数。 | 500m | | WRITE_BUFFER_PARTITION_MAX_BYTES | dws.client.write.buffer.partition-max-bytes | -- | 每个partition攒批大小达到该值为避免OOM，会强制将整个partition缓存刷库，默认初始化为WRITE_BUFFER_TABLE_MAX_BYTES/WRITE_PARTITION_MIN。 | 200m | | WRITE_BUFFER_JVM_PROCESSORS | dws.client.write.buffer.jvm-processors | -- | 若无WRITE_BUFFER_ALL_MAX_BYTES配置时，则初始化为JVM最大内存 \* WRITE_BUFFER_JVM_PROCESSORS。 | 0.4 | | WRITE_PARTITION_MIN | dws.client.write.partition-min | -- | 设置同一个表的partition缓存个数，如果WRITE_PARTITION_POLICY为DN时会为DN个数\*该参数个，多个partition时数据根据分布列做hash分布到不同partition中，不同partition中的数据可并发入库。 | 1 | | WRITE_PARTITION_MAX | dws.client.write.partition-max | -- | 当WRITE_PARTITION_POLICY为DYNAMIC时根据资源使用情况动态调整partition数量在WRITE_PARTITION_MIN和WRITE_PARTITION_MAX之间（当前未实现）。 | 1 | | WRITE_FIXED_COPY | dws.client.write.fixed-copy | -- | fixed copy模式，该模式仅在WRITE_MODE配置为copy\*时生效，开启后第一条数据提交到client时就会建立和数据库的copy流通道，数据会被直接写入数据库IO流中，不在缓存中攒批，为了数据去重，重复数据会被放入下个批次中以及重试获得全部数据因此部分场景还是需要将数据写入缓存。注意： 1、当前模式只支持入库，不支持删除操作，同时要求所有数据入库字段一致 2、当RETRY_MAX_TIMES=1 不使用client的重试并且是纯写入场景此时可以完全不在client攒批直接流式写入dws数据库，以节省客户端内存资源，纯写入场景：无主键表、自增主键（client不设置主键字段的值）、有主键但客户端强制设置WRITE_MODE=copy，由业务保证数据无重复、WRITE_MEMORY_DUPLICATE_REMOVAL=false关闭内存去重 3、此模式下需要一直持有一个连接池资源。 | false | | WRITE_MODE | dws.client.write.mode | writeMode | 数据写入模式： - AUTO: 数据量低于copyWriteBatchSize使用UPSERT方式入库，否则使用COPY_UPSERT方式入库。 - COPY_MERGE: - 有主键使用copy+merge入库。 - 无主键使用copy入库。 - COPY_UPSERT： - 无主键使用copy入库。 - 有主键使用copy + upsert入库。 - UPSERT： - 无主键使用insert into。 - 有主键使用upsert入库。 - UPDATE: - 使用update where语法更新数据，若原表无主键可选择指定uniqueKeys，指定字段不要求必须是唯一索引，但非唯一索引可能会影响性能。 - COPY_UPDATE: - 数据先通过copy方式入库到临时表，通过临时表加速使用update from where方式更新目标数据。 - UPDATE_AUTO: - 批量小于copyWriteBatchSize使用UPDATE，否则使用COPY_UPDATE。 | AUTO | | WRITE_CONFLICT_STRATEGY | dws.client.write.update.conflict-policy | conflictStrategy | 数据库存在主键时的主键冲突策略： - INSERT_OR_IGNORE：主键冲突时忽略新数据。 - INSERT_OR_UPDATE：主键冲突时使用新的数据列更新原数据对应列。 - INSERT_OR_REPLACE：主键冲突时使用新数据替换原数据，数据库中新数据不包含列设置为null，对于全列更新和INSERT_OR_UPDATE无差异。 | INSERT_OR_UPDATE | | READ_THREAD_SIZE | dws.client.read.thread-size | -- | client内部读取的最大线程数量。 | 1 | | WRITE_THREAD_SIZE | dws.client.write.thread-size | ConnectionSize | client内部入库线程池大小，每个线程会占用一个连接因此也是连接池大小，连接池用于执行入库、查询、SQL事件，在入库中缓存达到写入条件时将会提交到线程池中执行入库，如果提交时线程池资源无空闲将会等待空闲后提交，提交成功缓存将会被清空，业务才能继续写缓存，否则将会被阻塞；可通过配置参数WRITE_PARTITION_MIN增加同一表缓存份数，结合线程池大小提升整体入库速度。注意：在1.x版本中缓存按表维度收集，因此同一张表只会有一份缓存，缓存数据需要保证入库的先后顺序和缓存写入顺序一致，该版本中未强制保证一致，在数据库入库速度小于缓存写入速度时会产生并发入库行为，会导致锁冲突或者乱序问题，因此建议设置1。 | 1 | | READ_BATCH_SIZE | dws.client.read.batch-size | -- | 最多一次将readBatchSize条Get请求合并提交。 | 128 | | READ_QUEUE_SIZE | dws.client.read.queue-size | -- | Get请求缓冲池大小。 | 256 | | READ_TIMEOUT | dws.client.read.timeout | -- | Get操作的超时时间，默认值0表示不超时。 | 0 | | WRITE_NOT_WRITE | dws.client.write.disable | -- | 关闭入库，数据会完整的按照正常入库流程执行客户端所有操作，但是会跳过数据写入数据库这一步，可用于测试客户端排除DWS后的整体链路性能以分析性能问题。 | false | | WRITE_TABLE_COMPARE_FIELD | dws.client.write.table.compare-field | compareField | 用于配置比较字段包含的字段必须满足数据值比当前值小才能入库相当于在入库时加了一个where条件例如： INSERT INTO "test"."compare_test"("id", "age", "update_time") VALUES (?, ?, ?) ON CONFLICT ("id") DO UPDATE SET "id"=EXCLUDED."id", "age"=EXCLUDED."age", "update_time"=EXCLUDED."update_time" WHERE "update_time"\< EXCLUDED."update_time" | - | | LOG_SWITCH | dws.client.log.enable | logSwitch | 日志开关，开启后会打印比较详细的过程日志，便于在调试或定位问题时开启。 | false | | onFlushSuccess | -- | onFlushSuccess | 数据入库成功后的回调函数。 | - | | onError | -- | onError | 后台任务执行失败的回调函数。 | - | | WRITE_TABLE_UNIQUE_KEY | dws.client.write.table.unique-key | uniqueKeys | 该参数用于在表中无主键但是有唯一索引时，在入库时使用该参数指定字段做唯一约束，在update场景中该字段不需要是唯一索引或者主键，但upsert场景必须要唯一索引或主键。 | - | | WRITE_COPY_FORMAT | dws.client.write.copy-format | copyMode | 使用copy入库的格式： CSV：将数据拼接成字符串数据用双引号包裹的CSV格式，其中字段间以逗号分割，数据间以换行分割。使用jdbc copy api入库，该方式性能略低于DELIMITER方式，但比较稳定可靠。 DELIMITER：将数据字段使用copy api入库，其中字符间以0x1E分隔，数据间以0x1F分隔。该方式要求数据不包含分隔符，如包含将报错不能正常入库，且该方式定义null字符串为null数据，如果数据为null字符串将被设置为null。 | CSV | | WRITE_COPY_DELIMITER | dws.client.write.copy-delimiter | -- | COPY入库时字段之间的分隔符。 | -- | | WRITE_COPY_EOF | dws.client.write.copy-eof | -- | COPY时字符串分割方式一条数据的结尾符号。 | -- | | WRITE_COPY_FORMAT_CSV_ESCAPE_QUOTA | dws.client.write.copy-format.csv.escape-quota | -- | COPY入库使用CSV模式时是否将字段用引号包围。 | true | | WRITE_TABLE_FIELD_CASE_SENSITIVE | dws.client.write.table.case-sensitive | caseSensitive | 表字段大小写是否敏感。 | false | | WRITE_TABLE_CREATE_TEMP_MODE | dws.client.write.table.create-temp-mode | createTempTableMode | 在使用copy merge/upsert时，创建临时表方式： - AS：使用create temp table \*\*\* as select \* from \*\*\* as方式创建，该方式支持表中带自增字段的使用，但性能略低。 - LIKE：使用create temp table \*\*\* like方式创建，该方式不支持表中带自增字段等临时表不支持类型的字段。 - CUSTOM（仅2.x支持）：自定义临时表的创建，创建时整体使用like方式，创建SQL可通过多个字段影响：create WRITE_TABLE_TEMP_TYPE temp table name_tmp like name WRITE_TABLE_TEMP_INCLUDING with ( WRITE_TABLE_TEMP_WITH ) DISTRIBUTE BY WRITE_TABLE_TEMP_DISTRIBUTE | AS | | WRITE_TABLE_TEMP_TYPE | dws.client.write.table.temp-type | -- | 当WRITE_TABLE_CREATE_TEMP_MODE=CUSTOM时配置临时表类型： COMMON：默认类型。 VOLATILE: volatile类型，需要内核支持。 | COMMON | | WRITE_TABLE_TEMP_INCLUDING | dws.client.write.table.temp-including | -- | 当WRITE_TABLE_CREATE_TEMP_MODE=CUSTOM时配置临时表including参数。 | including defaults | | WRITE_TABLE_TEMP_WITH | dws.client.write.table.temp-with | -- | 当WRITE_TABLE_CREATE_TEMP_MODE=CUSTOM时配置临时表with参数。 | - | | WRITE_TABLE_TEMP_DISTRIBUTE | dws.client.write.table.temp-distribute | -- | 当WRITE_TABLE_CREATE_TEMP_MODE=CUSTOM时配置临时表distribute参数。 | - | | WRITE_FORMAT_NUMBER_TO_DATE | dws.client.write.format.number-to-date | numberAsEpochMsForDatetime | 如果数据库字段是时间类型（date/time/timestamp）并且数据源为数字类型，是否使用数字转时间功能，转换时会统一转换成标准时间戳（毫秒），然后创建对应类型java对象。说明： - 1.x在copy入库场景该参数不生效。 - 1.0.9前该参数为开启状态，且如果数据是数字类型字符串也将视为时间戳。 | false | | WRITE_FORMAT_DATE_LONG_AS_EPOCH_MS | dws.client.write.format.long-as-epoch-ms | -- | 如果数据库字段类型是时间类型，是否将源数据中的long当时间戳做转换。 | false | | WRITE_FORMAT_INT_TO_DATE_TYPE | dws.client.write.format.number-to-date.int-type | -- | 当WRITE_FORMAT_NUMBER_TO_DATE=true时，如果数据为int，转换逻辑： days：当作相对于元年的天数转换。 seconds：当作相对于元年的秒。 | - | | WRITE_FORMAT_LONG_TO_DATE_TYPE | dws.client.write.format.number-to-date.long-type | -- | 当WRITE_FORMAT_NUMBER_TO_DATE=true时，如果数据为long，转换逻辑： days：当作相对于元年的天数转换。 seconds：当作相对于元年的秒。 ms：当作相对于元年的毫秒。 | - | | WRITE_MEMORY_DUPLICATE_REMOVAL | dws.client.write.memory-duplicate-removal | -- | 配置dws-client的内存去重能力，关闭后重复数据完全由数据库去重。 | true | | WRITE_FORMAT_NUMBER_STRING_TO_DATE | dws.client.write.format.number-string-date | -- | 如果数据库字段是时间类型（date/time/timestamp）并且数据源为数字字符串，是否将其转换为数字，转换后可使用WRITE_FORMAT_NUMBER_TO_DATE功能。 | false | | WRITE_FORMAT_STRING_TO_DATE | dws.client.write.format.string-date | stringToDatetimeFormat | 全局配置，如果数据库字段是date类型并且数据源为字符串类型，通过SimpleDateFormat按此格式转换为日期类型，然后通过日期中的时间戳构造数据库对应类型数据。说明： 2.0.0-r3版本之前date、time、timestamp类型会同时使用这个参数。 | - | | WRITE_FORMAT_STRING_TO_TIMESTAMP | dws.client.write.format.string-timestamp | -- | 全局配置，如果数据库字段是timestamp类型，并且数据源为字符串类型，通过SimpleDateFormat按此格式转换为日期类型，然后通过日期中的时间戳构造数据库对应类型数据。 | - | | WRITE_FORMAT_STRING_TO_TIME | dws.client.write.format.string-time | -- | 全局配置，如果数据库字段是time类型，并且数据源为字符串类型，通过SimpleDateFormat按此格式转换为日期类型，然后通过日期中的时间戳构造数据库对应类型数据。 | - | | WRITE_FORMAT_DATE_STRING_FORMAT | dws.client.write.format.date-string-formats | -- | 如果数据库字段是date、time、timestamp类型，并且数据源为字符串类型，指定每个字段的时间格式，按k-v格式配置，文件配置时格式为字段名称:格式多个配置使用;分割，如果格式中有:等字符需要使用双引号，如：update_time:"yyyy-MM-dd hh:mm:ss"。 | - | | WRITE_FORMAT_DATE_0000 | dws.client.write.format.date-0000 | -- | 如果数据库字段是时间类型（date/time/timestamp）并且数据源为字符串 0000-00-00 00:00:00 ，是否转换为时间戳0对应时间。 | false | | WRITE_UPDATE_INCLUDE_PRIMARY_KEY | dws.client.write.update.include-pk | updateAll | upsert时set字段是否包含主键，hstore表在全列更新（set字段包含数据库所有字段）时不需要反查数据库，性能会更好。 | true | | JDBC_AUTO_COMMIT | dws.client.jdbc.auto-commit | autoCommit | 入库时使用自动事务。 | false | | WRITE_FORMAT_STRING_U0000 | dws.client.write.format.string-u0000 | -- | 是否删除字符串中的特殊字符\\u0000，如果不删除copy会报错，只能使用upsert入库。 | true | | WRITE_FORMAT_DECIMAL_DEF_TYPE | dws.client.write.format.decimal-def.type | -- | 当数据库为NUMERIC或者DECIMAL时，输入为空或者空字符串时填充默认值策略： null：填充为空。 zero：填充为0。 custom：通过参数WRITE_FORMAT_DECIMAL_DEF_CUSTOM自定义。 | null | | WRITE_FORMAT_DECIMAL_DEF_CUSTOM | dws.client.write.format.decimal-def.custom | -- | WRITE_FORMAT_DECIMAL_DEF_TYPE=custom时，设置自定义值。 | - | | NETWORK_INTERNAL_PRIVATE_IP | dws.client.network.internal-privateIp | -- | WRITE_PARTITION_POLICY=DN时如果客户端程序和dws集群网络不在同一网络（内部通信网络），通过此参数配置网络ip映射格式为：内部IP:外部IP多个使用;分割。**当开启直连DN后，此参数是必须配置的，否则连接不上DN节点。** | - | | DISABLE_AUTO_CONFIG_INTERNAL_PRIVATE_IP | dws.client.auto-config.internal-privateIp.disable | -- | 禁用自动IP映射配置。默认情况下此配置为关闭状态，会去内核的 pgxc_node_ips表中查找DN节点的IP映射关系，但目前内核**未支持此表**，所以配置与否此配置并无影响。 | false | | TIMEOUT_TASK | dws.client.timeout.task | -- | 执行任务的超时时间，包括重试时间。 | 10m | | TIMEOUT_SQL_STATEMENT | dws.client.timeout.statement | -- | SQL执行超时时间。 | 5m | | TRACE_SWITCH | dws.client.trace.enable | - | 打印trace记录，日志会记录主要的步骤及耗时，用于分析性能问题。 | false | | SYSTEM_TIMEZONE | dws.client.system.timezone | -- | 用于设置系统的时区以便于在做时间转换时时区正确，该参数会影响整个Java进程的时区，理论上需要和dws时区一致，如果不设置则保持和客户端进程默认时区一致。 | - | | PRINT_METRICS_NAMES | dws.client.print-metrics.names | -- | 用于配置需要打印的指标名称，配置为正则表达式，匹配到的指标就会被打印，可配置.\*打印所有指标，确认选择哪些指标： rps_.\*：用于打印每个表的入库速率 write_rpc：整个实例的写入速度 action_submit_wait：提交线程池的等待时间 buffer_size_.\*：每次入库攒批数量 action_cost_.\*：每次执行入库耗时 addBatch_cost_.\*：执行addBatch的用时统计 executeBatch_cost_.\*：执行executeBatch的用时统计 buildBuffer_cost_.\*：构建copy buffer的用时统计 copy_cost_.\*：执行copy的用时统计 | - | | PRINT_METRICS_PERIOD | dws.client.print-metrics.period | -- | 打印指标频率。 | 30s | | WRITE_HSTORE_UPSERT_AUTO_COMMIT | dws.client.write.hstore-upsert-auto-commit | -- | hstore是否开启单行提交。 | false | | READ_SCAN_FETCH_SIZE | dws.client.scan.fetch-size | -- | Scan时每次从数据库读取的条数。 | 2000 | | READ_SCAN_TIMEOUT | dws.client.scan.timeout | -- | Scan的超时时间。 | 60s | | IGNORE_TABLE_NOT_EXIST | dws.client.table.ignore-not-exist | -- | 获取表结构时如果表不存在是否忽略，若配置为true，则表不存在也不会抛异常，建议配置为false进行正常检查。 | false | | WRITE_PREPARED_STATEMENT_TYPE | dws.client.write.prepared-statement.type | -- | 设置statement时是否携带type。 | false | | WRITE_DN_APPEND_LAST_DATA_CHANGED_NUM | dws.client.write.dn.append-last-data-changed-num | -- | 直连DN时是否开启刷新物化视图。 | true | | WRITE_DN_CHECK_CLUSTER_STATUS | dws.client.write.dn.check-cluster-status | -- | 直连DN时是否在CN上检查集群状态，若CN状态异常则拒绝入库并抛出异常。 | true | | ENABLE_PRIMARY_TABLE_WITH_COPY_TO_COPY_UPSERT | dws.client.primary-copy-to-copy-upsert.enable | -- | 此参数**2.1.0.3版本新增**。是否支持将有主键表使用copy模式强制转换为copy_upsert。此参数为了适配有主键时走更新的逻辑，避免同主键数据走copy报错。 | false | | ENABLE_COPY_COMPATIBLE_ILLEGAL_CHARS | dws.client.copy.compatible-illegal-chars.enable | -- | 此参数**2.1.0.3版本新增**。COPY语句是否支持compatible_illegal_chars参数。开启此参数后，copy语句会拼接一个compatible_illegal_chars参数，避免特殊字符在copy时报错。 | false | 2.x删除参数 | 参数 | 说明 | 删除原因 | |:---|:---|:---| | logDataTables | 入库时需要打印数据的表，便于在定位问题时对比数据。 | 基本无使用场景，开发节点debug更方便，生产也不可能配置此参数。 | | batchOutWeighRatio | 为提高整体吞吐，当对autoFlushBatchSize要求不是很严格时，可设置该参数，当往buffer中提交数据时buffer中数据量大于batchOutWeighRatio \* autoFlushBatchSize时提交线程将会执行提交入库的任务，该参数用于避免业务线程提交任务，尽量使用后台线程执行提交。 | WRITE_FORCE_FLUSH_BATCH_SIZE参数功能一致，更直观使用。 |