配置Event Time Event Time是指事件产生的时间，即数据产生时自带时间戳。 语法格式 1 2 3 CREATE SOURCE STREAM stream_name(...) WITH (...) TIMESTAMP BY {attr_name}.rowtime SET WATERMARK (RANGE {time_interval} | ROWS {literal}, {time_interval}); 语法说明 设置Event Time需要选定流中的某一个属性来作为时间戳，同时需要设置Watermark策略。 由于网络等原因，有时会导致乱序的产生；对于迟来的数据，需要Watermark来保证一个特定的时间后去触发Window进行计算。Watermark主要是用来处理乱序数据，流处理从事件产生，到发送到 DLI 服务，中间有一个过程。 Watermark有两种设置策略： 按时间周期 1 SET WATERMARK(range interval {time_unit}, interval {time_unit}) 按事件个数 1 SET WATERMARK(rows literal, interval {time_unit}) 一个逗号表示一个参数，第一个参数表示Watermark发送周期，第二个参数表示允许最大延迟时间。 注意事项 无。 示例 time2事件产生时间开始，每10s发送一次watermark，事件最大允许延迟时间为20s。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 CREATE SOURCE STREAM student_scores ( student_number STRING, /* 学号 */ student_name STRING, /* 姓名 */ subject STRING, /* 学科 */ score INT, /* 成绩 */ time2 TIMESTAMP ) WITH ( type = "dis", region = "", channel = "dliinput", partition_count = "1", encode = "csv", field_delimiter="," ) TIMESTAMP BY time2.rowtime SET WATERMARK (RANGE interval 10 second, interval 20 second); INSERT INTO score_greate_90 SELECT student_name, sum(score) over (order by time2 RANGE UNBOUNDED PRECEDING) FROM student_scores; 每收到10个数据发送一次watermark，事件最大允许延迟时间为20s。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 CREATE SOURCE STREAM student_scores ( student_number STRING, /* 学号 */ student_name STRING, /* 姓名 */ subject STRING, /* 学科 */ score INT, /* 成绩 */ time2 TIMESTAMP ) WITH ( type = "dis", region = "", channel = "dliinput", partition_count = "1", encode = "csv", field_delimiter="," ) TIMESTAMP BY time2.rowtime SET WATERMARK (ROWS 10, interval 20 second); INSERT INTO score_greate_90 SELECT student_name, sum(score) over (order by time2 RANGE UNBOUNDED PRECEDING) FROM student_scores;

配置Processing Time Processing Time是指系统时间，与数据本身的时间戳无关，即在Flink算子内计算完成的时间。 语法格式 1 2 3 4 CREATE SOURCE STREAM stream_name(...) WITH (...) TIMESTAMP BY proctime.proctime; CREATE TEMP STREAM stream_name(...) TIMESTAMP BY proctime.proctime; 语法说明 设置Processing Time只需在timestamp by后配置proctime.proctime即可，后续可以直接使用proctime字段。 注意事项 无。 示例 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 CREATE SOURCE STREAM student_scores ( student_number STRING, /* 学号 */ student_name STRING, /* 姓名 */ subject STRING, /* 学科 */ score INT /* 成绩 */ ) WITH ( type = "dis", region = "", channel = "dliinput", partition_count = "1", encode = "csv", field_delimiter="," )TIMESTAMP BY proctime.proctime; INSERT INTO score_greate_90 SELECT student_name, sum(score) over (order by proctime RANGE UNBOUNDED PRECEDING) FROM student_scores;

运维规范 在日常运维工作中需关注以下运维规范，预防潜在风险并掌握关键应急方案。 表2 GeminiDB Redis运维规范 编号 运维规范 说明 1 确保您的华为云账号绑定的电话号码、电子邮箱等联系方式有效，以便及时收到服务相关通知信息。 华为云服务会在变更、升级、故障通知等必要场景通过网站、邮箱、短信、站内信等方式向您送达通知信息，请确保账号绑定的联系方式有效。 2 订阅重要告警 订阅例如大Key访问、存储容量高、连接数使用率高、CPU使用率高等告警，从而在数据库实例出现风险时提早发现并及时处理。 推荐告警配置请参考设置告警规则。 3 使用负载均衡地址访问时，需要配置访问控制管理，而不是安全组。 负载均衡地址不支持安全组，创建完成后请配置IP访问控制，如未配置白名单，所有与VPC互通的IP地址均可访问实例。具体操作请参考设置负载均衡内网访问控制。 4 配置实例自动扩容 GeminiDB Redis支持自动扩容，可以有效避免数据量突增问题。建议参见存储空间自动扩容开启自动扩容。 5 保持负载处于健康水位 如业务数据量长期高于80%，建议及时扩容。 如业务流量超过实例QPS基准，或CPU利用率长期高于80%，建议及时升级规格或增加节点，避免过载影响业务访问。 在业务流量、连接数等指标突增导致实例算力资源超载的紧急场景，应通过水平添加节点快速提升集群算力。垂直升配CPU规格为滚动模式，耗时较长，不建议用于紧急场景。 6 重命名高危命令 对于高危命令（如flushall、keys），建议您禁用或重命名，以增强实例的安全性。具体操作请参考文档命令重命名。 7 定期进行大key在线诊断 大Key访问是Redis使用中的常见问题。建议在控制台定期查看实例的大Key诊断报告。具体操作请参考大Key诊断。 9 割接结束后执行dbsize dbsize可以保证最终一致性，例如，在割接场景，无过期key，数据导入完成后的的数分钟执行，可得到准确值，确保数据导入前后的一致。

开发规范 在开发业务程序时需重点关注以下开发规范，避免因使用方式不当影响业务稳定。 表1 GeminiDB Redis开发规范 编号 开发规范 说明 1 业务程序需具备合理的自动重连机制。 在规格变更、补丁升级、HA倒换、网络链路抖动或丢包等场景中，业务程序与数据库实例的连接可能出现短暂中断。业务程序应当支持自动重连。 说明： 尽量避免使用Lettuce客户端，因为该客户端在请求多次超时后，不再发起自动重连。建议使用更稳定的Jedis客户端替换Lettuce客户端。 2 业务程序需使用连接池并配置足够连接数。 为避免并发激增时程序获取不到连接，建议业务程序使用连接池，并配置合理的连接池参数。客户端的连接池推荐配置请参考客户端连接池参数配置。 3 针对部分重要操作，业务程序需要具备合理的命令重试机制。 在连接中断或请求超时等场景，业务程序请求可能会短时内执行失败。因此，针对部分重要操作需要设计业务容错机制，使用合理的命令重试间隔和次数，确保重要数据写入或修改成功。 4 使用正确的高可用连接地址，避免单点故障场景影响业务。 业务程序内网连接数据库时，应使用负载均衡地址，实现高可用。应避免将业务程序直连到某个单独的计算节点。 说明： 针对公网访问数据库的场景，同样应避免将业务程序直连到某个单独的计算节点，请参考通过弹性负载均衡绑定弹性公网IP连接实例使用外置ELB进行配置。 5 使用连接池，避免使用单连接以及大量短连接。 由于单连接有高可用风险，而短连接性能差且会消耗大量数据库CPU和网络资源，容易造成瓶颈，因此建议业务程序使用主流SDK的连接池连接GeminiDB Redis实例。 6 当实例中Key数量大于1000个时，禁止执行Keys命令 Keys命令是典型的高危命令，会一次性获取整个实例中的全部数据内容并返回给客户端。当实例中Key数量较多时，执行Keys命令会引起请求阻塞或实例状态异常。 7 避免元素Value过大，避免单Key包含的元素数量过多。 根据最佳实践，建议String类型Key的Value小于10KB，Hash/List/Zset/Set等类型Key中包含元素数小于5000，且单个元素Value小于1KB。 说明： 与社区Redis一样，GeminiDB Redis对大Key存储并不做严格约束，因此需要用户根据规范合理开发业务程序。 8 避免单个命令中一次性操作过多元素或产生过大的网络收发包。 建议MSET/MGET等批量操作Key命令中同时操作的Key数量小于1000。 建议HMSET、HGETALL、LRANGE、ZADD、ZRANGE等同时操作Key中多元素的命令同时操作元素数量小于1000。 由于社区Redis语法中ZREMRANGEBYSCORE命令不提供limit参数，无法控制单次删除的元素数量，建议使用ZRANGEBYSCORE (带limit)+ZREM的先查后删方式替代。 说明： 与社区Redis一样，GeminiDB Redis对大Key访问并不做严格约束，因此需要用户根据规范合理开发业务程序。 9 合理打散Key分布，避免出现热点Key或热点hashtag瓶颈。 对单个Key或具有同hashtag的一组Key的高频访问依赖会产生热Key问题，容易引发计算资源倾斜，请求排队，响应变慢等瓶颈问题。热Key的产生往往来自业务设计对Key的拆分程度不足，需要业务拆分优化。 说明： 反面案例： 对全局超大排行榜高频集中访问。 全实例中仅保存几个Hash Key，每个Key用于存储一整张表格的信息量。 商品库存业务中，全实例Key仅使用少量hashtag分桶，导致热点hashtag查询排队。 10 单次Pipeline中打包命令数量建议小于100条。 使用Pipeline时应遵循“单次少量，分多次”原则。 根据使用经验，并非Pipeline打包数量越大性能越好，不合理的Pipeline用法将可能导致实例CPU、带宽等资源瓶颈，甚至引起请求阻塞。 说明： 与社区Redis相比，GeminiDB Redis对Pipeline的使用并不做严格约束，因此需要用户根据规范合理开发业务程序。 11 避免在Lua脚本中引入高耗时代码。 使用Lua脚本时应遵循轻量化原则，避免引入执行命令次数过多、长时间sleep、大的循环语句等不合理用法。 12 避免在事务中打包过多命令。 使用事务时应遵循轻量化原则，避免单次事务中打包过多命令或复杂度过高的命令。执行的事务中包含命令过多会引起请求阻塞或实例状态异常。 13 避免在不同数据类型下使用相同的Key名称。 社区Redis强制禁止在不同数据类型下使用相同的Key名称。虽然GeminiDB Redis对此不做强制约束，但开发时应当避免此类用法，保持程序清晰和易维护。

请求消息示例 POST /v2/usg/acs/auth/appauth Connection: keep-alive Content-Type: application/json X-Request-ID: 5162fa32dc7e47afafeee39a72a2eec3 Accept-Language: zh-CN Host: api.meeting.huaweicloud.com X-Token-Type: LongTicket Authorization: HMAC-SHA256 signature=3eca3f0f1e90ed55de38388066d02f1b7a86571a8ce30823af1df7c4edd7e086,access=ZmRiOGU0Njk5NTg2NDU4YmJkMTBjODM0ODcyZGNjNjI= User-Agent: Apache-HttpClient/4.5.3 (Java/1.8.0_191) { "appId": "fdb8e4699586458bbd10c834872dcc62", "clientType": 72, "expireTime": 1627722929, "nonce": "EycLQsHwxhzK9OW8UEKWNfH2I3CGR2nINuU1EBpv162d42d92s", "userEmail": "******", "userId": "testuser@mycorp.com", "userName": "testuser", "userPhone": "173****9092" }

CURL命令示例 curl -k -i -H 'content-type: application/json' -X POST -H 'Content-Type: application/json,Accept-Language: zh-CN,X-Token-Type: LongTicket,Authorization: HMAC-SHA256 signature=3eca3f0f1e90ed55de38388066d02f1b7a86571a8ce30823af1df7c4edd7e086,access=ZmRiOGU0Njk5NTg2NDU4YmJkMTBjODM0ODcyZGNjNjI=' -d '{"appId": "fdb8e4699586458bbd10c834872dcc62","clientType": 72,"corpId": "807074304","expireTime": 1597824907000,"nonce": "EycLQsHwxhzK9OW8UEKWNfH2I3CGR2nINuU1EBpQ","userEmail": "******","userId": "alice@ent01","userName": "alice","userPhone": "173****9092"}' 'https://api.meeting.huaweicloud.com/v2/usg/acs/auth/appauth'

响应消息示例 HTTP/1.1 200 "X-Envoy-Upstream-Service-Time": "230", "Server": "api-gateway", "X-Request-Id": "085d1f96cd9ddd6f3c50d70a0b2eb239", "X-Content-Type-Options": "nosniff", "Connection": "keep-alive", "X-Download-Options": "noopen", "Pragma": "No-cache", "Date": "Sat, 31 Jul 2021 06:18:07 GMT", "X-Frame-Options": "SAMEORIGIN", "Strict-Transport-Security": "max-age=31536000; includeSubDomains", "Cache-Control": "no-cache", "X-Xss-Protection": "1; mode=block", "Content-Security-Policy": "connect-src 'self' *.huaweicloud.com ;style-src 'self' 'unsafe-inline' 'unsafe-eval';object-src 'self'; font-src 'self' data:;", "Expires": "Thu, 01 Jan 1970 00:00:00 GMT", "Content-Length": "1250", "Content-Type": "application/json" { "accessToken":"cnr1316vcp2ceIkbfko3z13Y2J8UdioOw0ER4kTK", "tokenIp":"49.4.112.60", "validPeriod":56326, "expireTime":1627768613, "createTime":1627712287360, "user":{ "realm":"huaweicloud.com", "userId":"53e2759d388e413abf6a56743a2694c5", "ucloginAccount":"Auto-53e2759d388e413abf6a56743a2694c5", "serviceAccount":"sip:+99111283523475338@huaweicloud.com", "numberHA1":"065eb94e5b090f70c77d4d1439f35b8e", "alias1":null, "companyId":"651543334", "spId":"8a8df0a174a1c6680174a26f578b0000", "companyDomain":null, "userType":2, "adminType":2, "name":"testuser@mycorp.com", "nameEn":"", "isBindPhone":null, "freeUser":false, "thirdAccount":"testuser@mycorp.com", "visionAccount":null, "headPictureUrl":null, "password":null, "status":0, "paidAccount":null, "paidPassword":null, "weLinkUser":false, "appId":"fdb8e4699586458bbd10c834872dcc62", "tr069Account":null, "corpType":5, "cloudUserId":"", "grayUser":true }, "clientType":72, "forceLoginInd":null, "firstLogin":false, "pwdExpired":false, "daysPwdAvailable":-19678, "proxyToken":null, "tokenType":0, "refreshToken":"cnr13168neNyRDfomYEIci7zVjBBybZQG90fYdX2", "refreshValidPeriod":2592000, "refreshExpireTime":1630304287, "refreshCreateTime":1627712287360 }

请求参数 表2 参数说明 参数 是否必须 类型 位置 描述 Authorization 是 String Header 携带应用鉴权信息。 规则：HMAC-SHA256 signature=HexEncode(HMAC256((appId + ":" + userId + ":" + expireTime + ":" + nonce), appKey)),access=base64(appId) 样例：HMAC-SHA256 signature=07f31aa9eafb06652c6899248b145c1a3264242e2ccf4c81b1b6eb99bb5c,access=ZmRiOGU0Njk5NTg2NDU4YmJkMTBjODM0ODcyZGNjNjI= 说明： 如携带了userId信息，则Body中，也需填写对应的userId信息。 （必填）鉴权头域携带access内容为对应颁发应用id进行base64编码。 X-Token-Type 是 String Header Token类型设置为：LongTicket Content-Type 是 String Header Body的媒体格式。 样例：application/json; charset=UTF-8 X-Request-ID 否 String Header 请求requestId，用来标识一路请求，用于问题跟踪定位，建议使用UUID，若不携带，则后台自动生成。 Accept-Language 否 String Header 语言参数，默认为中文zh-CN，英文为en-US。 appId 是 String Body App ID。如何获取App ID请参考“App ID的申请”。 clientType 是 Integer Body 登录账号类型。 72：API调用类型 corpId 否 String Body 企业ID。 说明： 当SP应用场景携带，如果corpId和userId字段未携带或值为空字符串时，当作SP默认管理员登录。 仅在SP模式下需要填写，单企业模式不要填写，否则会鉴权失败。 expireTime 是 Long Body 应用鉴权信息过期时间戳，单位秒。 说明： 当收到App ID鉴权请求时服务端的Unix时间戳大于expireTime时，本次鉴权失败。 样例：如果要求App ID鉴权信息10分钟后过期，expireTime = 当前Unix时间戳 + 60*10。 如果要求应用鉴权信息始终不过期，expireTime = 0。 nonce 是 String Body 随机字符串，用于计算应用鉴权信息。 minLength：32 maxLength：64 userEmail 否 String Body email地址。 userId 否 String Body 第三方用户ID。 说明： 当userId字段未携带或值为空字符串时，当作企业默认管理员登录。 userName 否 String Body 用户名称。 userPhone 否 String Body 手机号，例如中国大陆手机+86xxxxxxx deptCode 否 String Body 部门编码。通过“查询部门及其一级子部门列表”接口获取。

响应参数 表4 响应参数 参数 类型 描述 accessToken String Access Token字符串。 clientType Integer 登录账号类型。 72：API调用类型 createTime Long Access token的创建时间戳，单位：毫秒。 daysPwdAvailable Integer 密码有效天数。 delayDelete Boolean 是否延时删除状态。 expireTime Long Access Token的失效时间戳，单位：秒。 firstLogin Boolean 是否首次登录。 说明： 首次登录表示尚未修改过密码。首次登录时，系统会提醒用户需要修改密码。 默认值：false。 forceLoginInd Integer 抢占登录标识。 0：非抢占 1：抢占（未启用） proxyToken ProxyTokenDTO object 代理鉴权信息。 pwdExpired Boolean 密码是否过期。 默认值：false。 refreshCreateTime Long Refresh Token的创建时间戳，单位：毫秒。 refreshExpireTime Long Refresh Token的失效时间戳，单位：秒。 refreshToken String Refresh Token字符串。 refreshValidPeriod Long Refresh Token有效时长，单位：秒。 tokenIp String 用户IP。 tokenType Integer Token类型。 0：用户AC CES S TOKEN 1：会控TOKEN 2：一次性TOKEN user UserInfo object 用户鉴权信息。 validPeriod Long Access Token的有效时长，单位：秒。

描述 该接口使用App ID方式进行鉴权，鉴权通过后生成一个Access Token。App ID鉴权的原理介绍，请参考App ID鉴权介绍。 当clientType取值为72时，同一个userId，同时最多能创建64个Token。比如已经创建了64个Token，并且Token都在有效期内，再用同一个userId创建一个Token，前64个Token中最早创建的Token将失效。 当clientType取值为非72时，同一个userId，同时最多能创建1个Token。 Token有效期是12~24小时。

JSON-C格式 JSON-C格式与JSON格式类似，区别是对于删除操作，JSON数据放在old上，JSON-C放在data上。对于timestamp类型数据转换成yyyy-mm-dd hh:mm:ss的字符串。 JSON-C定义详情参考表5： 表5 JSON-C参数说明 参数名称 说明 mysqlType 源端表字段名称和类型。 id DRS内部定义的事件操作的序列号，单调递增。 es 源库产生这一条记录的时间，13位Unix时间戳，单位为毫秒。 ts 写入到目标kafka的时间，13位Unix时间戳，单位为毫秒。 database 数据库名称（Oracle数据库填写schema）。 table 表名。 type 操作类型，比如DELETE，UPDATE，INSERT，DDL。 isDdl 是否是DDL操作。 sql DDL的SQL语句，在DML操作中，取值为""。 sqlType 源端表字段的jdbc类型。 data 最新的数据，为JSON数组，如果type参数是插入则表示最新插入的数据，如果是更新，则表示更新后的最新数据；如果是删除，则表示被删除的数据。 old 旧数据，如果type参数是更新，则表示更新前的数据；如果是插入，取值为null。 pkNames 主键名称。

JSON格式 MySQL、 GaussDB (MySQL)到Kafka的JSON格式定义详情参考表2，DDS到Kafka的JSON格式定义详情参考表3，Oracle、PostgreSQL、GaussDB、Microsoft SQL Server到Kafka的JSON格式定义详情参考表4。 表2 MySQL到Kafka的参数说明 参数名称 说明 mysqlType 源端表字段名称和类型。 id DRS内部定义的事件操作的序列号，单调递增。 es 源库产生这一条记录的时间，13位Unix时间戳，单位为毫秒。 ts 写入到目标kafka的时间，13位Unix时间戳，单位为毫秒。 database 数据库名称。 table 表名。 type 操作类型，比如DELETE，UPDATE，INSERT，DDL，全量同步为INIT和INIT_DDL。 isDdl 是否是DDL操作。 sql DDL的SQL语句，在DML操作中，取值为""。 sqlType 源端表字段的jdbc类型。 data 最新的数据，为JSON数组，如果type参数是插入则表示最新插入的数据，如果是更新，则表示更新后的最新数据。 old 旧数据，如果type参数是更新，则表示更新前的数据；如果是删除，则表示被删除的数据；如果是插入，取值为null。 pkNames 主键名称。 { "mysqlType":{ "c11":"binary", "c10":"varchar", "c13":"text", "c12":"varbinary", "c14":"blob", "c1":"varchar", "c2":"varbinary", "c3":"int", "c4":"datetime", "c5":"timestamp", "c6":"char", "c7":"float", "c8":"double", "c9":"decimal", "id":"int" }, "id":27677, "es":1624614713000, "ts":1625058726990, "database":"test01", "table":"test ", "type":"UPDATE", "isDdl":false, "sql":"", "sqlType":{ "c11":-2, "c10":12, "c13":-1, "c12":-3, "c14":2004, "c1":12, "c2":-3, "c3":4, "c4":94, "c5":93, "c6":1, "c7":6, "c8":8, "c9":3, "id":4 }, "data":[ { "c11":"[]", "c10":"华为云huaweicloud", "c13":"asfiajhfiaf939-0239uoituqorjoqirfoidjfqrniowejoiwqjroqwjrowqjojoiqgoiegnkjgoi23roiugouofdug9u90weurtg103", "c12":"[106, 103, 111, 106, 103, 111, 105, 100, 115, 106, 103, 111, 106, 111, 115, 111, 103, 57, 51, 52, 48, 57, 52, 51, 48, 57, 116, 106, 104, 114, 103, 106, 101, 119, 57, 116, 117, 48, 57, 51, 52, 48, 116, 101, 114, 111, 101, 106, 103, 57, 56, 51, 48, 52, 105, 101, 117, 114, 103, 57, 101, 119, 117, 114, 103, 48, 119, 101, 117, 116, 57, 114, 48, 52, 117, 48, 57, 53, 116, 117, 51, 48, 57, 50, 117, 116, 48, 57, 51, 117, 116, 48, 119, 57, 101]", "c14":"[106, 103, 111, 106, 103, 111, 105, 100, 115, 106, 103, 111, 106, 111, 115, 111, 103, 57, 51, 52, 48, 57, 52, 51, 48, 57, 116, 106, 104, 114, 103, 106, 101, 119, 57, 116, 117, 48, 57, 51, 52, 48, 116, 101, 114, 111, 101, 106, 103, 57, 56, 51, 48, 52, 105, 55, 57, 56, 52, 54, 53, 52, 54, 54, 54, 49, 52, 54, 53, 33, 64, 35, 36, 37, 94, 42, 40, 41, 95, 41, 43, 95, 43, 124, 125, 34, 63, 62, 58, 58, 101, 117, 114, 103, 57, 101, 119, 117, 114, 103, 48, 119, 101, 117, 116, 57, 114, 48, 52, 117, 48, 57, 53, 116, 117, 51, 48, 57, 50, 117, 116, 48, 57, 51, 117, 116, 48, 119, 57, 101]", "c1":"cf3f70a7-7565-44b0-ae3c-83bec549ea8e:104", "c2":"[]", "c3":"103", "c4":"2021-06-25 17:51:53", "c5":"1624614713.201", "c6":"!@#$%90weurtg103", "c7":"10357.0", "c8":"1.2510357E7", "c9":"9874510357", "id":"104" } ], "old":[ { "c11":"[]", "c10":"华为云huaweicloud", "c13":"asfiajhfiaf939-0239", "c12":"[106, 103, 111, 106, 103, 111, 105, 100, 115, 106, 103, 111, 106, 111, 115, 111, 103, 57, 51, 52, 48, 57, 52, 51, 48, 57, 116, 106, 104, 114, 103, 106, 101, 119, 57, 116, 117, 48, 57, 51, 52, 48, 116, 101, 114, 111, 101, 106, 103, 57, 56, 51, 48, 52, 105, 101, 117, 114, 103, 57, 101, 119, 117, 114, 103, 48, 119, 101, 117, 116, 57, 114, 48, 52, 117, 48, 57, 53, 116, 117, 51, 48, 57, 50, 117, 116, 48, 57, 51, 117, 116, 48, 119, 57, 101]", "c14":"[106, 103, 111, 106, 103, 111, 105, 100, 115, 106, 103, 111, 106, 111, 115, 111, 103, 57, 51, 52, 48, 57, 52, 51, 48, 57, 116, 106, 104, 114, 103, 106, 101, 119, 57, 116, 117, 48, 57, 51, 52, 48, 116, 101, 114, 111, 101, 106, 103, 57, 56, 51, 48, 52, 105, 55, 57, 56, 52, 54, 53, 52, 54, 54, 54, 49, 52, 54, 53, 33, 64, 35, 36, 37, 94, 42, 40, 41, 95, 41, 43, 95, 43, 124, 125, 34, 63, 62, 58, 58, 101, 117, 114, 103, 57, 101, 119, 117, 114, 103, 48, 119, 101, 117, 116, 57, 114, 48, 52, 117, 48, 57, 53, 116, 117, 51, 48, 57, 50, 117, 116, 48, 57, 51, 117, 116, 48, 119, 57, 101]", "c1":"cf3f70a7-7565-44b0-ae3c-83bec549ea8e:104", "c2":"[]", "c3":"103", "c4":"2021-06-25 17:51:53", "c5":"1624614713.201", "c6":"!@#$%90weurtg103", "c7":"10357.0", "c8":"1.2510357E7", "c9":"9874510357", "id":"103" } ], "pkNames":[ "id" ] } 表3 DDS到Kafka的参数说明 参数名称 说明 id DRS内部定义的事件操作的序列号，单调递增。 op 操作类型，比如DELETE，UPDATE，INSERT，DDL。 dbType 源库类型：Mongo。 db 数据库名称。 coll 集合名称。 value 这一条记录的变更值。 where 这一条记录的变更条件。 recordType 具体的记录类型，比如insert、update、replace、doc。其中，update和replace表示op中的UPDATE具体操作。doc表示op中的DELETE删除的是文档数据而非视图数据。 extra 拓展字段，一般和recordType保持一致，作为扩展oplog记录使用。 es 这一条记录的commit时间，13位Unix时间戳，单位为毫秒。 ts 写入到目标kafka的时间，13位Unix时间戳，单位为毫秒。 clusterTime 与事件关联的oplog条目的时间戳，格式为timestamp:incre（timestamp是10位unix时间戳，单位为秒；incre代表当前命令在同一秒内执行的次序）。 // insert操作 { "id": 256, "op": "INSERT", "dbType": "MongoDB", "db": "ljx", "coll": "ljx", "value": "{\"_id\": ObjectId(\"64650cf67dc36a464e76e583\"), \"c1\": \"baz\", \"tags\": [\"mongodb\", \"database\", \"NoSQL\"]}", "where": null, "recordType": "insert", "extra": "insert", "es": 1684315111439, "ts": 1684315111576, "clusterTime": "1684344064:1" } // replace操作 { "id": 340, "op": "UPDATE", "dbType": "MongoDB", "db": "ljx", "coll": "ljx", "value": "{\"_id\": ObjectId(\"64650cf67dc36a464e76e583\"), \"c1\": \"sss\"}", "where": "{\"_id\": ObjectId(\"64650cf67dc36a464e76e583\")}", "recordType": "replace", "extra": "replace", "es": 1684315951831, "ts": 1684315951961, "clusterTime": "1684344904:9" } // update 更新值操作 { "id": 386, "op": "UPDATE", "dbType": "MongoDB", "db": "ljx", "coll": "ljx", "value": "{\"$set\": {\"c1\": \"aaa\"}}", "where": "{\"_id\": ObjectId(\"64650cf67dc36a464e76e583\")}", "recordType": "update", "extra": "update", "es": 1684316412008, "ts": 1684316412146, "clusterTime": "1684345365:1" } // update 更新键操作 { "id": 414, "op": "UPDATE", "dbType": "MongoDB", "db": "ljx", "coll": "ljx", "value": "{\"$unset\": {\"c1\": true}, \"$set\": {\"column1\": \"aaa\"}}", "where": "{\"_id\": ObjectId(\"64650cf67dc36a464e76e583\")}", "recordType": "update", "extra": "update", "es": 1684316692054, "ts": 1684316692184, "clusterTime": "1684345648:1" } // remove 操作 { "id": 471, "op": "DELETE", "dbType": "MongoDB", "db": "ljx", "coll": "ljx", "value": "{\"_id\": ObjectId(\"64650cf67dc36a464e76e583\")}", "where": null, "recordType": "doc", "extra": "doc", "es": 1684317252747, "ts": 1684317252869, "clusterTime": "1684346209:1" } 表4 其他数据库到Kafka的参数说明 参数名称 说明 columnType 源端表字段名称和数据类型。 说明： 数据类型不带长度、精度等。 dbType为Oracle、Microsoft SQL Server时暂为空。 dbType 源库类型。 schema schema名称。 opType 操作类型，比如DELETE，UPDATE，INSERT，DDL。 id DRS内部定义的事件操作的序列号，单调递增。 es 源库不同引擎对应类型如下： GaussDB主备版：当前事务的commit时间，13位Unix时间戳，单位为毫秒。 GaussDB分布式：当前事务的commit时间，13位Unix时间戳，单位为毫秒。 PostgreSQL：这一条记录上一个事务的commit时间，13位Unix时间戳，单位为毫秒。 Oracle：这一条记录的commit时间，13位Unix时间戳，单位为毫秒。 Microsoft SQL Server：这一条记录的commit时间，13位Unix时间戳，单位为毫秒。 ts 写入到目标kafka的时间，13位Unix时间戳，单位为毫秒。 database 数据库名称，dbType为Oracle时暂时为空。 table 表名。 type 操作类型，比如DELETE，UPDATE，INSERT，DDL。 isDdl 是否是DDL操作。 sql DDL的SQL语句，在DML操作中，取值为""。 sqlType 源端表字段的jdbc类型。 data 最新的数据，为JSON数组，如果type参数是插入则表示最新插入的数据，如果是更新，则表示更新后的最新数据。 old 旧数据，如果type参数是更新，则表示更新前的数据；如果是删除，则表示被删除的数据；如果是插入，取值为null。 pkNames 主键名称。 { "columnType": { "timestamp_column": "timestamp without time zone", "tstzrange_column": "tstzrange", "int4range_column": "int4range", "char_column": "character", "jsonb_column": "json", "boolean_column": "boolean", "bit_column": "bit", "smallint_column": "smallint", "bytea_column": "bytea" }, "dbType": "GaussDB Primary/Standby", "schema": "schema01", "opType": "UPDATE", "id": 332, "es": 1639626187000, "ts": 1639629261915, "database": "database01", "table": "table01", "type": "UPDATE", "isDdl": false, "sql": "", "sqlType": { "timestamp_column": 16, "tstzrange_column": 46, "int4range_column": 42, "char_column": 9, "jsonb_column": 22, "boolean_column": 8, "bit_column": 20, "smallint_column": 2, "bytea_column": 15 }, "data": [ { "timestamp_column": "2021-12-16 12:31:49.344365", "tstzrange_column": "(\"2010-01-01 14:30:00+08\",\"2010-01-01 15:30:00+08\")", "int4range_column": "[11,20)", "char_column": "g", "jsonb_column": "{\"key1\": \"value1\", \"key2\": \"value2\"}", "boolean_column": "false", "bit_column": "1", "smallint_column": "12", "bytea_column": "62797465615f64617461" } ], "old": [ { "timestamp_column": "2014-07-02 06:14:00.742", "tstzrange_column": "(\"2010-01-01 14:30:00+08\",\"2010-01-01 15:30:00+08\")", "int4range_column": "[11,20)", "char_column": "g", "jsonb_column": "{\"key1\": \"value1\", \"key2\": \"value2\"}", "boolean_column": "true", "bit_column": "1", "smallint_column": "12", "bytea_column": "62797465615f64617461" } ], "pkNames": null }

请求消息示例 GET /v1/mmc/management/record/downloadurls?confUUID=51adf610220411eaaae03f22d33cc26b Connection: keep-alive X-Access-Token: stbX5ElstO5QwOwPB9KGQWUZ1DHoFrTsQjjC user-agent: WeLink -desktop Host: api.meeting.huaweicloud.com User-Agent: Apache-HttpClient/4.5.3 (Java/1.8.0_191)

响应消息示例 HTTP/1.1 200 Date: Wed, 18 Dec 2019 06:20:40 GMT Content-Type: application/json;charset=UTF-8 Content-Length: 505 Connection: keep-alive Expires: 0 Pragma: No-cache Cache-Control: no-cache http_proxy_id: 4556e88832e5990723d1712395f5bee8 Server: api-gateway X-Request-Id: 629891c82bb852d8796e2f6acc74721e { "recordUrls": [ { "confUuid": "ef67f6ada67e11eba6374db4b9a61d2c", "urls": [ { "token": "f8fe906eaa6d690ef72bc831df54ffd9fc906412aefd329ace96d100cf1bc4be", "fileType": "Aux", "url": "https://100.85.230.37/download/typeThree/video/resource/00037/00037ed2-351a-4741-8ce6-a2078f21ba6b/videoAux/0_0.mp4" }, { "token": "ad8a6f6e009d643ca21f8be306e9e2cadd726360236f07bd176c1b85423b7136", "fileType": "Hd", "url": "https://100.85.230.37/download/typeThree/video/resource/00037/00037ed2-351a-4741-8ce6-a2078f21ba6b/videoHD/0_0.mp4" }, { "token": "fe7a59c69e3f97e831c83d55193a061e5e33e019f4704e5eb441c7f1fa629ad2", "fileType": "Sd", "url": "https://100.85.230.37/download/typeThree/video/resource/00037/00037ed2-351a-4741-8ce6-a2078f21ba6b/videoSD/0_0.mp4" }, { "token": "38e6b3fe7f7c62dd2141a408f4f64b911d1b58a5e04a4f6e0cfd2602181a8ad3", "fileType": "Aux", "url": "https://100.85.230.37/download/typeThree/video/resource/00037/00037ed2-351a-4741-8ce6-a2078f21ba6b/videoAux/0_1.mp4" }, { "token": "843731642aba1ebb720195a7c44f3f1e32ab409d29b2ecd1c58f30ee269f6da6", "fileType": "Hd", "url": "https://100.85.230.37/download/typeThree/video/resource/00037/00037ed2-351a-4741-8ce6-a2078f21ba6b/videoHD/0_1.mp4" }, { "token": "9fd3471e9dc047c3c345308f0cbed005f227bf48aa47875c7fe752c5a817cbd9", "fileType": "Sd", "url": "https://100.85.230.37/download/typeThree/video/resource/00037/00037ed2-351a-4741-8ce6-a2078f21ba6b/videoSD/0_1.mp4" } ] } ] }

下载示例 通过调用接口，获取到下载链接和鉴权token后，可以通过以下示例代码（Java）下载录制文件： /** * 录制文件下载 * * @param downloadUrl 录制文件下载链接 * @param localPath 本地保存路径 * @param token 下载鉴权token */ public static void httpDownload(String downloadUrl, String localPath, String token) { int byteRead; try { URL url = new URL(downloadUrl); HttpsURLConnection connection = (HttpsURLConnection) url.openConnection(); // 请求头域中携带下载鉴权token connection.setRequestProperty("Authorization", token); connection.setHostnameVerifier(new HostnameVerifier() { @Override public Boolean verify(String hostname, SSLSession sslSession) { return true; } }); TrustManager[] trustManagers = new TrustManager[]{ new X509TrustManager() { public void checkClientTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException { } public void checkServerTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException { } public X509Certificate[] getAcceptedIssuers() { return null; } } }; SSLContext ctx = SSLContext.getInstance("TLS"); ctx.init(null, trustManagers, null); connection.setSSLSocketFactory(ctx.getSocketFactory()); // 获取文件流 InputStream inStream = connection.getInputStream(); // 保存到本地路径下 FileOutputStream fs = new FileOutputStream(localPath); byte[] buffer = new byte[1024]; while ((byteRead = inStream.read(buffer)) != -1) { fs.write(buffer, 0, byteRead); } inStream.close(); fs.close(); } catch (IOException | KeyManagementException | NoSuchAlgorithmException e) { e.printStackTrace(); } }