南大通用GBase 8a和OGG + Kafka同步相关参数介绍

本文介绍GBase 8a和OGG +Kafka相关参数,主要用于consume功能,与加载导出等无关。

gbase_8a_gcluster.cnf

9.5.2版本kafka参数列表

gbase> show variables like '%kaf%';
+----------------------------------------------------+-----------+
| Variable_name                                      | Value     |
+----------------------------------------------------+-----------+
| _gbase_kafka_transaction_mode                      | OFF       |
| _t_gcluster_kafka_allow_offset_jump                | 1         |
| _t_gcluster_kafka_consumer_compare_field_only_once | 0         |
| _t_gcluster_kafka_consumer_force_compare_field     | 1         |
| _t_gcluster_kafka_ignore_when_update_not_hit       | 0         |
| _t_gcluster_kafka_null_transform                   | 0         |
| _t_gcluster_kafka_seek_offset                      | 0         |
| _t_gcluster_kafka_trust_kafka_returned_offset      | 0         |
| _t_kafka_varchar_auto_truncate                     | 0         |
| gbase_kafka_broker_version                         |           |
| gbase_kafka_keytab                                 |           |
| gbase_kafka_principal                              |           |
| gcluster_kafka_batch_commit_dml_count              | 100000    |
| gcluster_kafka_consume_batch                       | 10        |
| gcluster_kafka_consume_timeout                     | 2000      |
| gcluster_kafka_consumer_escape_zero                | 0         |
| gcluster_kafka_consumer_latency_time_statistics    | 0         |
| gcluster_kafka_consumer_output_charset_name        |           |
| gcluster_kafka_consumer_special_use_charset_name   |           |
| gcluster_kafka_consumer_support_partial_update     | 0         |
| gcluster_kafka_data_buf_size                       | 0         |
| gcluster_kafka_dataflow                            | 0         |
| gcluster_kafka_debug_on                            | 0         |
| gcluster_kafka_delete_execute_directly             | 0         |
| gcluster_kafka_ignore_if_table_not_exist           | 0         |
| gcluster_kafka_ignore_pos_field                    | 0         |
| gcluster_kafka_loader_max_start_count              | 20        |
| gcluster_kafka_local_queue_size                    | 201000    |
| gcluster_kafka_max_message_size                    | 100000000 |
| gcluster_kafka_parallel_commit                     | 1         |
| gcluster_kafka_primarykey_can_be_null              | 0         |
| gcluster_kafka_result_check                        | 0         |
| gcluster_kafka_user_allowed_max_latency            | 10000     |
+----------------------------------------------------+-----------+
33 rows in set (Elapsed: 00:00:00.00)

gcluster_kafka_consumer_enable

打开 kafka consumer 功能,如果不打开,则 consumer 相关命令都不可用(报错)

  • 1 打开
  • 0 关闭, 默认

该参数在集群show 时看不到,但在默认的配置文件里能看到。

gcluster_assign_kafka_topic_period

自动接管 consumer 的时间周期,单位为秒。

例如 A 节点宕机了,最大需要等待 gcluster_assign_kafka_topic_period秒之后,A 节点负责的同步任务会被其他节点接管。最小值 20s,最大值 120s。

gcluster_kafka_max_message_size

从 kafka topic 获得消息的最大长度,
单位为字节,最大值 1000000000 字节,这个值需要大于等于 kafka server 的配置(message.max.bytes),否则可能造成消费问题,如果 kafka 队列中存在一条消息,其大小超过 gcluster_kafka_max_message_size 就会造成消费卡住。

gcluster_kafka_batch_commit_dml_count

一次提交 dml 操作的数量。

适当调大能明显提高性能,但是如果一个 topic 涉及的表很多(几百个表)则建议该参数调小,表越多越应该调小,调小的目的是使得一次提交命中的表少一些,具体需要结合具体用户场景、同步速度、资源占用情况具体对待。未来启用新事务后,表数量多对性能的影响会降低,会再次更新手册。需要注意的是,此参数是一个意向值,程序未必会严格按照此参数来提交,比如如果一个事务包含大量 DML 操作,那么程序必须确保事务完整性;再比如从 kafka 取消息、解析消息的速度慢于往单机提交数据的速度,那么程序也会选择先提交,而不是一定要等待满足 gcluster_kafka_batch_commit_dml_count 参数。

gcluster_kafka_user_allowed_max_latency

允许消息在 GBase 8a MPP Cluster 集群层缓存多长时间,超时之后必须马上提交,单位是毫秒。

此参数与gcluster_kafka_batch_commit_dml_count 作用类似,都是决定什么时候提交的。多攒一些数据再提交,有利于降低磁盘占用,如果用户对数据延迟不太敏感,而对磁盘占用比较敏感,可以通过这个参数来调节。典型值一般可以设置为 50000~20000,需要注意提交动作本身也需要消耗时间。

gcluster_kafka_local_queue_size

储存 dml 操作的队列的长度,建议至少为 gcluster_kafka_batch_commit_dml_count 的二倍多一些。

gcluster_kafka_consume_batch

consumer 一次读取 kafka 消息的条数。
如果 kafka 队列里的消息 size 较小,可以设大,反之设小,此参数对性能的影响不大,所以一般没必要设太大,建议设为 10~1000。

gcluster_kafka_ignore_pos_field

控制单个 consumer 是否比对 POS(防止
重复消费)。客户多线程往 kafka 中写入数据,写入 kafka 的数据不能确保 POS有序,原 consumer 消费数据时会做 POS 检查导致无序的数据入库时会有遗漏。现在参数 gcluster_kafka_ignore_pos_field,控制 consumer 是否进行 POS 检查。

  • POS 检查开启,consumer 消费时会丢弃已消费序号之前的消息;
  • POS 检查关闭,consumer 会将 kafka 的每条消息均入库,所以需要生产端确保发送到 kafka的消息无重复。
  • 默认值为 0,即检查重复消息;
  • 值为 1 时,不检查重复消息。

用于 Consumer 消费 only insert 消息,客户能保证 kafka 消息无重复的特殊场景。配置方法可以手动修改 gclusterdb.kafka_consumers。如:

Update gclusterdb.kafka_consumers set common_options='gcluster_kafka_ignore_pos_field=1' where name='consumer_1';

最后重启consumer_1。

_t_kafka_varchar_auto_truncate

在consumer消费kafka信息时,遇到长度超数据库定义长度的字段(仅限 varchar 类型),开启可以自动进行截位并正常消费入库模式。缺省值为 0;设置值为 1 时,表示让 consumer 对 json消息中的 after 内容进行长度判断,如果长度超过了目标表的列宽,则自动按列宽(字符长度)截断,只对 varchar 列做处理。

gcluster_kafka_message_format_type

设定 consumer 在解析 kafka 消息时,以什么格式来解析。
取值范围:JSON、PUREDATA、AUTO_DETECT
说明:

  • puredata 对应 rtsync 生产的 protobuf 消息;
  • AUTO_DETECT(默认)是让 consumer 自己侦测消息格式,这时候 consumer 会先尝试用 puredata 格式进行解析,通过就认为是 puredata 格式,否则就认为是json 格式。

注:consumer 启动后,只在解析第一条消息时做这个判断,后面直接用这个判断结果。

gcluster_kafka_ignore_pos_field

控制单个 consumer 是否比对 POS(防止重复消费)。

客户多线程往 kafka 中写入数据,写入 kafka 的数据不能确保 POS 有序,原 consumer 消费数据时会做 POS 检查导致无序的数据入库时会有遗漏。参数 gcluster_kafka_ignore_pos_field,控制 consumer 是否进行 POS 检查。

  • POS检查开启,consumer 消费时会丢弃已消费序号之前的消息;
  • POS 检查关闭,consumer 会将 kafka 的每条消息均入库,所以需要生产端确保发送到 kafka 的消息无重复。

配置方法:手动修改 gclusterdb.kafka_consumers

Update gclusterdb.kafka_consumers set common_options='gcluster_kafka_ignore_pos_field=1' where name='consumer_1';

重启 consumer_1。

gcluster_kafka_broker_version

设定 kafka server 的版本,例如 0.9.0,0.8.2… 当 kafka server 低于 0.10 版本的时候必须设置此参数,而高于 0.10版本不要设置。

gbase_kafka_principal

配置 kafka kerberos principal

gbase_kafka_keytab

配置 kafka kerberos keytab file path

gcluster_kafka_ignore_if_table_not_exist

consumer 处理一个消息时,如果消息指定的目标表不存在,是否自动忽略此消息。1 代表忽略。

gcluster_kafka_parallel_commit

consumer 向 gnode 发送 sql 是否采用并行方式。默认值是 1,代表不并行。这个参数目前不要使用,会造成主备不一致,或者 delete 不掉数据。