更新日期：2024年09月11日
当 TAPEDEV 和 LTAPEDEV 指定目录时，使用 IFX_ONTAPE_FILE_PREFIX 环
境变量来指定备份文件名的前缀（替换 hostname_servernum 格式）。如果没有设置值，
那么各个等级的文件名为 hostname_servernum_Ln，而日志文件的文件名为
hostname_servernum_Log nnnnnnnnnn。
如果将 IFX_ONTAPE_FILE_PREFIX 的值设置为 My_Backup，那么备份文件名具有以下
名称：
My_Backup_L0
My_Backup_L1
My_Backup_L2
My_Backup_Log0000000001
My_Backup_Log0000000002
GBase 8s SQL 指南：参考
南大通用数据技术股份有限公司 - 176 -
string 用于备份文件名的前缀。
8a Consumer 是GBase8ampp 自带的消费组件，用来同步消费Kafka 数据
到8a MPP Cluster 数据库中，
备注：目前由其它源库同步到GBase8a MPP 时，采用的都是：
GBase RTSync（管理、读）+ 8a Consumer（写）的方式进行环境部署
配置步骤如下：
步骤一：修改8a mpp 配置文件
修改8a mpp 所有节点的/opt/gcluster/config/gbase_8a_gcluster.cnf
gcluster_kafka_consumer_enable=1
gcluster_lock_level=10
_gbase_transaction_disable=1
_gcluster_insert_cache_buffer_flag=1
gcluster_assign_kafka_topic_period=20
gcluster_kafka_max_message_size=1000000
gcluster_kafka_batch_commit_dml_count=100000
gcluster_kafka_local_queue_size=210000
gcluster_kafka_user_allowed_max_latency=20000
gcluster_kafka_consume_batch=100
gcluster_kafka_parallel_commit=5
gcluster_kafka_primarykey_can_be_null=1
gcluster_kafka_data_buf_size=20000
gcluster_kafka_message_format_type=PUREDATA（与RTSync 配置的数据格式
保持一致）
修改8a mpp 所有节点的/opt/gnode/config/gbase_8a_gbase.cnf
GBase RTSync 同步工具手册
- 18 -
南大通用数据技术股份有限公司
gbase_tx_log_mode=ONLY_SPECIFY_USE
_gbase_transaction_disable=1
gbase_buffer_insert=1024M （有大事务时需要调大，根据8a 的服务器内存情
况）
gbase_tx_log_flush_time=5
备注：修改完以上两个配置文件，则需要service gcware restart 重新启动
集群服务才生效
步骤二：创建8a consumer
通过gccli 命令登录8a mpp 命令行执行以下8a consumer 相关语句
创建语句：
create kafka consumer test transaction topic testtp brokers
‘192.168.5.11:9092,192.168.5.12:9092’;
说明：
1、创建consumer 时topic 需要与RTSync 的
config_kafka_orato8aMQ.properties 配置文件中的topic.name 一致。
2、brokers 就是配置部署的kafka 节点信息
步骤三：启动8a consumer
启动语句：
#start kafka consumer test
步骤三：监控及查询consumer 同步状态
查看consumer task 属性：
# show kafka consumer test;
从系统表中查询数据同步状态
select * from information_schema.kafka_consumer_status;
gbase> select * from information_schema.kafka_consumer_status;
+---------+---------------+---------+------------+------------+-----
GBase RTSync 同步工具手册
南大通用数据技术股份有限公司
- 19 -
-------+----------------+---------------+-----------+
| TOPIC
| IP
| STATUS
| MIN_OFFSET | MAX_OFFSET |
CUR_OFFSET | PROCESS_OFFSET | COMMIT_OFFSET | EXCEPTION |
+---------+---------------+---------+------------+------------+-----
-------+----------------+---------------+-----------+
| k1topic | 192.168.6.161 | STARTED |
0
|
0
|
0
|
0
|
0 |
|
+---------+---------------+---------+------------+------------+-----
-------+----------------+---------------+-----------+
本示例实现了使用可更新结果集获取AUTO_INCREMENT 列的值。
示例如下：
package com.gbase.jdbc.simple;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
public class SampleUpdatableResultSet {
GBase UP 产品手册 6 应用开发指南
文档版本04（2021-04-21） 南大通用数据技术股份有限公司 1105
private
static
final
String
URL
=
"jdbc:gbase://192.168.111.95:5258/test?user=sysdba&password=";
public static void main(String[] args) {
try {
(new SampleUpdatableResultSet()).test();
} catch (Exception ex) {
}
}
public void test() throws Exception {
Connection conn = null;
try {
Class.forName("com.gbase.jdbc.Driver").newInstance();
conn = DriverManager
.getConnection(URL);
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement(
java.sql.ResultSet.TYPE_FORWARD_ONLY,
java.sql.ResultSet.CONCUR_UPDATABLE);
// 创建表
stmt.executeUpdate("DROP TABLE IF EXISTS autoIncTutorial");
stmt.executeUpdate("CREATE TABLE autoIncTutorial ("
+ "priKey INT NOT NULL AUTO_INCREMENT, "
+ "dataField VARCHAR(64), PRIMARY KEY (priKey))");
// 获取自增一字段值
rs = stmt.executeQuery("SELECT priKey, dataField "
+ "FROM autoIncTutorial");
rs.moveToInsertRow();
rs.updateString("dataField", "AUTO INCREMENT here?");
rs.insertRow();
rs.last();
int autoIncKeyFromRS = rs.getInt("priKey");
rs.close();
rs = null;
GBase UP 产品手册 6 应用开发指南
文档版本04（2021-04-21） 南大通用数据技术股份有限公司 1106
System.out.println("Key returned for inserted row: "
+ autoIncKeyFromRS);
} finally {
if (rs != null) {
try {
rs.close();
} catch (SQLException ex) {
}
}
if (stmt != null) {
try {
stmt.close();
} catch (SQLException ex) {
}
}
}
} catch (SQLException ex) {
// 处理错误
System.out.println("SQLException: " + ex.getMessage());
System.out.println("SQLState: " + ex.getSQLState());
System.out.println("VendorError: " + ex.getErrorCode());
} finally {
conn.close();
}
}
}