# 使用DLI Flink作业实时同步Kafka数据至DWS集群 本实践演示通过**数据湖探索服务 DLI** Flink作业（以Flink 1.15版本为例）将**分布式消息服务 Kafka**的消费数据实时同步至DWS数据仓库，实现Kafka实时入库到DWS的过程。演示过程包括实时写入和更新已有数据的场景。 - 了解DLI请参见[数据湖产品介绍](https://support.huaweicloud.com/productdesc-dli/dli_01_0378.html)。 - 了解Kafka请参见[分布式消息服务Kafka产品介绍](https://support.huaweicloud.com/productdesc-kafka/kafka-pd-190605001.html)。 图1Kafka实时入库DWS ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001644891425.png "点击放大") 本实践预计时长90分钟，实践用到的云服务包括**虚拟私有云 VPC及子网** 、**弹性负载均衡 ELB** 、**弹性云服务器 ECS** 、**对象存储服务 OBS、分布式消息服务 Kafka** 、**数据湖探索 DLI** 和**数据仓库服务 DWS，**基本流程如下： 1. [准备工作]：注册账号，准备网络等。 2. [步骤一：创建Kafka实例]：购买分布式消息服务Kafka版，并准备Kafka数据。 3. [步骤二：创建绑定ELB的DWS集群和目标表]：购买DWS集群并绑定EIP。 4. [步骤三：创建DLI弹性资源池及队列]：创建DLI弹性资源池，并为资源池添加队列。 5. [步骤四：创建Kafka和DWS的增强型跨源连接]：用于连接Kafka和DWS。 6. [步骤五：准备DWS对接Flink工具dws-connector-flink]：使用该插件，提高Kafka入库DWS的性能。 7. [步骤六：创建并编辑DLI Flink作业]：创建Flink SQL作业，配置SQL代码。 8. [步骤七：通过Kafka客户端生产和修改消息]：完成数据实时入仓。 #### 场景描述 假设数据源Kafka的样例数据是一个用户信息表，如[表1]所示， 包含 id，name，age三个字段。其中id是唯一且固定的字段，多个业务系统会共用，业务上一般不需要修改，仅修改姓名name，年龄age。 首先，通过Kafka生产以下三组数据，通过DLI Flink作业完成数据同步到数据仓库服务 DWS。接着，需要修改id为2和3的用户为新的jim和tom，再通过DLI Flink作业完成数据的更新并同步到DWS。 表1样例数据 | id | name | age | |:---|:---|:---| | 1 | lily | 16 | | 2 | lucy \> jim | 17 | | 3 | lilei \> tom | 15 | #### 约束限制 - 确保VPC、ECS、OBS、Kafka、DLI和DWS服务在同一个区域内，例如华北-北京四。 - 确保Kafka、DLI、DWS网络互通。本实践将Kafka和DWS创建在同一个区域和虚拟私有云下，同时在Kafka和DWS的安全组中放通了DLI的队列所在网段，确保网络互通。 - 为确保DLI到DWS的连接链路稳定，请创建完DWS集群后为集群绑定ELB服务。 #### 准备工作 - 已注册华为云账号，具体请参见[注册华为云账号](https://support.huaweicloud.com/usermanual-account/account_id_001.html)，且在使用DWS 前检查账号状态，账号不能处于欠费或冻结状态。 - 已创建虚拟私有云和子网，参见[创建虚拟私有云和子网](https://support.huaweicloud.com/usermanual-vpc/zh-cn_topic_0013935842.html)。 #### 步骤一：创建Kafka实例 1. 登录[Kafka管理控制台购买页面](https://console.huaweicloud.com/dms/?engine=kafka&locale=zh-cn#/queue/newCreateKafkaV2)。 2. 关键参数如下表说明，其他参数项如表中未说明，默认即可： 表2kafka实例参数 | 参数项 | 参数值 | |:---|:---| | 计费模式 | 按需计费 | | 区域 | 华北-北京四 | | 可用区 | 可用区1（如遇售罄，选择其他可用区） | | 套餐规格 | 入门规格 | | 虚拟私有云 | 选择已创建的虚拟私有云，如果没有，则需要创建。 | | 安全组 | 选择已创建的安全组，如果没有，则需要创建。 | | 其他参数 | 保持默认 | 3. 单击"确认订单"，核对无误，单击"提交"。等待创建成功。 4. 创建成功后，在Kafka实例列表中，单击创建好的Kafka实例名称，进入基本信息页面。 5. 左侧选择"Topic管理"，单击"创建Topic"。 Topic名称设置为"topic-demo"，其他可保持默认。 图2创建Topic ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001638370685.png "点击放大") 6. 单击"确定"，在Topic列表中可以看到topic-demo已创建成功。 7. 左侧导航栏选择"消费组管理"，单击"创建消费组"。 8. 消费组名称输入"kafka01"，单击"确定"。 #### 步骤二：创建绑定ELB的DWS集群和目标表 1. [创建独享型弹性负载均衡服务ELB](https://support.huaweicloud.com/usermanual-elb/elb_lb_000006.html)，网络类型选择IPv4私网即可，区域、VPC选择与Kafka实例保持一致，本实践为"华北-北京四"。 2. [创建集群](https://support.huaweicloud.com/mgtg-dws/dws_01_0019.html)，为DWS绑定弹性负载均衡 ELB，同时为确保网络连通，DWS集群的区域、VPC选择与Kafka实例保持一致，本实践为"华北-北京四"，虚拟私有云与上面创建Kafka的虚拟私有云保持一致。 3. 在[DWS控制台](https://console.huaweicloud.com/dws)的"集群 \> 集群列表"页面，单击指定集群所在行操作列的"登录"按钮。 ![](https://support.huaweicloud.com/bestpractice-dws/public_sys-resources/note_3.0-zh-cn.png) 本实践以8.1.3.x版本为例，8.1.2及以前版本不支持此登录方式，可以使用gsql登录。 4. 登录成功后，进入SQL编辑器。 5. 复制如下SQL语句，在SQL窗口中，单击"执行SQL"，创建目标表user_dws。 ``` CREATE TABLE user_dws ( id int, name varchar(50), age int, PRIMARY KEY (id) ); ``` #### 步骤三：创建DLI弹性资源池及队列 1. 登录[华为云DLI控制台](https://console.huaweicloud.com/dli/#/main/dashboard)。 2. 左侧导航栏选择"资源管理 \> 弹性资源池"，进入弹性资源池管理页面。 3. 单击右上角"购买弹性资源池"，关键参数如下说明，**其他参数项如表中未说明，默认即可** 。 表3弹性资源池 | 参数项 | 参数值 | |:---|:---| | 计费模式 | 按需计费 | | 区域 | 华北-北京四 | | 名称 | dli_dws | | 规格 | 基础版 | | 网段 | 172.16.0.0/18，需选择与Kafka和DWS不在同一个网段。例如，如果Kafka和DWS在192.168.x.x网段，则DLI选择172.16.x.x。 | 4. 单击"立即购买"，单击"提交"。 等待资源池创建成功，继续执行下一步。 5. 在弹性资源池页面，单击创建好的资源池所在行右侧的"添加队列"，填写如下参数，其他参数项如表中未说明，默认即可。 表4添加队列 | 参数项 | 参数值 | |:---|:---| | 名称 | dli_dws | | 类型 | 通用队列 | 6. 单击"下一步"，单击"确定"。队列创建成功。 #### 步骤四：创建Kafka和DWS的增强型跨源连接 1. 更新DLI的委托权限。 1. 回到DLI管理控制台，左侧选择"全局配置 \> 服务授权"。 2. 勾选"DLI UserInfo Agency Access"、"DLI Datasource Connections Agency Access"、"DLI Notification Agency Access"。 3. 单击"更新委托权限"。单击"确定"。 图3更新DLI委托权限 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002330155020.png "点击放大") 2. 放通Kafka的安全组，允许DLI队列所在的网段可以访问Kafka。 1. 回到Kafka控制台，单击Kafka实例名称进入基本信息。查看"连接信息"的"内网连接地址"，并记录下此地址，以备后续步骤使用。 图4kafka内网连接地址 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002182693701.png "点击放大") 2. 单击网络的安全组名称。 图5kafka安全组 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002182852897.png "点击放大") 3. 选择"入方向规则 \> 添加规则"，如下图，添加DLI队列的网段地址，本实践为172.16.0.0/18，实际请与[步骤三：创建DLI弹性资源池及队列]的时候填入的网段保持一致。 图6kafka安全组添加规则 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001588460008.png "点击放大") 4. 单击"确定"。 3. 回到DLI管理控制台，单击左侧的"跨源管理"，选择"增强型跨源"，单击"创建"。 4. 填写如下参数，其他参数项如表中未说明，默认即可。 表5DLI到Kafka的连接 | 参数项 | 参数值 | |:---|:---| | 连接名称 | dli_kafka | | 弹性资源池 | 选择上面创建的DLI队列名称dli_dws。 | | 虚拟私有云 | 选择Kafka所在的虚拟私有云。 | | 子网 | 选择Kafka所在的子网。 | | 其他参数 | 保持默认。 | 图7创建连接 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001588619968.png "点击放大") 5. 单击"确定"。等待Kafka连接创建成功。 ![](https://support.huaweicloud.com/bestpractice-dws/public_sys-resources/note_3.0-zh-cn.png) 如果增强型跨源连接创建时，没有选择弹性资源池，也可以在跨源连接创建成功后，手动绑定弹性资源池： 1. 选择跨源连接所在行右侧的"更多 \> 绑定弹性资源池"。 2. 选择相应的弹性资源池，单击"确定"。 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002182702529.png "点击放大") 6. 左侧导航栏选择"资源管理 \> 队列管理"，选择dli_dws所在行操作列的"更多 \> 测试地址连通性"。 7. 在地址栏中，输入[2.a]获取的Kafka实例的内网IP和端口（Kafka的地址有三个，输入一个即可）。 图8测试kafka连通性 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001638459789.png "点击放大") 8. 单击"测试"，验证DLI连通Kafka成功。 9. 如果连通性测试失败，请参见以下步骤操作。 确保Kafka的安全组已放开DLI弹性资源池所在网段，确保已更新委托。 1. 在DLI的"跨源管理"页面，在刚创建的跨源连接名称右侧的"操作"列，选择"更多 \> 解绑弹性资源池"。 2. 将原有的弹性资源池去勾选，单击"确定"。 3. 重新在右侧的"操作"列，选择"更多 \> 绑定弹性资源池"。 4. 将上面创建的弹性资源池dli_dws勾选上，单击"确定"。 5. 等待约2分钟，再重新进行连接测试即可。 6. 如果仍连接失败，请参见[创建跨源成功但测试网络连通性失败怎么办](https://support.huaweicloud.com/dli_faq/dli_03_0179.html)逐一排查。 10. 进入到DWS管理控制台，左侧导航栏单击"集群 \> 集群列表"，单击集群名称进入DWS集群详情。 11. 如下图，记录下DWS集群的内网域名、端口和弹性负载均衡地址，以备后面步骤需要。 图9内网域名和ELB地址 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001595206680.png "点击放大") 12. 单击安全组名称。 图10DWS安全组 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001638255829.png "点击放大") 13. 选择"入方向规则 \> 添加规则"，如下图，添加DLI队列的网段地址，本实践为172.16.0.0/18，实际请与[步骤三：创建DLI弹性资源池及队列]的时候填入的网段保持一致。 图11DWS安全组添加规则 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001588140832.png "点击放大") 14. 单击"确定"。 15. 再切换到DLI控制台，左侧选择"资源管理 \> 队列管理"，选择dli_dws所在行操作列的"更多 \> 测试地址连通性"。 16. 在地址栏中，输入[11]获取的DWS集群的弹性负载均衡IP和端口。 图12测试DWS连通 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001638380441.png "点击放大") 17. 单击"测试"，验证DLI连通DWS成功。 #### 步骤五：准备DWS对接Flink工具dws-connector-flink dws-connector-flink是一款基于DWS JDBC接口实现对接Flink的一个工具。在配置DLI作业阶段，将该工具及依赖放入Flink类加载目录，提升Flink作业入库DWS的能力。 1. 浏览器访问。 2. 在软件列表中选择Flink相应版本号1.15，本实践选择**DWS Connector Flink SQL 1 15** 。 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002330051610.png "点击放大") 3. 选择最新分支，实际请以官网发布的新分支为准。 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002330211450.png "点击放大") 4. 单击jar图标下载即可。 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002364049849.png "点击放大") 5. 创建OBS桶，本实践桶名设置为obs-flink-dws，并将此文件**dws-connector-flink-sql-1.15-2.12_2.0.0.r4.jar**上传到OBS桶下，注意桶也保持与DLI在一个区域下，本实践为"华北-北京四"。 #### 步骤六：创建并编辑DLI Flink作业 1. 创建OBS委托策略。 1. 鼠标悬浮在控制台右上角的账户名称上，单击"统一身份认证"。 2. 左侧选择"委托"，单击右上角"创建委托"。 - 委托名称：dli_ac_obs。 - 委托类型：云服务。 - 云服务：数据湖探索 DLI。 - 持续时间：永久。 图13创建OBS委托策略 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002183094949.png "点击放大") 3. 单击"完成"，单击"立即授权"。 4. 进入授权页面，单击"新建策略"。 5. 配置策略信息。输入策略名称，本例：dli_ac_obs，选择"JSON视图"。 6. 在策略内容中粘贴自定义策略。"OBS桶名"替换为实际的桶名。 ``` { "Version": "1.1", "Statement": [ { "Effect": "Allow", "Action": [ "obs:object:GetObject", "obs:object:DeleteObjectVersion", "obs:bucket:GetBucketLocation", "obs:bucket:GetLifecycleConfiguration", "obs:object:AbortMultipartUpload", "obs:object:DeleteObject", "obs:bucket:GetBucketLogging", "obs:bucket:HeadBucket", "obs:object:PutObject", "obs:object:GetObjectVersionAcl", "obs:bucket:GetBucketAcl", "obs:bucket:GetBucketVersioning", "obs:bucket:GetBucketStoragePolicy", "obs:bucket:ListBucketMultipartUploads", "obs:object:ListMultipartUploadParts", "obs:bucket:ListBucketVersions", "obs:bucket:ListBucket", "obs:object:GetObjectVersion", "obs:object:GetObjectAcl", "obs:bucket:GetBucketPolicy", "obs:bucket:GetBucketStorage" ], "Resource": [ "OBS:*:*:object:*", "OBS:*:*:bucket:OBS桶名" ] }, { "Effect": "Allow", "Action": [ "obs:bucket:ListAllMyBuckets" ] } ] } ``` 7. 新建策略完成后，单击"下一步"，返回委托授权页面。 8. 选择上面创建的自定义策略。 9. 单击"下一步"，选择委托的授权范围，选择"所有资源"。 10. 单击"确定"，完成授权。 授权后需等待15-30分钟才可生效。 2. 回到DLI管理控制台，左侧选择"作业管理 \> Flink作业"，单击右上角"创建作业"。 3. 类型选择"Flink OpenSource SQL"，名称填写kafka-dws。 图14创建作业 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001638260509.png "点击放大") 4. 单击"确定"。系统自动进入到作业的编辑页面。 5. 在页面右侧填写如下参数，其他参数项如表中未说明，默认即可。 表6flink作业参数 | 参数项 | 参数值 | |:---|:---| | 所属队列 | dli_dws | | Flink版本 | 1.15 | | UDF Jar | 选择[步骤五：准备DWS对接Flink工具dws-connector-flink]的OBS桶中的jar文件。 | | 委托 | 选择[1]创建的委托。 | | OBS桶 | 选择[步骤五：准备DWS对接Flink工具dws-connector-flink]的桶。 | | 开启Checkpoint | 勾选 | | 其他参数 | 保持默认 | 6. 将以下符合Flink要求的SQL代码复制到左侧的SQL代码窗。 其中"Kafka实例内网IP地址和端口"参见[2.a]获取，"DWS内网域名"由[11]获取。 以下是Flink SQL作业代码中常见的参数解释： - **connector** ：用于指定表的数据源的连接器类型，Kafka数据源指定为kafka，DWS可指定为dws，更多取值可参见[Connector列表](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_15027.html)。 - **write.mode**：入库方式，支持auto，copy_merge，copy_upsert，upsert，update，copy_update，copy_auto等。 - **autoFlushBatchSize** ：指定在自动刷新前要缓冲的记录数。当缓冲的记录数达到这个值时，Flink会将数据批量写入目标系统，例如5000行后再写入**Sink**目标端。在Flink SQL中，Sink（中文常译为"接收器"或"输出端"）是指数据流处理管道中的最终输出目的地，负责将处理后的数据写入外部存储系统或发送给下游应用。 - **autoFlushMaxInterval**：指定自动刷新的最大时间间隔。即使缓冲的记录数未达到autoFlushBatchSize，超过这个时间间隔后也会触发刷新。 - **key-by-before-sink：**指定是否在数据写入Sink目标端前按主键分组。这可以确保相同主键的记录被连续写入，对于某些需要主键有序的系统（如某些数据库）很有用。该参数旨在解决多并发写入的场景下且write.mode为upsert时，如果多个子任务中写入Sink的一批数据具有不止一条相同的主键，并且主键相同的这些数据先后顺序不一致，就会导致两个子任务在向DWS根据主键获取行锁时发生互锁的问题。 ``` CREATE TABLE user_kafka ( id string, name string, age int ) WITH ( 'connector' = 'kafka', 'topic' = 'topic-demo', 'properties.bootstrap.servers' = 'Kafka实例内网IP地址和端口', 'properties.group.id' = 'kafka01', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ); CREATE TABLE user_dws ( id string, name string, age int, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'dws', 'url' = 'jdbc:postgresql://DWS内网域名:8000/gaussdb', 'tableName' = 'public.user_dws', 'username' = 'dbadmin', 'password' = '数据库用户dbadmin密码', 'writeMode' = 'auto', 'autoFlushBatchSize'='50000', 'autoFlushMaxInterval'='5s', 'key-by-before-sink'='true' ); INSERT INTO user_dws select * from user_kafka; --将处理结果写入Sink ``` 7. 单击"格式化"，再单击"保存"。 ![](https://support.huaweicloud.com/bestpractice-dws/public_sys-resources/notice_3.0-zh-cn.png) 请务必先单击"格式化"将SQL代码进行格式化处理，否则可能会因为代码复制和粘贴操作过程中引入新的空字符，而导致作业执行失败。 图15作业的SQL语句 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002147852000.png "点击放大") 8. 回到DLI控制台首页，左侧选择"作业管理 \> Flink作业"。 9. 单击作业名称kafka-dws右侧的"启动"，单击"立即启动"。 等待约1分钟，再刷新页面，状态在"运行中"表示作业成功运行。 图16作业运行状态 ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000002364273285.png "点击放大") #### 步骤七：通过Kafka客户端生产和修改消息 1. 参见ECS文档创建一台ECS，具体创建步骤此处不再赘述。创建时，确保ECS的区域、虚拟私有云保持与Kafka一致。 2. 安装JDK。 1. 登录ECS，进入到/usr/local，下载JDK包。 ``` cd /usr/local wget https://download.oracle.com/java/17/latest/jdk-17_linux-x64_bin.tar.gz ``` 2. 解压下载好的JDK包。 ``` tar -zxvf jdk-17_linux-x64_bin.tar.gz ``` 3. 执行以下命令进入/etc/profile文件。 ``` vim /etc/profile ``` 4. 按i进入编辑模式，将以下内容增加到/etc/profile文件的末尾。 ``` export JAVA_HOME=/usr/local/jdk-17.0.7 #jdk安装目录 export JRE_HOME=${JAVA_HOME}/jre export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib:${JAVA_HOME}/test:${JAVA_HOME}/lib/gsjdbc4.jar:${JAVA_HOME}/lib/dt.jar:${JAVA_HOME}/lib/tools.jar:$CLASSPATH export JAVA_PATH=${JAVA_HOME}/bin:${JRE_HOME}/bin export PATH=$PATH:${JAVA_PATH} ``` ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001588460012.png "点击放大") 5. 按ESC，输入:wq!按回车，保存退出。 6. 执行命令，使环境变量生效。 ``` source /etc/profile ``` 7. 执行以下命令，提示如下信息表示jdk安装成功。 ``` java -version ``` ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001638458765.png "点击放大") 3. 安装Kafka客户端。 1. 进入/opt目录，执行以下命令获取Kafka客户端软件包。 ``` cd /opt wget https://archive.apache.org/dist/kafka/2.7.2/kafka_2.12-2.7.2.tgz ``` 2. 解压下载好的软件包。 ``` tar -zxf kafka_2.12-2.7.2.tgz ``` 3. 进入到Kafka客户端目录。 ``` cd /opt/kafka_2.12-2.7.2/bin ``` 4. 执行以下命令连接Kafka。其中 {连接地址}为Kafka的内网连接地址，参见[2.a]获取，topic为[5]创建的Kafka的topic名称。 ``` ./kafka-console-producer.sh --broker-list {连接地址} --topic {Topic名称} ``` 本实践示例如下： ./kafka-console-producer.sh --broker-list 192.168.0.136:9092,192.168.0.214:9092,192.168.0.217:9092 --topic topic-demo ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001638379621.png "点击放大") 如上图出现\>符号，无其他报错，表示连接成功。 5. 在已连接kafka的客户端窗口下，根据[场景描述]规划的数据，复制以下内容（注意一次复制一行），按回车发送，进行生产消息。 ``` {"id":"1","name":"lily","age":"16"} {"id":"2","name":"lucy","age":"17"} {"id":"3","name":"lilei","age":"15"} ``` ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001591678852.png "点击放大") 6. 回到DWS控制台，左侧选择"集群 \> 集群列表"，单击DWS集群右侧"登录"，进入SQL页面。 7. 执行以下SQL语句，确认数据实时入库成功。 ``` SELECT * FROM user_dws ORDER BY id; ``` ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001588300556.png "点击放大") 8. 继续回到ECS中连接Kafka的客户端窗口，复制以下内容（注意一次复制一行），按回车发送，进行生产消息。 ``` {"id":"2","name":"jim","age":"17"} {"id":"3","name":"tom","age":"15"} ``` 9. 回到DWS已打开的SQL窗口，执行以下SQL语句，发现id为2和3的姓名已修改为jim和tom。 符合场景描述预期，本实践结束。 ``` SELECT * FROM user_dws ORDER BY id; ``` ![](https://support.huaweicloud.com/bestpractice-dws/figure/zh-cn_image_0000001590314144.png "点击放大")