华为云用户手册

通用文件系统 通用文件系统为用户提供一个完全托管的共享文件存储，能够弹性伸缩至PB规模、TB级带宽，具备高可用性和持久性，为海量数据、高带宽型应用提供有力支持。 适用于多种应用场景，包括高性能计算、 媒体处理 、文件共享、内容管理和Web服务等，并且支持低频存储。 如表1，和传统的SFS容量型文件系统对比，通用文件系统具备更高性能。 表1 SFS容量型和通用文件系统 参数 SFS容量型 通用文件系统 最大带宽 2GB/s 1.25TB/s 最高IOPS 2K 百万 时延 3~20ms 10ms 最大容量 4PB EB 优势 大容量、高带宽、低成本 大容量、高带宽、低成本 应用场景 大容量扩展以及成本敏感型业务，如媒体处理、文件共享、高性能计算、数据备份等。 大容量扩展以及成本敏感型业务，如媒体处理、文件共享、高性能计算、数据备份等。 时延是指低负载情况下的最低延迟，非稳定时延。 10MB以上为大文件，1MB以上为大IO。 SFS容量型和通用文件系统不适合海量小文件业务，推荐使用SFS Turbo文件系统。 SFS容量型文件系统目前已售罄，请使用通用文件系统，涉及数据迁移的客户可提交工单解决。

跨Region容灾 RDS支持使用 数据复制服务 （Data Replication Service，简称DRS）创建灾备任务，当主实例所在区域发生突发性自然灾害等状况，主节点（Master）和备节点（Slave）均无法连接时，可将异地灾备实例切换为主实例，在应用端修改数据库链接地址后，即可快速恢复应用的业务访问。数据复制服务提供的实时灾备功能，可实现主实例和跨区域的灾备实例之间的单主灾备（详见“MySQL到MySQL单主灾备”）或双主灾备（详见“MySQL到MySQL双主灾备”）。

商品 商品界面详情，如图5所示。 图5 商品界面 商品界面说明，如表5所示。 表5 界面元素说明 界面元素 说明 创建商品 单击“创建商品”，弹出“创建商品”对话框，参考商品管理配置商品信息。 也可直接参考商品管理，新增商品。 单击图标后，商品界面缩小展示。单击图标，可恢复为默认展示。 商品列表 默认展示全量商品。 插入素材 单击商品中的“插入素材”，下方展示当前商品已有的贴图、视频、文本、音频内容。 需要选中具体的素材，单击“+”图标，将选中素材添加至当前场景的直播画面中。 支持编辑商品，操作如下所示： 单击图标，选择一种素材类型，新增素材。 单击图标，删除不用的素材。 在 视频直播 画面预览区域，如果存在多种视频直播素材，支持对素材进行图层管理，如：上移一层、下移一层、置顶、置底、复制、应用到全局、删除等。还支持设置大小，拖动位置。

文本 文本界面详情，如图6所示。 图6 文本界面 文本界面说明，如表6所示。 表6 界面元素说明 界面元素 说明 界面默认展示两列文本，单击图标后，缩小展示为1列文本。单击图标，可恢复为2列展示。 文本列表 系统支持的文本列表。字体分别为：鸿蒙常规、鸿蒙粗体、鸿蒙细体、鸿蒙中黑。 插入文本 将鼠标放在文本上方，文本右上角展示图标。单击图标，添加至当前场景的视频直播画面中。 在左侧的“文本设置”界面，设置文本内容和字体样式，如图7所示。 在视频直播画面预览区域，如果存在多种视频直播素材，支持对文本进行图层管理，如：上移一层、下移一层、置顶、置底、复制、应用到全局、删除等。还支持设置大小，拖动位置。 图7 文本设置

视频 视频界面详情，如图4所示。 上传后的视频不支持修改。如需修改视频内容，需本地修改完成后，再上传至控制台使用。 插入的视频不支持单独调节宽或高，仅支持宽高等比例缩放。如需调节视频的宽和高，请本地调整好后再导入至平台使用。 视频直播不支持插入透明视频，仅视频制作支持。 用户上传的MOV文件如果携带alpha通道（即透明视频），上传至MetaStudio控制台后，会转码为WEBM格式视频，保存在MetaStudio资产库。虽然视频直播的视频素材支持导入带alpha通道的MOV文件，但是导入至MetaStudio控制台后，已转码为WEBM格式。而转码后的WEBM格式视频素材，不支持插入至视频直播画面中。 图4 视频界面 视频界面说明，如表4所示。 表4 界面元素说明 界面元素 说明 本地导入 单击“本地导入”，从本地选择MOV、MP4、M4V、MKV、FLV、3GP、WMV或AVI格式的视频文件上传。 也支持直接拖拽本地视频文件至视频列表区域，完成上传操作。 界面默认展示两列视频，单击图标后，缩小展示为1列视频。单击图标，可恢复为2列展示。 关键词搜索框，可输入视频名称进行精准搜索。 示例：营销视频。 筛选条件 支持通过视频来源的维度，筛选展示下方的视频。 视频列表 默认展示全量视频。 查看视频 将鼠标放在视频上方，视频右上角展示图标。单击图标，可查看视频。 插入视频 将鼠标放在视频上方，视频右上角展示图标。单击图标，添加至当前场景的视频直播画面中。 插入的视频播放的时候默认无声音，仅有画面。 在视频直播画面预览区域，如果存在多种视频素材，支持对视频进行图层管理，如：上移一层、下移一层、置顶、置底、复制、应用到全局、删除等。还支持设置大小，拖动位置。

背景 背景界面详情，如图2所示。 上传后的背景图片不支持修改。如需修改图片内容，需本地修改完成后，再上传至控制台使用。 图2 背景界面 背景界面说明，如表2所示。 表2 界面元素说明 界面元素 说明 本地导入 单击“本地导入”，从本地选择PNG、JPG和JPEG格式的图片上传。暂不支持4K以上的图片。 也支持直接拖拽本地图片至图片列表区域，完成上传操作。 界面默认展示两列图片，单击图标后，缩小展示为1列图片。单击图标，可恢复为2列展示。 关键词搜索框，可输入图片名称进行精准搜索。 示例：科技。 筛选条件 支持通过背景、风格和布局的维度，筛选展示下方的图片。 图片列表 默认展示全量图片。选中场景，单击图片，可以替换当前场景的背景。 将鼠标放在图片上方，图片右上角展示图标。单击图标，可以将当前图片作为背景替换至所有场景。

贴图 贴图界面详情，如图3所示。 上传后的贴图不支持修改。如需修改贴图内容，需本地修改完成后，再上传至控制台使用。 如果贴图是动态GIF图片，生成预览时，可以查看动态效果。 图3 贴图界面 贴图界面说明，如表3所示。 表3 界面元素说明 界面元素 说明 本地导入 单击“本地导入”，从本地选择PNG、JPG、JPEG和BMP格式的图片上传。 也支持直接拖拽本地图片至图片列表区域，完成上传操作。 界面默认展示两列图片，单击图标后，缩小展示为1列图片。单击图标，可恢复为2列展示。 关键词搜索框，可输入图片名称进行精准搜索。 示例：小贴士。 筛选条件 支持通过贴图来源的维度，筛选展示下方的图片。 图片列表 默认展示全量图片。 查看贴图 将鼠标放在图片上方，图片右上角展示图标。单击图标，可查看大图。 插入贴图 将鼠标放在图片上方，图片右上角展示图标。单击图标，添加至当前场景的视频直播画面中。 在视频直播画面预览区域，如果存在多种视频素材，支持对贴图进行图层管理，如：上移一层、下移一层、置顶、置底、复制、应用到全局、删除等。还支持设置大小，拖动位置。

角色 角色界面详情，如图1所示。 图1 角色界面 角色界面说明，如表1所示。 表1 界面元素说明 界面元素 说明 定制数字人 单击“定制数字人”，跳转至形象制作页面，界面详情和操作指导，详见创建分身数字人定制任务。 角色界面默认展示两列数字人，单击图标后，缩小展示为1列数字人。单击图标，可恢复为2列展示。 关键词搜索框，可输入数字人名称进行精准搜索。 示例：yun lan。 刷新图标。如果有关键词，按照关键词进行刷新展示；如果没有关键词，恢复为默认展示。 筛选条件 支持通过角色、性别、姿势、行业、区域和分辨率的维度，筛选展示下方的数字人。 数字人列表 默认展示全量数字人。单击数字人形象，自动替换更新至视频直播的当前场景中。 在视频直播画面预览区域，如果存在多种素材，支持对数字人进行图层管理，如：上移一层、下移一层、置顶、置底、应用到全局、删除等。还支持设置大小，拖动位置。 将鼠标放在数字人上方，数字人右上角展示图标。单击图标，可以将数字人更新至当前视频直播的所有场景中。

风格/情感声音录制 MetaStudio支持6种情感：高兴、悲伤、生气、害怕、惊讶、安慰；6种风格：直播、教培、新闻、营销、客服、故事；外加一个通用风格，共计13种情感/风格。 费用：真人声音录制需要消耗资源，风格/情感的训练无需收费。 音频录制：每种风格/情感录制3句音频，文本样例可参考表1，也可以阅读自己熟悉的情感/风格文本，生成30秒音频。其他音频录制要求，可参考真人声音录制。 音频提交：在创建声音录制任务的时候，将真人声音录制和风格/情感录制的音频全量上传至控制台即可。 目前仅进阶版和高品质声音制作支持风格/情感训练。 表1 情感/情感声音示例 类别 情感/风格 文本 示例音频 情感 悲伤 我的心像是被撕裂了一般疼痛的难受，我的眼泪默默地滑落像是断了线的珠子。 悲伤 生气 郭宇恒你快去写作业，你为什么不能把作业写完了再看电视呢？啊？要妈妈说很多遍吗？为什么就不听呢？ 生气 高兴 当录取通知书到手，满心欢喜简直要溢出来啦，好开心呀，终于可以准备去期待已久旅行啦。 高兴 惊讶 诶，好奇怪，我打开冰箱门发现，昨天放进去的两个蛋糕好像不见了？蛋糕怎么会平白无故没了呢？ 惊讶 安慰 别难过啦，生活就像天气，有阴也有晴。当下的挫折只是暂时的，往后定有美好在等你。 安慰 恐惧 我的内心充满了恐惧，仿佛有双手、正在勒紧我的脖子我都无法呼吸了。 恐惧 风格 直播 宝子们，福利来啦！今天这款人气单品，现在下单立享五折优惠，库存有限，手慢无哦 直播 新闻 今日，旨在改善民生的新政策正式实施，从住房到医疗，将惠及众多不同年龄段的市民群体。 新闻 故事 很久很久以前，在一片古老森林深处，矗立着一座神秘的城堡，传说其中藏有无尽的宝藏。 故事 教培 同学们，遇到这类难题，这个解题思路很关键，仔细听，掌握了它，同类题都能迎刃而解。 教培 营销 家人们注意！此次特惠商品数量有限，先到先得，如此超值好物，赶紧抢购别错过！ 营销 客服 亲，您反馈的问题我们已收到，售后团队马上为您核实处理，定会给您满意答复 。 客服 通用 中性 静静地躺在草地上、看着天空中悠闲的云朵，聆听着轻柔的音乐。 中性 父主题： 声音制作

请求参数 表2 请求Header参数 参数 是否必选 参数类型 描述 X-Auth-Token 是 String 用户Token。 用于获取操作API的权限。获取方法请参见认证鉴权。 Content-Type 是 String 发送的实体的MIME类型，参数值为“application/json”。 Enterprise-Project-Id 否 String 企业项目ID。FRS支持通过企业项目管理（EPS）对不同用户组和用户的资源使用，进行分账，当前仅支持按需计费模式。 获取方法：进入“企业项目管理”页面，单击企业项目名称，在企业项目详情页获取Enterprise-Project-Id（企业项目ID）。 说明： 创建企业项目后，在传参时，有以下三类场景。 携带正确的ID，正常使用FRS服务，账单的企业项目会被分类到企业ID对应的企业项目中。 携带格式正确但不存在的ID，正常使用FRS服务，账单的企业项目会显示对应不存在的企业项目ID。 不携带ID或格式错误ID（包含特殊字符等），正常使用FRS服务，账单的企业项目会被分类到"default"中。 表3 请求Body参数 参数名 参数类型 是否必选 说明 image_url String 与image_file、image_base64三选一 图片的URL路径，目前仅支持华为云上OBS的URL，使用时只需保证FRS有权限读取该OBS桶的数据。开通读取权限的操作请参见服务授权。 image_file File 与image_url、image_base64三选一 本地图片文件，图片不能超过8MB，建议小于1MB。上传文件时，请求格式为multipart。 image_base64 String 与image_file、image_url三选一 图像数据，Base64编码，要求： Base64编码后大小不超过8MB，建议小于1MB。 图片为JPG/JPEG/BMP/PNG格式。 external_image_id String 否 用户指定的图片外部ID，与当前图像绑定。用户没提供，系统会生成一个。 该ID长度范围为1～36位，可以包含字母、数字、中划线或者下划线，不包含其他的特殊字符。 external_fields Object 否 根据用户自定义数据类型，填入相应的数值。 需在创建人脸库时定义external_fields字段，才可以在添加人脸时使用该字段，参考自定义字段。 single boolean 否 是否将图片中的最大人脸添加至人脸库。可选值包括： true：传入的单张图片中如果包含多张人脸，则只将最大人脸添加到人脸库中。 false：默认为false。传入的单张图片中如果包含多张人脸，则将所有人脸添加至人脸库中。

功能介绍 添加人脸到人脸库中。将单张图片中的人脸添加至人脸库中，支持添加最大人脸或所有人脸。 前提条件： 请确保您已开通人脸搜索服务。 约束限制： 只支持识别JPG、PNG、JPEG、BMP格式的图片。 application/json请求的body中，请使用标准Json格式。 Base64编码中请勿使用回车换行。 系统不保存用户图片。 图片大小小于8MB，由于过大图片会导致时延较长，并且图片信息量不大，建议小于1MB。 图片分辨率小于4096*4096，图片中人脸像素大于80*80，建议120*120以上。 为保证识别效果，人脸图片建议要求如下： 光照大于200lux、无反光强光阴影现象。 人脸无遮挡、整体清晰无拖尾抖动等运动模糊。 侧脸不超过30°、俯仰角小于15°、偏转角小于15°、图片中人脸保持竖置正脸。 其他的约束限制信息请参见约束与限制章节。 建议： 由于过大图片对识别算法精度无明显提升，同时会导致时延较长，建议传入图片小于1MB，一般500KB左右足够。 OBS上存储的图片也建议小于1MB。

响应参数 状态码：200 表4 响应Body参数 参数 参数类型 描述 face_set_id String 人脸库ID。 调用失败时无此字段。 face_set_name String 人脸库名称。 调用失败时无此字段。 faces Array of FaceSetFace objects 人脸库当中的人脸结构，详见FaceSetFace。 调用失败时无此字段。 表5 FaceSetFace 参数 参数类型 描述 bounding_box BoundingBox object 人脸在图像中的位置。 BoundingBox结构见BoundingBox。 external_fields Object 用户添加的额外字段。 external_image_id String 人脸所在的外部图片ID。 face_id String 人脸ID，由系统内部生成的唯一ID。 表6 BoundingBox 参数 参数类型 描述 width Integer 矩形框宽度。 top_left_y Integer 矩形框左上角纵坐标。 top_left_x Integer 矩形框左上角横坐标。 height Integer 矩形框高度。 状态码： 400 表7 响应Body参数 参数 参数类型 描述 error_code String 调用失败时的错误码，具体请参考错误码。 调用成功时无此字段。 error_msg String 调用失败时的错误信息。 调用成功时无此字段。

通过SQL API 访问数据源 创建 DLI 跨源访问 dws 的关联表。 1 2 3 4 5 6 7 sparkSession.sql( "CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS ( 'url'='jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres',\ 'dbtable'='customer',\ 'user'='dbadmin',\ 'password'='######',\ 'driver'='org.postgresql.Driver')") 建表参数详情可参考表1。 插入数据 1 sparkSession.sql("insert into dli_to_dws values(2,'John',24)") 查询数据 1 jdbcDF = sparkSession.sql("select * from dli_to_dws").show() 操作结果

完整示例代码 通过DataFrame API访问 认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 # _*_ coding: utf-8 _*_ from __future__ import print_function from pyspark.sql.types import StructType, StructField, IntegerType, StringType from pyspark.sql import SparkSession if __name__ == "__main__": # Create a SparkSession session. sparkSession = SparkSession.builder.appName("datasource-dws").getOrCreate() # Set cross-source connection parameters url = "jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres" dbtable = "customer" user = "dbadmin" password = "######" driver = "org.postgresql.Driver" # Create a DataFrame and initialize the DataFrame data. dataList = sparkSession.sparkContext.parallelize([(1, "Katie", 19)]) # Setting schema schema = StructType([StructField("id", IntegerType(), False),\ StructField("name", StringType(), False),\ StructField("age", IntegerType(), False)]) # Create a DataFrame from RDD and schema dataFrame = sparkSession.createDataFrame(dataList, schema) # Write data to the DWS table dataFrame.write \ .format("jdbc") \ .option("url", url) \ .option("dbtable", dbtable) \ .option("user", user) \ .option("password", password) \ .option("driver", driver) \ .mode("Overwrite") \ .save() # Read data jdbcDF = sparkSession.read \ .format("jdbc") \ .option("url", url) \ .option("dbtable", dbtable) \ .option("user", user) \ .option("password", password) \ .option("driver", driver) \ .load() jdbcDF.show() # close session sparkSession.stop() 通过SQL API访问 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 # _*_ coding: utf-8 _*_ from __future__ import print_function from pyspark.sql import SparkSession if __name__ == "__main__": # Create a SparkSession session. sparkSession = SparkSession.builder.appName("datasource-dws").getOrCreate() # Createa data table for DLI - associated DWS sparkSession.sql( "CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS (\ 'url'='jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres',\ 'dbtable'='customer',\ 'user'='dbadmin',\ 'password'='######',\ 'driver'='org.postgresql.Driver')") # Insert data into the DLI data table sparkSession.sql("insert into dli_to_dws values(2,'John',24)") # Read data from DLI data table jdbcDF = sparkSession.sql("select * from dli_to_dws").show() # close session sparkSession.stop()

操作前准备 import相关依赖包 1 2 3 from __future__ import print_function from pyspark.sql.types import StructType, StructField, IntegerType, StringType from pyspark.sql import SparkSession 创建会话 1 sparkSession = SparkSession.builder.appName("datasource-dws").getOrCreate()

通过DataFrame API访问数据源 连接参数配置 1 2 3 4 5 url = "jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres" dbtable = "customer" user = "dbadmin" password = "######" driver = "org.postgresql.Driver" 设置数据 1 dataList = sparkSession.sparkContext.parallelize([(1, "Katie", 19)]) 设置schema 1 2 3 schema = StructType([StructField("id", IntegerType(), False),\ StructField("name", StringType(), False),\ StructField("age", IntegerType(), False)]) 创建DataFrame 1 dataFrame = sparkSession.createDataFrame(dataList, schema) 保存数据到DWS 1 2 3 4 5 6 7 8 9 dataFrame.write \ .format("jdbc") \ .option("url", url) \ .option("dbtable", dbtable) \ .option("user", user) \ .option("password", password) \ .option("driver", driver) \ .mode("Overwrite") \ .save() mode 有四种保存类型： ErrorIfExis：如果已经存在数据，则抛出异常。 Overwrite：如果已经存在数据，则覆盖原数据。 Append：如果已经存在数据，则追加保存。 Ignore：如果已经存在数据，则不做操作。这类似于SQL中的“如果不存在则创建表”。 读取DWS上的数据 1 2 3 4 5 6 7 8 9 jdbcDF = sparkSession.read \ .format("jdbc") \ .option("url", url) \ .option("dbtable", dbtable) \ .option("user", user) \ .option("password", password) \ .option("driver", driver) \ .load() jdbcDF.show() 操作结果

完整示例代码 通过SQL API 访问DWS表 import org.apache.spark.sql.SparkSession; public class java_dws { public static void main(String[] args) { SparkSession sparkSession = SparkSession.builder().appName("datasource-dws").getOrCreate(); sparkSession.sql("CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS ('url'='jdbc:postgresql://10.0.0.233:8000/postgres','dbtable'='test','user'='dbadmin','password'='**')"); //*****************************SQL model*********************************** //Insert data into the DLI data table sparkSession.sql("insert into dli_to_dws values(3,'Liu'),(4,'Xie')"); //Read data from DLI data table sparkSession.sql("select * from dli_to_dws").show(); //drop table sparkSession.sql("drop table dli_to_dws"); sparkSession.close(); } }

通过SQL API 访问数据源 创建DLI跨源访问DWS的关联表，填写连接参数。 1 sparkSession.sql("CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS ('url'='jdbc:postgresql://10.0.0.233:8000/postgres','dbtable'='test','user'='dbadmin','password'='**')"); 插入数据 1 sparkSession.sql("insert into dli_to_dws values(3,'Liu'),(4,'Xie')"); 查询数据 1 sparkSession.sql("select * from dli_to_dws").show(); 插入数据后：

整体作业开发流程 整体作业开发流程参考图1。 图1 作业开发流程 步骤1：创建队列：创建DLI作业运行的队列。 步骤2：创建Kafka的Topic：创建Kafka生产消费数据的Topic。 步骤3：创建RDS数据库和表：创建RDS MySQL数据库和表信息。 步骤4：创建增强型跨源连接：DLI上创建连接Kafka和RDS的跨源连接，打通网络。 步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。 步骤6：发送数据和查询结果：Kafka上发送流数据，在RDS上查看运行结果。

场景描述 该场景为根据商品的实时点击量，获取每小时内点击量最高的3个商品及其相关信息。商品的实时点击量数据为输入源发送到Kafka中，再将Kafka数据的分析结果输出到RDS中。 例如，输入如下样例数据： {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 08:01:00", "product_id":"0002", "product_name":"name1"} {"user_id":"0002", "user_name":"Bob", "event_time":"2021-03-24 08:02:00", "product_id":"0002", "product_name":"name1"} {"user_id":"0002", "user_name":"Bob", "event_time":"2021-03-24 08:06:00", "product_id":"0004", "product_name":"name2"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 08:10:00", "product_id":"0003", "product_name":"name3"} {"user_id":"0003", "user_name":"Cindy", "event_time":"2021-03-24 08:15:00", "product_id":"0005", "product_name":"name4"} {"user_id":"0003", "user_name":"Cindy", "event_time":"2021-03-24 08:16:00", "product_id":"0005", "product_name":"name4"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 08:56:00", "product_id":"0004", "product_name":"name2"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 09:05:00", "product_id":"0005", "product_name":"name4"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 09:10:00", "product_id":"0006", "product_name":"name5"} {"user_id":"0002", "user_name":"Bob", "event_time":"2021-03-24 09:13:00", "product_id":"0006", "product_name":"name5"} 预期输出： 2021-03-24 08:00:00 - 2021-03-24 08:59:59,0002,name1,2 2021-03-24 08:00:00 - 2021-03-24 08:59:59,0004,name2,2 2021-03-24 08:00:00 - 2021-03-24 08:59:59,0005,name4,2 2021-03-24 09:00:00 - 2021-03-24 09:59:59,0006,name5,2 2021-03-24 09:00:00 - 2021-03-24 09:59:59,0005,name4,1

步骤6：发送数据和查询结果 使用Kafka客户端向步骤2：创建Kafka的Topic中的Topic发送数据，模拟实时数据流。 Kafka生产和发送数据的方法请参考：DMS - 连接实例生产消费信息。 发送样例数据如下： {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 08:01:00", "product_id":"0002", "product_name":"name1"} {"user_id":"0002", "user_name":"Bob", "event_time":"2021-03-24 08:02:00", "product_id":"0002", "product_name":"name1"} {"user_id":"0002", "user_name":"Bob", "event_time":"2021-03-24 08:06:00", "product_id":"0004", "product_name":"name2"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 08:10:00", "product_id":"0003", "product_name":"name3"} {"user_id":"0003", "user_name":"Cindy", "event_time":"2021-03-24 08:15:00", "product_id":"0005", "product_name":"name4"} {"user_id":"0003", "user_name":"Cindy", "event_time":"2021-03-24 08:16:00", "product_id":"0005", "product_name":"name4"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 08:56:00", "product_id":"0004", "product_name":"name2"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 09:05:00", "product_id":"0005", "product_name":"name4"} {"user_id":"0001", "user_name":"Alice", "event_time":"2021-03-24 09:10:00", "product_id":"0006", "product_name":"name5"} {"user_id":"0002", "user_name":"Bob", "event_time":"2021-03-24 09:13:00", "product_id":"0006", "product_name":"name5"} 登录RDS控制台，单击RDS数据库实例，单击创建的数据库名，如“testrdsdb”，在创建的表“clicktop”所在行的“操作”列，单击“SQL查询”，输入以下查询语句。 select * from `clicktop`; 在“SQL查询”界面，单击“执行SQL”，查看RDS表数据已写入成功。 图2 RDS表数据

完整示例代码 通过SQL API访问 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 import org.apache.spark.sql.SparkSession; public class java_rds { public static void main(String[] args) { SparkSession sparkSession = SparkSession.builder().appName("datasource-rds").getOrCreate(); // Create a data table for DLI-associated RDS sparkSession.sql("CREATE TABLE IF NOT EXISTS dli_to_rds USING JDBC OPTIONS ('url'='jdbc:mysql://192.168.6.150:3306','dbtable'='test.customer','user'='root','password'='**','driver'='com.mysql.jdbc.Driver')"); //*****************************SQL model*********************************** //Insert data into the DLI data table sparkSession.sql("insert into dli_to_rds values(3,'Liu',21),(4,'Joey',34)"); //Read data from DLI data table sparkSession.sql("select * from dli_to_rds").show(); //drop table sparkSession.sql("drop table dli_to_rds"); sparkSession.close(); } }

整体作业开发流程 整体作业开发流程参考图1。 图1 作业开发流程 步骤1：创建队列：创建DLI作业运行的队列。 步骤2：创建RDS MySQL数据库和表：创建RDS MySQL的数据库和表。 步骤3：创建DWS数据库和表：创建用于接收数据的DWS数据库和表。 步骤4：创建增强型跨源连接：DLI上创建连接RDS和DWS的跨源连接，打通网络。 步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。 步骤6：发送数据和查询结果：RDS MySQL的表上插入数据，在DWS上查看运行结果。

步骤3：创建DWS数据库和表 连接已创建的DWS集群。 请参考使用gsql命令行客户端连接DWS集群。 执行以下命令连接DWS集群的默认数据库“gaussdb”： gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r gaussdb：DWS集群默认数据库。 DWS集群连接地址：请参见获取集群连接地址进行获取。如果通过公网地址连接，请指定为集群“公网访问地址”或“公网访问 域名 ”，如果通过内网地址连接，请指定为集群“内网访问地址”或“内网访问域名”。如果通过弹性负载均衡连接，请指定为“弹性负载均衡地址”。 dbadmin：创建集群时设置的默认管理员用户名。 -W：默认管理员用户的密码。 在命令行窗口输入以下命令创建数据库“testdwsdb”。 CREATE DATABASE testdwsdb; 执行以下命令，退出gaussdb数据库，连接新创建的数据库“testdwsdb”。 \q gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r 执行以下命令创建表。 create schema test; set current_schema= test; drop table if exists dwsresult; CREATE TABLE dwsresult ( car_id VARCHAR, car_owner VARCHAR, car_age INTEGER , average_speed FLOAT8, total_miles FLOAT8 );

约束限制 仅支持Flink1.15版本使用委托授权访问临时凭证： 在创建作业时，请配置作业使用Flink1.15版本 已在作业中配置允许DLI访问DEW的委托信息。flink.dli.job.agency.name=自定义委托名称。 自定义委托请参考自定义DLI委托权限。 请注意配置参数不需要用"" 或 '' 包裹。 Flink1.15基础镜像内置了3.1.62版本的huaweicloud-sdk-core。

功能描述 DLI提供了一个通用接口，可用于获取用户在启动Flink作业时设置的委托的临时凭证。该接口将获取到的该作业委托的临时凭证封装到com.huaweicloud.sdk.core.auth.BasicCredentials类中。 获取到的委托的临时认证封装到com.huaweicloud.sdk.core.auth.ICredentialProvider接口的getCredentials()返回值中。 返回类型为com.huaweicloud.sdk.core.auth.BasicCredentials。 仅支持获取AK、SK、SecurityToken。 获取到AK、SK、SecurityToken后，请参考如何使用凭据管理服务替换硬编码的数据库账号密码查询凭据。

监控指标 指标ID 指标名称 指标含义 取值范围 单位 进制 测量对象（维度） 监控周期（原始指标） successful_calls_of_service 成功调用次数 统计周期内调用服务成功次数。 单位：次/分钟 ≥ 0 times/min 不涉及 接口 1分钟 api_calls_with_4xx_response 4xx失败调用次数 该指标用于统计接口的4xx失败调用次数。 单位：次/分钟 ≥ 0 times/min 不涉及 接口 1分钟 api_calls_with_5xx_response 5xx失败调用次数 该指标用于统计接口的5xx失败调用次数。 单位：次/分钟 ≥ 0 times/min 不涉及 接口 1分钟 average_delay 平均时延 该指标用于统计接口的平均时延。 单位：毫秒 ≥ 0 ms 1000(SI) 接口 1分钟

监控指标详情 表1 OCR支持的监控指标 指标ID 指标名称 指标含义 取值范围 单位 进制 测量对象（维度） 监控周期（原始指标） successful_call_times_of_service 调用服务成功次数 统计周期内调用服务成功次数。 ≥ 0 times/min 不涉及 接口 1分钟 failed_call_times_of_service 调用服务失败次数 统计周期内调用服务失败次数。 ≥ 0 times/min 不涉及 接口 1分钟

使用限制 单账号跟踪的事件可以通过 云审计 控制台查询。多账号的事件只能在账号自己的事件列表页面去查看，或者到组织追踪器配置的OBS桶中查看，也可以到组织追踪器配置的 CTS /system日志流下面去查看。 用户通过云审计控制台只能查询最近7天的操作记录。如果需要查询超过7天的操作记录，您必须配置转储到 对象存储服务 （OBS）或 云日志 服务（LTS），才可在OBS桶或LTS日志组里面查看历史事件信息。否则，您将无法追溯7天以前的操作记录。 云上操作后，1分钟内可以通过云审计控制台查询管理类事件操作记录，5分钟后才可通过云审计控制台查询数据类事件操作记录。 CTS新版事件列表不显示数据类审计事件，您需要在旧版事件列表查看数据类审计事件。 云审计控制台对用户的操作事件日志保留7天，过期自动删除，不支持人工删除。

验证通用可用区和用户本地系统之间的网络连通性 当配置了通用可用区和用户本地系统之间的路由信息时，即在表1中的“目的地址”参数填写为虚拟私有云的大网段，或者又单独新增了对应子网网段的路由信息，您可以在本地系统通过VPC内网访问通用可用区的云服务。 以windows弹性云服务器为例进行说明： 在CloudPond通用可用区新建一台弹性 云服务器ECS 01（加入VPC01，选择子网网段10.0.0.0/16）。 参数“可用区”选择“通用可用区”。 方法请参见自定义购买弹性云服务器。 将E CS 01所在的安全组放通“入方向”的“ICMP”规则，出方向默认全部放通。详见安全组配置规则。 在弹性云服务器列表页面查询到ECS01，从“IP地址”列获取并记录其私有IP地址（如10.0.1.110）。 登录本地系统服务器Server01，执行到弹性云服务器ECS01的ping命令，验证Server01和ECS01的网络连通性。 ping 10.0.1.110 检查ping命令返回结果，可以ping通表示网络可连通。如果ping不通请排查原因并联系华为云运维团队处理。