华为云用户手册

使用模板创建测试工程 登录性能测试服务控制台，在左侧导航栏中选择“总览”，进入总览页。 PerfTest定义了如下模板： 选择一网通办系统场景，一网通办系统测试工程即创建成功。测试工程创建成功后，测试用例已按模板创建成功。 选择流媒体场景，流媒体测试工程即创建成功。测试工程创建成功后，测试用例已按模板创建成功。 选择秒杀场景，秒杀测试工程即创建成功。测试工程创建成功后，测试用例已按模板创建成功。 选择电商场景，电商测试工程即创建成功。测试工程创建成功后，测试用例已按模板创建成功。 选择极速创建场景，极速创建测试工程即创建成功。测试工程创建成功后，测试用例已按模板创建成功。 可根据实际业务需求修改测试用例，详细操作请参见PerfTest测试用例管理。

创建JMeter测试工程 登录性能测试服务控制台，在左侧导航栏中选择“JMeter测试工程”，单击“创建JMeter测试工程”。 参照表1设置基本信息。 表1 创建JMeter测试工程 参数 参数说明 工程名称 新建测试工程的名称。 说明： 工程名称只能以汉字、字母、数字、中划线（-）、下划线（_）、斜线（/）和点（.）组成，最长128个字节。 描述 新建测试工程的描述信息。 配置完成后，单击“确定”。 JMeter测试工程创建成功后，即可为测试工程添加测试计划，详细操作请参见管理JMeter测试计划。

IResultV1接口实现 package main import ( "fmt" "time" ) const ( InnerVarName = "__name" InnerVarGoroutineId = "__goroutine_id" InnerVarExecutorIndex = "__executor_index" InnerVarExecutorCount = "__executor_count" ) type IResultV1 interface { GetName() string GetUrl() string GetMethod() string GetRequestHeader() map[string]string GetRequestBody() string GetSentBytes() int GetResponseCode() int GetResponseHeader() map[string]string GetResponseBody() string GetReceivedBytes() int GetFailureMessage() string IsSuccess() bool GetBeginTime() int64 GetEndTime() int64 GetSubResults() []interface{} } //acquireResult generates root result. //just call one time on the main func and generate sub result using parent.addSub() func acquireResult(name string) *Result { result := &Result{} result.Name = name result.RequestHeader = map[string]string{} result.ResponseHeader = map[string]string{} result.ResponseCode = 200 result.Success = true result.BeginTime = time.Now().UnixMilli() result.EndTime = time.Now().UnixMilli() return result } type Result struct { Name string Url string Method string RequestHeader map[string]string RequestBody string SentBytes int ResponseCode int ResponseHeader map[string]string ResponseBody string ReceivedBytes int FailureMessage string Success bool BeginTime int64 EndTime int64 SubResults []interface{} SubIndex int } func (r *Result) GetName() string { return r.Name } func (r *Result) GetUrl() string { return r.Url } func (r *Result) GetMethod() string { return r.Method } func (r *Result) GetRequestHeader() map[string]string { return r.RequestHeader } func (r *Result) GetRequestBody() string { return r.RequestBody } func (r *Result) GetSentBytes() int { return r.SentBytes } func (r *Result) GetResponseCode() int { return r.ResponseCode } func (r *Result) GetResponseHeader() map[string]string { return r.ResponseHeader } func (r *Result) GetResponseBody() string { return r.ResponseBody } func (r *Result) GetReceivedBytes() int { return r.ReceivedBytes } func (r *Result) GetFailureMessage() string { return r.FailureMessage } func (r *Result) IsSuccess() bool { return r.Success } func (r *Result) GetBeginTime() int64 { return r.BeginTime } func (r *Result) GetEndTime() int64 { return r.EndTime } func (r *Result) GetSubResults() []interface{} { return r.SubResults } //begin records begin time, do not forget call this function to update func (r *Result) begin() { r.BeginTime = time.Now().UnixMilli() } //end records end time, do not forget call this function to update func (r *Result) end() { r.EndTime = time.Now().UnixMilli() } //addSub adds sub result to parent, call this function adding sub result always. //if name is not empty, renaming will be disabled func (r *Result) addSub(name string) *Result { if name == "" { name = fmt.Sprintf("%s-%d", r.Name, r.SubIndex) r.SubIndex++ } else { name = fmt.Sprintf("%s-%s", r.Name, name) } sub := acquireResult(name) r.SubResults = append(r.SubResults, sub) return sub } 表2 IResultV1接口结构说明表 名称 含义 补充说明 Name 请求名称 根result需使用内置参数__name设置。 Url 请求地址 - Method 方法 用于HTTP的POST，GET等。 RequestHeader 请求头 用于HTTP。 RequestBody 请求数据 建议不要记录请求较大的数据，例如上传文件，仅记录摘要即可。 SentBytes 发送字节数 - ResponseCode 响应码 记录响应状态，可以用于HTTP状态码或者自定义状态码，用于报告统计响应状态数量用于分析。 响应码范围：[100,599] ResponseHeader 响应头 用于HTTP。 ResponseBody 响应数据 建议不要记录响应较大的数据，例如下载文件，仅记录摘要即可。 ReceivedBytes 接收字节数 - FailureMessage 失败信息 - Success 是否成功 - BeginTime 请求开始时间 单位：毫秒 EndTime 请求结束时间 单位：毫秒 SubResults 子请求 当自定义请求需要多个子请求共同完成，需要使用该字段记录各个子请求执行情况。 SubIndex 子请求索引 当使用父Result的addSub方法生成子Result时，如果不为子Result自定义名称，将使用该字段自增产生索引用于生成子Result名称。

请求签名 func(map[string]string) interface{} 入参：性能测试服务内置参数以及自定义参数。 出参：interface需实现IResultV1接口，如有必要，请使用性能测试服务提供的IResultV1接口实现。 表1 性能测试服务内置参数 名称 含义 补充说明 __name 请求名称 - __goroutine_id 协程id - __executor_index 执行机索引 序号从0开始。 __executor_count 执行机总数 -

测试用例简介 PerfTest测试工程主要由测试用例、测试任务和性能报告组成。事务库和智能分析属于增强能力。 测试用例、测试任务、性能报告分别对应了性能测试活动的三个主要环节。 图1 性能测试活动流程 其中测试用例是基于某个性能压测场景建立的测试模型。测试任务是基于已定义的（若干个）测试模型发起一次性能测试的活动。性能报告是测试任务执行的结果，呈现的是被测系统在高并发场景下的性能指标。 图2 测试用例、测试任务和性能报告之间的关系 父主题： 测试用例说明

操作步骤 登录性能测试服务控制台，在左侧导航栏中选择“PerfTest测试工程”。 在待编辑PerfTest测试工程所在行，单击工程名称进入测试工程详情页面。 选择“测试用例”页签，可以在页面左侧查看当前已创建的用例目录结构。可以通过以下两种方式创建用例： 在当前目录下创建用例：直接单击“创建用例”。 图1 在当前目录下创建用例 在指定目录下创建用例：将鼠标光标放置在指定目录上，单击，在下拉框中选择“创建用例”。 图2 在指定目录下创建用例 单击“保存”。

什么是分布式消息服务RabbitMQ版 分布式消息服务RabbitMQ版完全兼容开源RabbitMQ，为您提供即开即用、消息特性丰富、灵活路由、高可用、监控和告警等特性，广泛应用于秒杀、流控、系统解耦等场景。 即开即用 分布式消息服务RabbitMQ版提供单机和集群的消息实例，拥有丰富内存规格，您可以通过控制台直接下单购买并创建，无需单独准备服务器资源。 消息特性丰富 支持AMQP协议，支持普通消息、广播消息、死信、延迟消息等特性。 灵活路由 在RabbitMQ中，生产者将消息发送到交换器，由交换器将消息路由到队列中。交换器支持Direct、Topic、Headers和Fanout四种路由方式，同时支持交换机组合和自定义。 高可用 RabbitMQ集群提供仲裁队列，在RabbitMQ节点间进行队列数据的复制，在一个节点宕机时，队列依旧可以正常运行。 监控和告警 支持对RabbitMQ实例状态进行监控，支持对集群每个代理的内存、CPU、网络流量等进行监控。如果集群或节点状态异常，将触发告警。

状态管理 函数实例的生成 创建函数实例调用句柄时，系统会自动生成新的状态实例，并加载到函数中执行。 创建函数实例调用句柄方式，各语言可参照demo模板。以下以java为例： 通过f = new Function (context, functionName) 不指定函数实例逻辑名称创建。 通过f = new Function (context, functionName, instanceName) 指定一个全新的函数实例逻辑名称 instanceName 值调用有状态函数。 函数实例调用句柄的恢复 通过指定的函数实例逻辑名称instanceName恢复获取函数实例调用句柄。 f = new Function (context) f.getInstance (functionName, instanceName) 函数实例的访问 开发者可以通过context.state访问当前函数实例绑定的状态数据值。 状态的操作 运行函数前，系统会依据函数实例调用句柄所绑定的函数实例ID自动加载状态数据到context.state中。 开发者可以通过context.state访问当前函数实例绑定的状态数据值。 通过f.saveState()接口保存状态数据的修改。 当函数执行结束后，如果调用了f.saveState()接口，系统会自动持久化状态数据。如果未调用f.saveState()接口，则调用结束之后，状态数据不会被保存。

后续操作 当函数挂载了文件系统配置后，对函数访问路径的读写就相当于对相关文件系统的读写。 如果把日志路径配置为函数访问路径的子目录，就可以轻松实现函数日志的持久化。 可以参见如下步骤，使用函数统计web服务器访问情况（应用模板），对运行在云上的服务器进行 日志分析 。 登录 函数工作流 控制台，在左侧导航栏选择“函数模板”。 在“函数模板”界面右上角搜索框中，输入“统计web服务器访问情况”进行搜索。 在搜索结果中，单击“使用模板”进入配置界面，如图2所示，请您根据实际业务进行参数配置。 参数配置完成后，单击“创建函数”，完成“统计web服务器访问情况”的函数创建。 图2 函数模板

调用API 配置请求地址相关参数。 API调用场景 API请求参数配置 使用 域名 调用API ROMA API允许使用服务分配的子域名或服务绑定的域名调用API，无需另外配置。 使用IP调用API ROMA API的实例配置参数“app_route”已设置为“on”，允许通过IP访问API。 允许使用IP地址调用API，无需另外配置。 配置认证参数。 API认证方式 API请求参数配置 APP认证 签名认证 使用获取的SDK对API请求进行签名，具体请参考APP认证开发。 简易认证 在ROMA API请求中添加Header参数“X-Apig-AppCode”，参数值为获取API的请求信息中获取到的AppCode。 双重认证 在API请求参数中增加自定义认证信息进行认证。 凭证认证 在API请求中添加Header参数“X-HW-ID”，获取API的请求信息中获取到凭证信息的Key。 在API请求中添加Header参数“X-HW-AppKey”，获取API的请求信息中获取到凭证信息的Secret。 app_api_key认证 ROMA API的实例配置参数“app_api_key”已设置为“on”，开启app_api_key认证。 在API请求中添加Header或Query参数“apikey”，参数值为获取API的请求信息中获取到的Key或AppKey。 app_secret认证 ROMA API的实例配置参数“app_secret”已设置为“on”，开启app_secret认证，且“app_api_key”已设置为“off”，关闭app_api_key认证。 在API请求中添加Header参数“X-HW-ID”，参数值为API所授权的Key或客户端的AppKey。 在API请求中添加Header参数“X-HW-AppKey”，参数值为获取API的请求信息中获取到的Secret或AppSecret。 app_basic认证 ROMA API的实例配置参数“app_basic”已设置为“on”，开启app_basic认证。 在API请求中添加Header参数“Authorization”，参数值为"Basic "+base64(appkey+":"+appsecret)，其中appkey和appsecret分别为获取API的请求信息中获取到的Key和Secret（或AppKey和AppSecret）。 app_jwt认证 ROMA API的实例配置参数“app_jwt”已设置为“on”，开启app_jwt认证。 在API请求中添加Header参数“Timestamp”，参数值为当前时间的Unix时间戳。 在API请求中添加Header参数“Authorization”，参数值为sha256(appkey+appsecret+timestamp)，其中appkey和appsecret分别为获取API的请求信息中获取到的Key和Secret（或AppKey和AppSecret），timestamp为当前时间的Unix时间戳。 华为 IAM 认证 Token认证 先获取云服务平台的认证Token，然后在API请求中携带Token进行认证。 AK/SK认证 调用API时，使用获取的SDK对API请求进行签名，具体请参考IAM认证开发（AK/SK）。 双重认证 在API请求参数中增加自定义认证信息进行认证。 自定义认证 在API请求参数中携带认证信息进行认证。 无认证 无需认证，可直接调用API。

获取API的请求消息 在调用API前，您需要向API管理者获取API的请求信息，包括访问域名、请求协议、请求方法、请求路径以及请求参数。 根据API使用的安全认证方式，还要获取相关的请求认证信息： APP认证： 签名认证：向API管理者获取该API所授权应用的Key和Secret（或客户端的AppKey和AppSecret），以及用于调用API的SDK。 简易认证：向API管理者获取该API所在应用的AppCode。 其他认证：向API管理者获取该API所授权应用的Key和Secret（或客户端的AppKey和AppSecret）或凭证信息。 华为IAM认证：通过云服务平台的账号凭证（账号和密码 获取Token 或者AK/SK）进行认证。如果使用AK/SK进行认证，还需要向API管理者获取用于调用API的SDK。 自定义认证：向API管理者获取请求参数中要携带的自定义认证信息。 无认证：无需认证信息。

交换IP须知 交换IP过程中，会自动停止在线迁移任务。 交换实例IP地址时，会有一分钟内只读和秒级的闪断。 创建的目标端实例端口需要与源实例端口保持一致。 请确保您的客户端应用具备重连机制和处理异常的能力，否则在交换IP后有可能需要重启客户端应用。 源实例和目标实例不在同一子网时，交换IP地址后，会更新实例的子网信息。 如果源端是主备实例，交换IP时不会交换备节点IP，请确保应用中没有直接引用备节点IP。 如果应用中有直接引用域名，请选择交换域名，否则域名会挂在源实例中。 请确保目标Redis和源Redis密码一致，否则交换IP后，客户端会出现密码验证错误。 当源实例配置了白名单时，则在进行IP交换前，保证目标实例也配置同样的白名单。

前提条件 准备目标实例，如果已有目标Redis，不需要重复创建，但在迁移之前，您需要清空实例数据，清空操作请参考清空Redis实例数据。 如果没有清空，如果存在与源Redis实例相同的key，迁移后，会覆盖目标Redis实例原来的数据。 创建的目标Redis与源Redis和迁移任务资源所在VPC需在同一个VPC内。 如已有的目标Redis与源Redis的VPC不在同一个VPC内，需参考如下方式打通网络，确保迁移任务的虚拟机资源能访问源Redis和目标Redis： 当源Redis和目标Redis属于华为云同一Region，请参考VPC对等连接说明，查看和创建对等连接，打通网络。 当源Redis和目标Redis属于华为云不同Region，请参考云连接，查看和创建云连接，打通网络。 创建的目标实例端口需要与源实例保持一致。 进行实例交换IP满足的条件为： 进行实例IP交换依赖的是数据迁移功能，所以，源实例及目标实例必须支持数据迁移功能，详见D CS 支持的迁移能力。 交换IP支持的能力如下表。 表1 交换ip能力 源端 目标端 单机/主备/读写分离 单机/主备/读写分离/proxy集群 Proxy集群 单机/主备/读写分离/proxy集群

前提条件 准备目标实例，如果已有目标Redis，不需要重复创建，但在迁移之前，您需要清空实例数据，清空操作请参考清空Redis实例数据。 如果没有清空，如果存在与源Redis实例相同的key，迁移后，会覆盖目标Redis实例原来的数据。 创建的目标Redis与源Redis和迁移任务资源所在VPC需在同一个VPC内。 如已有的目标Redis与源Redis的VPC不在同一个VPC内，需参考如下方式打通网络，确保迁移任务的虚拟机资源能访问源Redis和目标Redis： 当源Redis和目标Redis属于华为云同一Region，请参考VPC对等连接说明，查看和创建对等连接，打通网络。 当源Redis和目标Redis属于华为云不同Region，请参考云连接，查看和创建云连接，打通网络。 创建的目标实例端口需要与源实例保持一致。 进行实例交换IP满足的条件为： 进行实例IP交换依赖的是数据迁移功能，所以，源实例及目标实例必须支持数据迁移功能，详见DCS支持的迁移能力。 交换IP支持的能力如下表。 表1 交换ip能力 源端 目标端 单机/主备/读写分离 单机/主备/读写分离/proxy集群 Proxy集群 单机/主备/读写分离/proxy集群

交换IP须知 交换IP过程中，会自动停止在线迁移任务。 交换实例IP地址时，会有一分钟内只读和秒级的闪断。 创建的目标端实例端口需要与源实例端口保持一致。 请确保您的客户端应用具备重连机制和处理异常的能力，否则在交换IP后有可能需要重启客户端应用。 源实例和目标实例不在同一子网时，交换IP地址后，会更新实例的子网信息。 如果源端是主备实例，交换IP时不会交换备节点IP，请确保应用中没有直接引用备节点IP。 如果应用中有直接引用域名，请选择交换域名，否则域名会挂在源实例中。 请确保目标Redis和源Redis密码一致，否则交换IP后，客户端会出现密码验证错误。 当源实例配置了白名单时，则在进行IP交换前，保证目标实例也配置同样的白名单。

响应示例 状态码： 400 { "error_code" : "EG.00014000", "error_msg" : "Bad request" } 状态码： 401 { "error_code" : "EG.00014010", "error_msg" : "Incorrect token or token resolution failed" } 状态码： 403 { "error_code" : "EG.00014030", "error_msg" : "No permissions to request this method" } 状态码： 404 { "error_code" : "EG.00014040", "error_msg" : "Not request resource found" } 状态码： 500 { "error_code" : "EG.00015000", "error_msg" : "Internal Server Error" }

响应参数 状态码： 200 表3 响应Header参数 参数 参数类型 描述 X-Request-Id String This field is the request ID number for task tracking. Format is request_uuid-timestamp-hostname. 表4 响应Body参数 参数 参数类型 描述 name String 事件流名称，租户下唯一，由字母、数字、点、下划线和中划线组成，必须字母或数字开头 最小长度：1 最大长度：128 description String 事件流描述 最大长度：255 source EventStreamingSource object 事件源，一个事件流中事件源只有一个 sink EventStreamingSink object 事件目标，一个事件流中只有一个 rule_config rule_config object 事件规则，包括过滤规则和转换规则 option RunOption object 运行时配置信息 status String 事件流状态 最小长度：1 最大长度：128 枚举值： CREATED RUNNING ERROR STOPPED id String 事件流ID created_time String 创建时间 updated_time String 更新时间 表5 EventStreamingSource 参数 参数类型 描述 source_kafka SourceKafkaMQParameters object 华为云DMS版kafka类型事件源 source_mobile_rocketmq SourceMobileMQParameters object 移动云版rocketMQ类型事件源 source_community_rocketmq SourceCommunityMQParameters object 社区版rocketMQ类型事件源 source_dms_rocketmq SourceDMSMQParameters object 华为云DMS版rocketMQ事件源 name String 事件源类型名称 枚举值： HC.Kafka HC.COMMUNITY_ROCKETMQ HC.DMS_ROCKETMQ HC.MOBILE_ROCKETMQ 表6 SourceKafkaMQParameters 参数 参数类型 描述 addr String kafka连接地址 group String kafka消费组 最小长度：1 最大长度：256 instance_name String kafka实例名称 最小长度：4 最大长度：64 security_protocol String 安全协议 instance_id String kafka实例ID 最小长度：1 最大长度：256 topic String kafka topic名称 最小长度：3 最大长度：200 seek_to String 消费点位 枚举值： latest earliest enable_sasl_ssl Boolean SASL_SSL是否开启 sasl_mechanism String SASL认证机制 枚举值： SC RAM -SHA-512 PLAIN ssl_certificate_url String SASL证书地址，配置的obs地址 最小长度：1 最大长度：512 ssl_certificate_pwd String SASL证书密码 最小长度：1 最大长度：256 user_name String 用户名 最小长度：1 最大长度：256 password String 用户密码 最小长度：1 最大长度：256 表7 SourceMobileMQParameters 参数 参数类型 描述 group_id String 消费组id 最小长度：3 最大长度：128 instance_id String 实例id 最小长度：3 最大长度：128 topic String topic 最小长度：3 最大长度：128 tag String 标签 最大长度：128 authentication_required Boolean 鉴权认证 msg_trace_switch Boolean 保存消息轨迹 access_key String AccessKey 最大长度：128 secret_key String SecretKey 最大长度：128 message_model String 订阅方式 缺省值：CLUSTERING 枚举值： CLUSTERING BROADCASTING addr_type String 接入点类型 缺省值：PUBLIC 枚举值： PUBLIC PRIVATE addr String 地址 最大长度：256 sdk_url String 依赖SDK 最大长度：256 consume_timeout Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 message_type String 消息类型 枚举值： NORMAL suspend_time Integer 失败重试的等待时间 最小值：10 最大值：1800 缺省值：1800 max_reconsumer_times Integer 最大重试次数 最小值：1 最大值：16 缺省值：3 consumer_thread_nums Integer 消费线程数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size Integer 批量消费最大消息数 最小值：1 最大值：1024 缺省值：20 consumer_max_wait Integer 批量消费最大等待时长，单位：秒 最小值：0 最大值：450 缺省值：5 vpc_id String 虚拟私有云 subnet_id String 子网 表8 SourceCommunityMQParameters 参数 参数类型 描述 instance_name String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 addr String rockectMq连接地址 group String 消费组 最小长度：3 最大长度：64 topic String topic名称 最小长度：3 最大长度：64 tag String 标签 最大长度：128 vpc_id String 虚拟云id 最大长度：256 subnet_id String 子网id 最大长度：256 ssl_enable Boolean 开启SSL enable_acl Boolean ACL访问控制 access_key String 用户名 最大长度：128 secret_key String 密码 message_type String 消息类型 枚举值： normal consume_timeout Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表9 SourceDMSMQParameters 参数 参数类型 描述 instance_name String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 group String 消费组 最小长度：3 最大长度：64 topic String topic名称 最小长度：3 最大长度：64 tag String 标签 最大长度：128 ssl_enable Boolean 开启SSL enable_acl Boolean ACL访问控制 access_key String 用户名 最大长度：128 secret_key String 密码 message_type String 消费方式，针对不同生产顺序消息类型，选择消费方式会导致不同结果，请严格按照需求选择消费方式。1、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：顺序消费，实际消息处理结果：按照消息组粒度，严格保证消息顺序。 同一消息组内的消息的消费顺序和发送顺序完全一致。2、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按时间顺序处理。3、生产顺序为：未设置消息组，消息乱序发送。消费方式为：顺序消费，实际消息处理结果：按队列存储粒度，严格顺序。 基于 Apache RocketMQ 本身队列的属性，消费顺序和队列存储的顺序一致，但不保证和发送顺序一致。4、生产顺序为：未设置消息组，消息乱序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按照时间顺序处理。 枚举值： NORMAL ORDER engine_version String mq实例版本 枚举值： 4.x 5.x consume_timeout Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表10 EventStreamingSink 参数 参数类型 描述 sink_fg SinkFGParameters object 函数目标参数 sink_kafka SinkKafkaParameters object kafka目标参数 name String 事件目标类型名称 枚举值： HC.FunctionGraph HC.Kafka 表11 SinkFGParameters 参数 参数类型 描述 invoke_type String 函数执行方式,同步/异步 枚举值： SYNC，ASYNC urn String 函数链接 最小长度：1 最大长度：512 agency String 租户委托 表12 SinkKafkaParameters 参数 参数类型 描述 topic String topic名称 keyTransform Array of TransForm objects key的转换规则 connectionId String 目标连接id 表13 TransForm 参数 参数类型 描述 type String 转换规则类型 枚举值： ORIGINAL CONSTANT VARIABLE value String 常量类型规则时，字段为常量内容定义； 变量类型规则时，为变量定义，内容必须为JsonObject字符串。 变量最多支持100个，且不支持嵌套结构定义； 变量名由字母、数字、点、下划线和中划线组成，必须字母或数字开头不能以HC.开头，长度不超过64个字符； 变量值表达式支持常量或JsonPath表达式，字符串长度不超过1024个字符。 最大长度：1024 template String 变量类型规则时，规则内容的模板定义，支持对已定义变量的引用。 最大长度：2048 表14 rule_config 参数 参数类型 描述 transform TransForm object 转换规则 filter Object 过滤规则 表15 RunOption 参数 参数类型 描述 thread_num Integer 并发数 最小值：1 最大值：1000 batch_window BatchWindow object 批量推送 表16 BatchWindow 参数 参数类型 描述 count Integer 批量推送条数 最小值：1 最大值：10000 缺省值：1 time Integer 重试次数 最小值：0 最大值：15 缺省值：3 interval Integer 批量推送间隔，单位秒 最小值：0 最大值：15 缺省值：0 状态码： 400 表17 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 401 表18 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 403 表19 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 404 表20 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 500 表21 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128

响应示例 状态码： 400 { "error_code" : "EG.00014000", "error_msg" : "Bad request" } 状态码： 401 { "error_code" : "EG.00014010", "error_msg" : "Incorrect token or token resolution failed" } 状态码： 403 { "error_code" : "EG.00014030", "error_msg" : "No permissions to request this method" } 状态码： 404 { "error_code" : "EG.00014040", "error_msg" : "Not request resource found" } 状态码： 500 { "error_code" : "EG.00015000", "error_msg" : "Internal Server Error" }

响应参数 状态码： 200 表16 响应Header参数 参数 参数类型 描述 X-Request-Id String This field is the request ID number for task tracking. Format is request_uuid-timestamp-hostname. 表17 响应Body参数 参数 参数类型 描述 eventStreamingID String 事件流ID 状态码： 400 表18 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 401 表19 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 403 表20 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 404 表21 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 500 表22 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128

请求参数 表2 请求Header参数 参数 是否必选 参数类型 描述 X-Auth-Token 是 String 用户Token。通过调用IAM服务获取用户Token接口获取（响应消息头中X-Subject-Token的值）。 最小长度：1 最大长度：16384 表3 请求Body参数 参数 是否必选 参数类型 描述 name 是 String 事件流名称，租户下唯一，由字母、数字、点、下划线和中划线组成，必须字母或数字开头 最小长度：1 最大长度：128 description 否 String 事件流描述 最大长度：255 source 是 EventStreamingSource object 事件源，一个事件流中事件源只有一个 sink 是 EventStreamingSink object 事件目标，一个事件流中只有一个 rule_config 否 rule_config object 事件规则，包括过滤规则和转换规则 option 否 RunOption object 运行时配置信息 表4 EventStreamingSource 参数 是否必选 参数类型 描述 source_kafka 否 SourceKafkaMQParameters object 华为云DMS版kafka类型事件源 source_mobile_rocketmq 否 SourceMobileMQParameters object 移动云版rocketMQ类型事件源 source_community_rocketmq 否 SourceCommunityMQParameters object 社区版rocketMQ类型事件源 source_dms_rocketmq 否 SourceDMSMQParameters object 华为云DMS版rocketMQ事件源 name 否 String 事件源类型名称 枚举值： HC.Kafka HC.COMMUNITY_ROCKETMQ HC.DMS_ROCKETMQ HC.MOBILE_ROCKETMQ 表5 SourceKafkaMQParameters 参数 是否必选 参数类型 描述 addr 否 String kafka连接地址 group 是 String kafka消费组 最小长度：1 最大长度：256 instance_name 否 String kafka实例名称 最小长度：4 最大长度：64 security_protocol 否 String 安全协议 instance_id 否 String kafka实例ID 最小长度：1 最大长度：256 topic 是 String kafka topic名称 最小长度：3 最大长度：200 seek_to 否 String 消费点位 枚举值： latest earliest enable_sasl_ssl 否 Boolean SASL_SSL是否开启 sasl_mechanism 否 String SASL认证机制 枚举值： SCRAM-SHA-512 PLAIN ssl_certificate_url 否 String SASL证书地址，配置的obs地址 最小长度：1 最大长度：512 ssl_certificate_pwd 否 String SASL证书密码 最小长度：1 最大长度：256 user_name 否 String 用户名 最小长度：1 最大长度：256 password 否 String 用户密码 最小长度：1 最大长度：256 表6 SourceMobileMQParameters 参数 是否必选 参数类型 描述 group_id 是 String 消费组id 最小长度：3 最大长度：128 instance_id 是 String 实例id 最小长度：3 最大长度：128 topic 是 String topic 最小长度：3 最大长度：128 tag 否 String 标签 最大长度：128 authentication_required 否 Boolean 鉴权认证 msg_trace_switch 否 Boolean 保存消息轨迹 access_key 否 String AccessKey 最大长度：128 secret_key 否 String SecretKey 最大长度：128 message_model 是 String 订阅方式 缺省值：CLUSTERING 枚举值： CLUSTERING BROADCASTING addr_type 是 String 接入点类型 缺省值：PUBLIC 枚举值： PUBLIC PRIVATE addr 是 String 地址 最大长度：256 sdk_url 是 String 依赖SDK 最大长度：256 consume_timeout 是 Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 message_type 是 String 消息类型 枚举值： NORMAL suspend_time 否 Integer 失败重试的等待时间 最小值：10 最大值：1800 缺省值：1800 max_reconsumer_times 否 Integer 最大重试次数 最小值：1 最大值：16 缺省值：3 consumer_thread_nums 否 Integer 消费线程数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size 否 Integer 批量消费最大消息数 最小值：1 最大值：1024 缺省值：20 consumer_max_wait 否 Integer 批量消费最大等待时长，单位：秒 最小值：0 最大值：450 缺省值：5 vpc_id 否 String 虚拟私有云 subnet_id 否 String 子网 表7 SourceCommunityMQParameters 参数 是否必选 参数类型 描述 instance_name 否 String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id 否 String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 addr 是 String rockectMq连接地址 group 是 String 消费组 最小长度：3 最大长度：64 topic 是 String topic名称 最小长度：3 最大长度：64 tag 否 String 标签 最大长度：128 vpc_id 是 String 虚拟云id 最大长度：256 subnet_id 是 String 子网id 最大长度：256 ssl_enable 否 Boolean 开启SSL enable_acl 否 Boolean ACL访问控制 access_key 否 String 用户名 最大长度：128 secret_key 否 String 密码 message_type 否 String 消息类型 枚举值： normal consume_timeout 否 Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums 否 Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size 否 Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表8 SourceDMSMQParameters 参数 是否必选 参数类型 描述 instance_name 否 String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id 是 String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 group 是 String 消费组 最小长度：3 最大长度：64 topic 是 String topic名称 最小长度：3 最大长度：64 tag 否 String 标签 最大长度：128 ssl_enable 否 Boolean 开启SSL enable_acl 否 Boolean ACL访问控制 access_key 否 String 用户名 最大长度：128 secret_key 否 String 密码 message_type 否 String 消费方式，针对不同生产顺序消息类型，选择消费方式会导致不同结果，请严格按照需求选择消费方式。1、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：顺序消费，实际消息处理结果：按照消息组粒度，严格保证消息顺序。 同一消息组内的消息的消费顺序和发送顺序完全一致。2、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按时间顺序处理。3、生产顺序为：未设置消息组，消息乱序发送。消费方式为：顺序消费，实际消息处理结果：按队列存储粒度，严格顺序。 基于 Apache RocketMQ 本身队列的属性，消费顺序和队列存储的顺序一致，但不保证和发送顺序一致。4、生产顺序为：未设置消息组，消息乱序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按照时间顺序处理。 枚举值： NORMAL ORDER engine_version 否 String mq实例版本 枚举值： 4.x 5.x consume_timeout 否 Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums 否 Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size 否 Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表9 EventStreamingSink 参数 是否必选 参数类型 描述 sink_fg 否 SinkFGParameters object 函数目标参数 sink_kafka 否 SinkKafkaParameters object kafka目标参数 name 否 String 事件目标类型名称 枚举值： HC.FunctionGraph HC.Kafka 表10 SinkFGParameters 参数 是否必选 参数类型 描述 invoke_type 否 String 函数执行方式,同步/异步 枚举值： SYNC，ASYNC urn 否 String 函数链接 最小长度：1 最大长度：512 agency 否 String 租户委托 表11 SinkKafkaParameters 参数 是否必选 参数类型 描述 topic 是 String topic名称 keyTransform 否 Array of TransForm objects key的转换规则 connectionId 是 String 目标连接id 表12 TransForm 参数 是否必选 参数类型 描述 type 是 String 转换规则类型 枚举值： ORIGINAL CONSTANT VARIABLE value 否 String 常量类型规则时，字段为常量内容定义； 变量类型规则时，为变量定义，内容必须为JsonObject字符串。 变量最多支持100个，且不支持嵌套结构定义； 变量名由字母、数字、点、下划线和中划线组成，必须字母或数字开头不能以HC.开头，长度不超过64个字符； 变量值表达式支持常量或JsonPath表达式，字符串长度不超过1024个字符。 最大长度：1024 template 否 String 变量类型规则时，规则内容的模板定义，支持对已定义变量的引用。 最大长度：2048 表13 rule_config 参数 是否必选 参数类型 描述 transform 否 TransForm object 转换规则 filter 否 Object 过滤规则 表14 RunOption 参数 是否必选 参数类型 描述 thread_num 是 Integer 并发数 最小值：1 最大值：1000 batch_window 否 BatchWindow object 批量推送 表15 BatchWindow 参数 是否必选 参数类型 描述 count 否 Integer 批量推送条数 最小值：1 最大值：10000 缺省值：1 time 否 Integer 重试次数 最小值：0 最大值：15 缺省值：3 interval 否 Integer 批量推送间隔，单位秒 最小值：0 最大值：15 缺省值：0

响应示例 状态码： 400 { "error_code" : "EG.00014000", "error_msg" : "Bad request" } 状态码： 401 { "error_code" : "EG.00014010", "error_msg" : "Incorrect token or token resolution failed" } 状态码： 403 { "error_code" : "EG.00014030", "error_msg" : "No permissions to request this method" } 状态码： 404 { "error_code" : "EG.00014040", "error_msg" : "Not request resource found" } 状态码： 500 { "error_code" : "EG.00015000", "error_msg" : "Internal Server Error" }

响应参数 状态码： 200 表16 响应Header参数 参数 参数类型 描述 X-Request-Id String This field is the request ID number for task tracking. Format is request_uuid-timestamp-hostname. 表17 响应Body参数 参数 参数类型 描述 eventStreamingID String 事件流ID 状态码： 400 表18 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 401 表19 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 403 表20 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 404 表21 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 500 表22 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128

请求参数 表2 请求Header参数 参数 是否必选 参数类型 描述 X-Auth-Token 是 String 用户Token。通过调用IAM服务获取用户Token接口获取（响应消息头中X-Subject-Token的值）。 最小长度：1 最大长度：16384 表3 请求Body参数 参数 是否必选 参数类型 描述 name 是 String 事件流名称，租户下唯一，由字母、数字、点、下划线和中划线组成，必须字母或数字开头 最小长度：1 最大长度：128 description 否 String 事件流描述 最大长度：255 source 是 EventStreamingSource object 事件源，一个事件流中事件源只有一个 sink 是 EventStreamingSink object 事件目标，一个事件流中只有一个 rule_config 否 rule_config object 事件规则，包括过滤规则和转换规则 option 否 RunOption object 运行时配置信息 表4 EventStreamingSource 参数 是否必选 参数类型 描述 source_kafka 否 SourceKafkaMQParameters object 华为云DMS版kafka类型事件源 source_mobile_rocketmq 否 SourceMobileMQParameters object 移动云版rocketMQ类型事件源 source_community_rocketmq 否 SourceCommunityMQParameters object 社区版rocketMQ类型事件源 source_dms_rocketmq 否 SourceDMSMQParameters object 华为云DMS版rocketMQ事件源 name 否 String 事件源类型名称 枚举值： HC.Kafka HC.COMMUNITY_ROCKETMQ HC.DMS_ROCKETMQ HC.MOBILE_ROCKETMQ 表5 SourceKafkaMQParameters 参数 是否必选 参数类型 描述 addr 否 String kafka连接地址 group 是 String kafka消费组 最小长度：1 最大长度：256 instance_name 否 String kafka实例名称 最小长度：4 最大长度：64 security_protocol 否 String 安全协议 instance_id 否 String kafka实例ID 最小长度：1 最大长度：256 topic 是 String kafka topic名称 最小长度：3 最大长度：200 seek_to 否 String 消费点位 枚举值： latest earliest enable_sasl_ssl 否 Boolean SASL_SSL是否开启 sasl_mechanism 否 String SASL认证机制 枚举值： SCRAM-SHA-512 PLAIN ssl_certificate_url 否 String SASL证书地址，配置的obs地址 最小长度：1 最大长度：512 ssl_certificate_pwd 否 String SASL证书密码 最小长度：1 最大长度：256 user_name 否 String 用户名 最小长度：1 最大长度：256 password 否 String 用户密码 最小长度：1 最大长度：256 表6 SourceMobileMQParameters 参数 是否必选 参数类型 描述 group_id 是 String 消费组id 最小长度：3 最大长度：128 instance_id 是 String 实例id 最小长度：3 最大长度：128 topic 是 String topic 最小长度：3 最大长度：128 tag 否 String 标签 最大长度：128 authentication_required 否 Boolean 鉴权认证 msg_trace_switch 否 Boolean 保存消息轨迹 access_key 否 String AccessKey 最大长度：128 secret_key 否 String SecretKey 最大长度：128 message_model 是 String 订阅方式 缺省值：CLUSTERING 枚举值： CLUSTERING BROADCASTING addr_type 是 String 接入点类型 缺省值：PUBLIC 枚举值： PUBLIC PRIVATE addr 是 String 地址 最大长度：256 sdk_url 是 String 依赖SDK 最大长度：256 consume_timeout 是 Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 message_type 是 String 消息类型 枚举值： NORMAL suspend_time 否 Integer 失败重试的等待时间 最小值：10 最大值：1800 缺省值：1800 max_reconsumer_times 否 Integer 最大重试次数 最小值：1 最大值：16 缺省值：3 consumer_thread_nums 否 Integer 消费线程数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size 否 Integer 批量消费最大消息数 最小值：1 最大值：1024 缺省值：20 consumer_max_wait 否 Integer 批量消费最大等待时长，单位：秒 最小值：0 最大值：450 缺省值：5 vpc_id 否 String 虚拟私有云 subnet_id 否 String 子网 表7 SourceCommunityMQParameters 参数 是否必选 参数类型 描述 instance_name 否 String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id 否 String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 addr 是 String rockectMq连接地址 group 是 String 消费组 最小长度：3 最大长度：64 topic 是 String topic名称 最小长度：3 最大长度：64 tag 否 String 标签 最大长度：128 vpc_id 是 String 虚拟云id 最大长度：256 subnet_id 是 String 子网id 最大长度：256 ssl_enable 否 Boolean 开启SSL enable_acl 否 Boolean ACL访问控制 access_key 否 String 用户名 最大长度：128 secret_key 否 String 密码 message_type 否 String 消息类型 枚举值： normal consume_timeout 否 Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums 否 Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size 否 Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表8 SourceDMSMQParameters 参数 是否必选 参数类型 描述 instance_name 否 String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id 是 String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 group 是 String 消费组 最小长度：3 最大长度：64 topic 是 String topic名称 最小长度：3 最大长度：64 tag 否 String 标签 最大长度：128 ssl_enable 否 Boolean 开启SSL enable_acl 否 Boolean ACL访问控制 access_key 否 String 用户名 最大长度：128 secret_key 否 String 密码 message_type 否 String 消费方式，针对不同生产顺序消息类型，选择消费方式会导致不同结果，请严格按照需求选择消费方式。1、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：顺序消费，实际消息处理结果：按照消息组粒度，严格保证消息顺序。 同一消息组内的消息的消费顺序和发送顺序完全一致。2、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按时间顺序处理。3、生产顺序为：未设置消息组，消息乱序发送。消费方式为：顺序消费，实际消息处理结果：按队列存储粒度，严格顺序。 基于 Apache RocketMQ 本身队列的属性，消费顺序和队列存储的顺序一致，但不保证和发送顺序一致。4、生产顺序为：未设置消息组，消息乱序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按照时间顺序处理。 枚举值： NORMAL ORDER engine_version 否 String mq实例版本 枚举值： 4.x 5.x consume_timeout 否 Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums 否 Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size 否 Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表9 EventStreamingSink 参数 是否必选 参数类型 描述 sink_fg 否 SinkFGParameters object 函数目标参数 sink_kafka 否 SinkKafkaParameters object kafka目标参数 name 否 String 事件目标类型名称 枚举值： HC.FunctionGraph HC.Kafka 表10 SinkFGParameters 参数 是否必选 参数类型 描述 invoke_type 否 String 函数执行方式,同步/异步 枚举值： SYNC，ASYNC urn 否 String 函数链接 最小长度：1 最大长度：512 agency 否 String 租户委托 表11 SinkKafkaParameters 参数 是否必选 参数类型 描述 topic 是 String topic名称 keyTransform 否 Array of TransForm objects key的转换规则 connectionId 是 String 目标连接id 表12 TransForm 参数 是否必选 参数类型 描述 type 是 String 转换规则类型 枚举值： ORIGINAL CONSTANT VARIABLE value 否 String 常量类型规则时，字段为常量内容定义； 变量类型规则时，为变量定义，内容必须为JsonObject字符串。 变量最多支持100个，且不支持嵌套结构定义； 变量名由字母、数字、点、下划线和中划线组成，必须字母或数字开头不能以HC.开头，长度不超过64个字符； 变量值表达式支持常量或JsonPath表达式，字符串长度不超过1024个字符。 最大长度：1024 template 否 String 变量类型规则时，规则内容的模板定义，支持对已定义变量的引用。 最大长度：2048 表13 rule_config 参数 是否必选 参数类型 描述 transform 否 TransForm object 转换规则 filter 否 Object 过滤规则 表14 RunOption 参数 是否必选 参数类型 描述 thread_num 是 Integer 并发数 最小值：1 最大值：1000 batch_window 否 BatchWindow object 批量推送 表15 BatchWindow 参数 是否必选 参数类型 描述 count 否 Integer 批量推送条数 最小值：1 最大值：10000 缺省值：1 time 否 Integer 重试次数 最小值：0 最大值：15 缺省值：3 interval 否 Integer 批量推送间隔，单位秒 最小值：0 最大值：15 缺省值：0

响应示例 状态码： 400 { "error_code" : "EG.00014000", "error_msg" : "Bad request" } 状态码： 401 { "error_code" : "EG.00014010", "error_msg" : "Incorrect token or token resolution failed" } 状态码： 403 { "error_code" : "EG.00014030", "error_msg" : "No permissions to request this method" } 状态码： 404 { "error_code" : "EG.00014040", "error_msg" : "Not request resource found" } 状态码： 500 { "error_code" : "EG.00015000", "error_msg" : "Internal Server Error" }

响应参数 状态码： 200 表4 响应Header参数 参数 参数类型 描述 X-Request-Id String This field is the request ID number for task tracking. Format is request_uuid-timestamp-hostname. 表5 响应Body参数 参数 参数类型 描述 total Integer 总数 size Integer 本页数量 items Array of EventStreamingDetail objects 对象列表 数组长度：0 - 100 表6 EventStreamingDetail 参数 参数类型 描述 name String 事件流名称，租户下唯一，由字母、数字、点、下划线和中划线组成，必须字母或数字开头 最小长度：1 最大长度：128 description String 事件流描述 最大长度：255 source EventStreamingSource object 事件源，一个事件流中事件源只有一个 sink EventStreamingSink object 事件目标，一个事件流中只有一个 rule_config rule_config object 事件规则，包括过滤规则和转换规则 option RunOption object 运行时配置信息 status String 事件流状态 最小长度：1 最大长度：128 枚举值： CREATED RUNNING ERROR STOPPED id String 事件流ID created_time String 创建时间 updated_time String 更新时间 表7 EventStreamingSource 参数 参数类型 描述 source_kafka SourceKafkaMQParameters object 华为云DMS版kafka类型事件源 source_mobile_rocketmq SourceMobileMQParameters object 移动云版rocketMQ类型事件源 source_community_rocketmq SourceCommunityMQParameters object 社区版rocketMQ类型事件源 source_dms_rocketmq SourceDMSMQParameters object 华为云DMS版rocketMQ事件源 name String 事件源类型名称 枚举值： HC.Kafka HC.COMMUNITY_ROCKETMQ HC.DMS_ROCKETMQ HC.MOBILE_ROCKETMQ 表8 SourceKafkaMQParameters 参数 参数类型 描述 addr String kafka连接地址 group String kafka消费组 最小长度：1 最大长度：256 instance_name String kafka实例名称 最小长度：4 最大长度：64 security_protocol String 安全协议 instance_id String kafka实例ID 最小长度：1 最大长度：256 topic String kafka topic名称 最小长度：3 最大长度：200 seek_to String 消费点位 枚举值： latest earliest enable_sasl_ssl Boolean SASL_SSL是否开启 sasl_mechanism String SASL认证机制 枚举值： SCRAM-SHA-512 PLAIN ssl_certificate_url String SASL证书地址，配置的obs地址 最小长度：1 最大长度：512 ssl_certificate_pwd String SASL证书密码 最小长度：1 最大长度：256 user_name String 用户名 最小长度：1 最大长度：256 password String 用户密码 最小长度：1 最大长度：256 表9 SourceMobileMQParameters 参数 参数类型 描述 group_id String 消费组id 最小长度：3 最大长度：128 instance_id String 实例id 最小长度：3 最大长度：128 topic String topic 最小长度：3 最大长度：128 tag String 标签 最大长度：128 authentication_required Boolean 鉴权认证 msg_trace_switch Boolean 保存消息轨迹 access_key String AccessKey 最大长度：128 secret_key String SecretKey 最大长度：128 message_model String 订阅方式 缺省值：CLUSTERING 枚举值： CLUSTERING BROADCASTING addr_type String 接入点类型 缺省值：PUBLIC 枚举值： PUBLIC PRIVATE addr String 地址 最大长度：256 sdk_url String 依赖SDK 最大长度：256 consume_timeout Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 message_type String 消息类型 枚举值： NORMAL suspend_time Integer 失败重试的等待时间 最小值：10 最大值：1800 缺省值：1800 max_reconsumer_times Integer 最大重试次数 最小值：1 最大值：16 缺省值：3 consumer_thread_nums Integer 消费线程数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size Integer 批量消费最大消息数 最小值：1 最大值：1024 缺省值：20 consumer_max_wait Integer 批量消费最大等待时长，单位：秒 最小值：0 最大值：450 缺省值：5 vpc_id String 虚拟私有云 subnet_id String 子网 表10 SourceCommunityMQParameters 参数 参数类型 描述 instance_name String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 addr String rockectMq连接地址 group String 消费组 最小长度：3 最大长度：64 topic String topic名称 最小长度：3 最大长度：64 tag String 标签 最大长度：128 vpc_id String 虚拟云id 最大长度：256 subnet_id String 子网id 最大长度：256 ssl_enable Boolean 开启SSL enable_acl Boolean ACL访问控制 access_key String 用户名 最大长度：128 secret_key String 密码 message_type String 消息类型 枚举值： normal consume_timeout Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表11 SourceDMSMQParameters 参数 参数类型 描述 instance_name String 实例名称，仅dms的rockectMq需要该字段 最大长度：128 instance_id String 实例ID，仅dms的rockectMq需要该字段 最大长度：128 group String 消费组 最小长度：3 最大长度：64 topic String topic名称 最小长度：3 最大长度：64 tag String 标签 最大长度：128 ssl_enable Boolean 开启SSL enable_acl Boolean ACL访问控制 access_key String 用户名 最大长度：128 secret_key String 密码 message_type String 消费方式，针对不同生产顺序消息类型，选择消费方式会导致不同结果，请严格按照需求选择消费方式。1、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：顺序消费，实际消息处理结果：按照消息组粒度，严格保证消息顺序。 同一消息组内的消息的消费顺序和发送顺序完全一致。2、生产顺序为：设置消息组，保证消息顺序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按时间顺序处理。3、生产顺序为：未设置消息组，消息乱序发送。消费方式为：顺序消费，实际消息处理结果：按队列存储粒度，严格顺序。 基于 Apache RocketMQ 本身队列的属性，消费顺序和队列存储的顺序一致，但不保证和发送顺序一致。4、生产顺序为：未设置消息组，消息乱序发送。消费方式为：并发消费，实际消息处理结果：并发消费，尽可能按照时间顺序处理。 枚举值： NORMAL ORDER engine_version String mq实例版本 枚举值： 4.x 5.x consume_timeout Integer 消费超时时间 最小值：1000 最大值：900000 缺省值：30000 consumer_thread_nums Integer 线程消费数 最小值：20 最大值：64 缺省值：20 consumer_batch_max_size Integer 批量消费最大消息数 最小值：1 最大值：32 缺省值：1 表12 EventStreamingSink 参数 参数类型 描述 sink_fg SinkFGParameters object 函数目标参数 sink_kafka SinkKafkaParameters object kafka目标参数 name String 事件目标类型名称 枚举值： HC.FunctionGraph HC.Kafka 表13 SinkFGParameters 参数 参数类型 描述 invoke_type String 函数执行方式,同步/异步 枚举值： SYNC，ASYNC urn String 函数链接 最小长度：1 最大长度：512 agency String 租户委托 表14 SinkKafkaParameters 参数 参数类型 描述 topic String topic名称 keyTransform Array of TransForm objects key的转换规则 connectionId String 目标连接id 表15 TransForm 参数 参数类型 描述 type String 转换规则类型 枚举值： ORIGINAL CONSTANT VARIABLE value String 常量类型规则时，字段为常量内容定义； 变量类型规则时，为变量定义，内容必须为JsonObject字符串。 变量最多支持100个，且不支持嵌套结构定义； 变量名由字母、数字、点、下划线和中划线组成，必须字母或数字开头不能以HC.开头，长度不超过64个字符； 变量值表达式支持常量或JsonPath表达式，字符串长度不超过1024个字符。 最大长度：1024 template String 变量类型规则时，规则内容的模板定义，支持对已定义变量的引用。 最大长度：2048 表16 rule_config 参数 参数类型 描述 transform TransForm object 转换规则 filter Object 过滤规则 表17 RunOption 参数 参数类型 描述 thread_num Integer 并发数 最小值：1 最大值：1000 batch_window BatchWindow object 批量推送 表18 BatchWindow 参数 参数类型 描述 count Integer 批量推送条数 最小值：1 最大值：10000 缺省值：1 time Integer 重试次数 最小值：0 最大值：15 缺省值：3 interval Integer 批量推送间隔，单位秒 最小值：0 最大值：15 缺省值：0 状态码： 400 表19 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 401 表20 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 403 表21 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 404 表22 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128 状态码： 500 表23 响应Body参数 参数 参数类型 描述 error_code String 错误码 最小长度：1 最大长度：128 error_msg String 错误消息 最小长度：1 最大长度：512 error_details String 具体错误消息 最小长度：1 最大长度：1024 request_id String 请求处理ID 最小长度：1 最大长度：128

URI GET /v1/{project_id}/eventstreamings 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 租户项目ID 表2 Query参数 参数 是否必选 参数类型 描述 offset 否 Integer 偏移量，表示从此偏移量开始查询，偏移量不能小于0 最小值：0 最大值：100 缺省值：0 limit 否 Integer 每页显示的条目数量，不能小于1或大于1000 最小值：1 最大值：1000 缺省值：15

创建rf_admin_trust委托（可选） 进入华为云官网，打开控制台管理界面，鼠标移动至个人账号处，打开“ 统一身份认证 ”菜单。 图1 控制台管理界面 图2 统一身份认证菜单 进入“委托”菜单，搜索“rf_admin_trust”委托。 图3 委托列表 如果委托存在，则不用执行接下来的创建委托的步骤 如果委托不存在时执行接下来的步骤创建委托 单击步骤2界面中的“创建委托”按钮，在委托名称中输入“rf_admin_trust”，委托类型选择“云服务”，输入“ RFS ”，单击“下一步”。 图4 创建委托 在搜索框中输入“Tenant Administrator”权限，并勾选搜索结果，单击“下一步”。 图5 选择策略 选择“所有资源”，并单击“下一步“完成配置。 图6 设置授权范围 “委托”列表中出现“rf_admin_trust”委托则创建成功。 图7 委托列表

资源和成本规划 该解决方案主要部署如下资源，不同产品的花费仅供参考，实际以收费账单为准，具体请参考华为云官网价格： 表1 资源和成本规格(按需计费) 华为云服务 配置示例 每月预估花费 对象存储服务 OBS并行文件系统 区域：华北-北京四 产品类型：并行文件系统 存储空间：数据存储（多AZ存储）0.1390元/GB/月 请求费用：所有请求类型 0.0100元/万次 默认存储类别：标准存储 桶策略：私有 流量费用： 内/公网流入流量（数据上传到OBS）免费 公网流出流量 / 00:00-08:00（闲时）0.2500元/GB 公网流出流量 / 08:00-24:00（忙时）0.5000元/GB 费用包括存储空间、请求费用、流量费用三部分，具体请参考OBS计费详情。 费用包括存储空间、请求费用、流量费用三部分，详细请参考每月账单。 MapReduce服务 MRS 按需计费：30.31元/小时 区域：华北-北京四 计费模式：按需计费 集群版本：LTS集群 集群类型：分析集群 Master节点规格：ac7.4xlarge.4|16核|64GB Master系统盘：超高IO 480GB Master数据盘：超高IO 600GB Master节点数量：2个 分析Core节点规格：ac7.4xlarge.4|16核|64GB 分析Core系统盘：超高IO | 480GB 分析Core数据盘：超高IO | 600GB 分析Core节点数量：3个 购买量：1 21819.60元 合计 21819.60元+OBS费用 表2 资源和成本规格(包年包月) 华为云服务 配置示例 每月预估花费 对象存储服务 OBS并行文件系统 区域：华北-北京四 产品类型：并行文件系统 存储空间：数据存储（多AZ存储）0.1390元/GB/月 请求费用：所有请求类型 0.0100元/万次 默认存储类别：标准存储 桶策略：私有 流量费用： 内/公网流入流量（数据上传到OBS）免费 公网流出流量 / 00:00-08:00（闲时）0.2500元/GB 公网流出流量 / 08:00-24:00（忙时）0.5000元/GB 费用包括存储空间、请求费用、流量费用三部分，具体请参考OBS计费详情。 费用包括存储空间、请求费用、流量费用三部分，详细请参考每月账单。 MapReduce服务 MRS 区域：华北-北京四 计费模式：包年包月 集群版本：LTS集群 集群类型：分析集群 Master节点规格：ac7.4xlarge.4|16核|64GB Master系统盘：超高IO 480GB Master数据盘：超高IO 600GB Master节点数量：2个 分析Core节点规格：ac7.4xlarge.4|16核|64GB 分析Core系统盘：超高IO | 480GB 分析Core数据盘：超高IO | 600GB 分析Core节点数量：3个 购买量：1 16321.60元 合计 16321.60元+OBS费用

快速部署 本章节主要帮助用户快速部署该解决方案。 表1 参数填写说明 参数名称 类型 是否可选 参数解释 默认值 agency_name string 必填 委托名称，用于创建具有访问OBS权限的ECS委托。取值范围： 1-64个字符的字符串。 空 obs_bucket_name string 必填 对象存储服务 OBS并行文件系统名称，命名格式为{obs_bucket_name}-demo，用于存储数据，全局唯一。取值范围： 1-56个字符，以小写字母或数字开头，仅支持小写字母、数字、-（中划线）。禁止以中划线（-）开头。 空 登录华为云解决方案实践，选择“基于OBS配置大数据存算分离集群”解决方案。单击“一键部署”，跳转至解决方案创建堆栈界面。 图1 解决方案实施库 在选择模板界面中，单击“下一步”。 图2 选择模板 在配置参数界面中，参考表1完成自定义参数填写，单击“下一步”。 图3 配置参数 （可选，如果使用华为主账号或admin用户组下的IAM子账户可不选委托）在资源设置界面中，在权限委托下拉框中选择“rf_admin_trust”委托，单击“下一步”。 图4 资源栈设置 在配置确认页面中，单击“创建执行计划”。 图5 配置确认 在弹出的创建执行计划框中，自定义填写执行计划名称，单击“确定”。 图6 创建执行计划 待执行计划状态为“创建成功，待部署”后，单击“部署”，并且在弹出的执行计划确认框中单击“执行”。 图7 执行计划 图8 执行计划确认 等待解决方案自动部署。部署成功后，单击“事件”，回显结果如下: 图9 资源创建成功 父主题： 实施步骤

名词解释 基本概念、云服务简介、专有名词解释 弹性云服务器 ECS：是一种可随时自助获取、可弹性伸缩的云服务器，可帮助您打造可靠、安全、灵活、高效的应用环境，确保服务持久稳定运行，提升运维效率。 虚拟私有云 VPC：为云服务器、云容器、云数据库等云上资源构建隔离、私密的虚拟网络环境。VPC丰富的功能帮助您灵活管理云上网络，包括创建子网、设置安全组和网络ACL、管理路由表、申请弹性公网IP和带宽等。 弹性公网IP EIP：提供独立的公网IP资源，包括公网IP地址与公网出口带宽服务。 MapReduce服务 MRS：提供Hudi、Doris、Spark、HBase、Flink、Clickhouse、Hadoop等开源大数据组件，支持湖仓一体、灵活的弹性控制能力。