gbase数据、南大通用产品文档：GBase8a创建consumer task

语法格式

transaction topic
CREATE KAFKA CONSUMER TRANSACTION TOPIC
BROKERS 'ip:port, ip:port,…’;

loader topic
CREATE KAFKA CONSUMER LOADER TOPIC
BROKERS 'ip:port, ip:port,…’ [PARTITIONS list>] DURATION INTO TABLE dbname.tbname ;

GBase 8a MPP Cluster 产品手册
5 数据库管理指南
文档版本953（2022-09-15）
南大通用数据技术股份有限公司
1269
表5- 163 参数说明
字段名称
含义说明
consumer name
消费任务的名称。唯一，不允许重复，最大长度64 bytes；
kafka_topic_name
需要消费的kafka topic 名称，最大长度64bytes。由用户事先
在kafka server 上创建。
注：
创建时kafka topic name 必须在kafka server 上已真实存在
Ip
kafka broker 的ip。
Port
kafka broker 的port。
Partition list
指定从topic 的哪个/哪些partition 消费数据，形式为’0,1,2’，
如果指定使用全部partitions 可以用all。
默认是从topic 的所有partition 消费数据。
创建loader topic consumer 时，partition 必须是kafka server 上
对应topic 的有效partition。
Time in ms
以ms 为单位的消耗时间。含义是：
Loader topic 的consumer task 会告诉gnode 的loader ，在
duratition 的时间内，持续从kafka server 读取数据，数据在内
存中缓存，达到duratition 时间后，再让数据落盘。
dbname.tbname
数据库名.表名，意为loader topic 的consumer 加载数据的目
标表。
Loader_options
从加载命令选项沿袭过来，
具体含义和用法可以参考5.2.2 章
节集群加载语法说明，可以根据加载的数据文件格式进行设
置。具体包括：
Charset、
data_format、
having lines、
separator null_value ‘’、
fields
terminated by ‘’、enclosed by ‘’、preserve blanks、length ‘’、
table_fields 、
autofill、
lines terminated by ‘’、
max_bad_records 、
datetime format 、
date format 、
timestamp format 、
time format、
parallel 、skip_bad_file 。
注意

transaction topic 类型的consumer 不允许两个consumer task 使用相同的
topic name+brokers 组合。

创建consumer 时，需要kafka 集群中topic 存在，且所写的broker：port
能正常提供服务。
示例
创建名为test1 的consumer task，从topic_1 消费数据：
Create kafka consumer test1 transaction topic topic_1
brokers ’10.10.10.10:9092,10.10.10.11:9092’;

GBase 8a MPP Cluster 产品手册
5 数据库管理指南
文档版本953（2022-09-15）
南大通用数据技术股份有限公司
1270

管理 Enterprise Replication
（SQL
管理 API）
随同 admin() 或 task() 函数，使用 cdr 参数来管理 Enterprise Replication。
语法

元素
描述
关键考虑
command_name cdr 命令名。
您不可包括任何连字符、标志或包含
cdr 命令行实用程序要求的
command_name 的其他选项。
option_name
command_name 的 cdr 命令行选项
的一个或多个元素。
这些元素必须以引号分隔。
还包括
（以
正确的顺序）任何连字符、标志或
command_name 要求的 cdr 命令行选
项的其他元素。您可使用缩写。

用法
使用这些函数产生的管理 Enterprise Replication 的效果与 cdr 命令行实用程序相同。

该 SQL 管理 API 支持 cdr 命令用于管理 Enterprise Replication。不支持下列监视
Enterprise Replication 的命令：
 cdr list grid
 cdr list replicate
 cdr list replicateset
 cdr list server
 cdr list template
 cdr stats recv
 cdr stats rqm
 cdr –V
 cdr view

GBase 8s 管理员参考
南大通用数据技术股份有限公司 - 698 -
第一个参数必须仅包括 cdr 命令名，与 GBase 8s Enterprise Replication 指南 中 cdr
实用程序附录指定的名称完全一致，比如 cdr define server。命令名区分大小写且不支
持缩写（比如 cdr
sto
replset 而不是 cdr
stop
replicateset）。在将这些参数传到 cdr
实用程序之前，该 SQL 管理 API 不执行任何验证。
第二个及任何后续参数包括命令选项。可在一个或最多六个参数中指定这些选项。
下列示例说明使用 SQL 管理 API 来定义 Enterprise Replication 服务器：
EXECUTE FUNCTION task ( 'cdr define server', '--connect=g_amsterdam
--ats=/local0/er/ats --ris=/local0/er/ris --init g_amsterdam' );
下列示例展示这些选项可如何分为几个参数；上面的语句还可写为：
EXECUTE FUNCTION task( 'cdr define server',
'--connect=g_amsterdam',
'--ats=/local0/er/ats',
'--ris=/local0/er/ris',
'--init g_amsterdam' );
下列示例显示一个参数内的双引号字符串：
EXECUTE FUNCTION task('cdr change replicate',
'-d repl_1 -"db1@server1:antonio.table1" "db2@server2:carlo.table2"');
1 六项参数的最大值。

sqldone() 函数确定数据库服务器当前是否正在处理 SQL 请求。

语法
mint sqldone();

用法

请在下列情况下使用 sqldone() 来检测数据库服务器的状态：
在调用 sqlbreak() 函数来确定数据库服务器是否正在处理 SQL 请求之前。
在信号句柄函数中，在调用 longjmp() 系统函数之前。如果 sqldone() 返回零（数据
库服务器空闲），则仅使用信号句柄函数中的 longjmp()。

当 sqldone() 函数确定当前数据库服务器未正在处理 SQL 请求时，您可假定数据库
服务器未开启任何其他处理，直到您的应用程序发出它的下一请求为止。

您可能想要为 -439 值创建定义的常量，来使得您的代码更加可读。例如，下列代码
片段创建 SERVER_BUSY 常量，然后使用它来检测 sqldone() 返回状态：
#define SERVER_BUSY -439
.
.
.
if (sqldone() == SERVER_BUSY)

返回代码
0
数据库服务器当前没有正在处理 SQL 请求：它是空闲的。

GBase 8s ESQL/C 编程指南
南大通用数据技术股份有限公司
- 947 -
-439
数据库服务器当前正在处理 SQL 请求。