DWS版本：8.1.1MRS：3.0.2Flink：1.12.0Flink读取kafka数据，sink到dws里面。程序正常运行12小时左右就报错。但是超时的参数都配置的没问题。statement_timeout=0;session_timeout=0;这俩参数都没问题。但是不知道为什么就会中断了？

认证文件都是按照步骤获取的。这个错误，能定位是哪个认证步骤出问题了吗？

【摘要】 MRS支持在大数据存储容量大、计算资源需要弹性扩展的场景下，用户将数据存储在OBS服务中，使用MRS集群仅做数据计算处理的存算分离模式。 本文将向您介绍如何在MRS集群中运行Flink作业来处理OBS中存储的数据。本文分享自华为云社区《【云小课】EI第47课 MRS离线数据分析-通过Flink作业处理OBS数据》，作者：Hello EI 。MRS支持在大数据存储容量大、计算资源需要弹性扩展的场景下，用户将数据存储在OBS服务中，使用MRS集群仅做数据计算处理的存算分离模式。Flink是一个批处理和流处理结合的统一计算框架，其核心是一个提供了数据分发以及并行化计算的流数据处理引擎。它的最大亮点是流处理，是业界最顶级的开源流处理引擎。本文将向您介绍如何在MRS集群中运行Flink作业来处理OBS中存储的数据。Flink最适合的应用场景是低时延的数据处理（Data Processing）场景：高并发pipeline处理数据，时延毫秒级，且兼具可靠性。在本示例中，我们使用MRS集群内置的Flink WordCount作业程序，来分析OBS文件系统中保存的源数据，以统计源数据中的单词出现次数。当然您也可以获取MRS服务样例代码工程，参考Flink开发指南开发其他Flink流作业程序。本案例基本操作流程如下所示：创建MRS集群创建并购买一个包含有Flink组件的MRS集群，详情请参见购买自定义集群。本文以购买MRS 3.1.0版本的集群为例，集群未开启Kerberos认证。在本示例中，由于我们要分析处理OBS文件系统中的数据，因此在集群的高级配置参数中要为MRS集群绑定IAM权限委托，使得集群内组件能够对接OBS并具有对应文件系统目录的操作权限。您可以直接选择系统默认的“MRS_ECS_DEFAULT_AGENCY”，也可以自行创建其他具有OBS文件系统操作权限的自定义委托。集群购买成功后，在MRS集群的任一节点内，使用omm用户安装集群客户端，具体操作可参考安装并使用集群客户端。例如客户端安装目录为“/opt/client”。准备测试数据在创建Flink作业进行数据分析前，我们需要在提前准备待分析的测试数据，并将该数据上传至OBS文件系统中。本地创建一个“mrs_flink_test.txt”文件，例如文件内容如下：This is a test demo for MRS Flink. Flink is a unified computing framework that supports both batch processing and stream processing. It provides a stream data processing engine that supports data distribution and parallel computing.在云服务列表中选择“存储 > 对象存储服务”，登录OBS管理控制台。单击“并行文件系统”，创建一个并行文件系统，并上传测试数据文件。例如创建的文件系统名称为“mrs-demo-data”，单击系统名称，在“文件”页面中，新建一个文件夹“flink”，上传测试数据至该目录中。则本示例的测试数据完整路径为“obs://mrs-demo-data/flink/mrs_flink_test.txt”。上传数据分析应用程序。使用管理台界面直接提交作业时，将已开发好的Flink应用程序jar文件也可以上传至OBS文件系统中，或者MRS集群内的HDFS文件系统中。本示例中我们使用MRS集群内置的Flink WordCount样例程序，可从MRS集群的客户端安装目录中获取，即“/opt/client/Flink/flink/examples/batch/WordCount.jar”。将“WordCount.jar”上传至“mrs-demo-data/program”目录下。创建并运行Flink作业方式1：在控制台界面在线提交作业。登录MRS管理控制台，单击MRS集群名称，进入集群详情页面。在集群详情页的“概览”页签，单击“IAM用户同步”右侧的“单击同步”进行IAM用户同步。单击“作业管理”，进入“作业管理”页签。单击“添加”，添加一个Flink作业。作业类型：Flink作业名称：自定义，例如flink_obs_test。执行程序路径：本示例使用Flink客户端的WordCount程序为例。运行程序参数：使用默认值。执行程序参数：设置应用程序的输入参数，“input”为待分析的测试数据，“output”为结果输出文件。例如本示例中，我们设置为“--input obs://mrs-demo-data/flink/mrs_flink_test.txt --output obs://mrs-demo-data/flink/output”。服务配置参数：使用默认值即可，如需手动配置作业相关参数，可参考运行Flink作业。确认作业配置信息后，单击“确定”，完成作业的新增，并等待运行完成。方式2：通过集群客户端提交作业。使用root用户登录集群客户端节点，进入客户端安装目录。su - omm cd /opt/client source bigdata_env执行以下命令验证集群是否可以访问OBS。hdfs dfs -ls obs://mrs-demo-data/flink提交Flink作业，指定源文件数据进行消费。flink run -m yarn-cluster /opt/client/Flink/flink/examples/batch/WordCount.jar --input obs://mrs-demo-data/flink/mrs_flink_test.txt --output obs://mrs-demo/data/flink/output2执行后结果类似如下：... Cluster started: Yarn cluster with application id application_1654672374562_0011 Job has been submitted with JobID a89b561de5d0298cb2ba01fbc30338bc Program execution finished Job with JobID a89b561de5d0298cb2ba01fbc30338bc has finished. Job Runtime: 1200 ms查看作业执行结果作业提交成功后，登录MRS集群的FusionInsight Manager界面，选择“集群 > 服务 > Yarn”。单击“ResourceManager WebUI”后的链接进入Yarn Web UI界面，在Applications页面查看当前Yarn作业的详细运行情况及运行日志。等待作业运行完成后，在OBS文件系统中指定的结果输出文件中可查看数据分析输出的结果。下载“output”文件到本地并打开，可查看输出的分析结果。a 3 and 2 batch 1 both 1 computing 2 data 2 demo 1 distribution 1 engine 1 flink 2 for 1 framework 1 is 2 it 1 mrs 1 parallel 1 processing 3 provides 1 stream 2 supports 2 test 1 that 2 this 1 unified 1使用集群客户端命令行提交作业时，若不指定输出目录，在作业运行界面也可直接查看数据分析结果。Job with JobID xxx has finished. Job Runtime: xxx ms Accumulator Results: - e6209f96ffa423974f8c7043821814e9 (java.util.ArrayList) [31 elements] (a,3) (and,2) (batch,1) (both,1) (computing,2) (data,2) (demo,1) (distribution,1) (engine,1) (flink,2) (for,1) (framework,1) (is,2) (it,1) (mrs,1) (parallel,1) (processing,3) (provides,1) (stream,2) (supports,2) (test,1) (that,2) (this,1) (unified,1)

按照用户文档提交flink的样例程序jar包。窗口日志显示：【Login successful for user flink_dev using keytab file user.keytab】说明已经登录认证成功。但是后面报错信息，依次是：【org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute job 'Flink Streaming Job'.】【Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'Flink Streaming Job'.】【Caused by: org.apache.flink.runtime.client.JobSubmissionException: Failed to submit JobGraph.】【Caused by: org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException: Unhandled error in ZooKeeperLeaderRetrievalDriver:Background exception was not retry-able or retry gave up】【Caused by: org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /flink_base/flink/default】尤其是最后一个报错：【Caused by: org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /flink_base/flink/default】是不是zookeeper连接不上？和zookeeper的认证有问题了？

①这个flink-conf.yaml在客户端配置后，是不是自己开发的jar包里面是不能有这个配置文件的？不然会影响客户端的这个文件？②这里的 security.kerberos.login.contexts:Client，里面的Client是固定参数吗？不用动这个值是吧？③security.kerberos.login.contexts：的值 Client,KafkaClien是不是固定值？不用动这个值是吧？

flink提交样例代码的FlinkKafkaJavaExample，运行ReaderFromKafka。采用的SASL_PLAINTEXT模式的kafka。提交任务到集群，报错------------------------------------------------------------ The program finished with the following exception:org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute job 'Flink Streaming Job'.    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:330)    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:198)    at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:767)    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:250)    at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:995)    at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1071)    at java.security.AccessController.doPrivileged(Native Method)    at javax.security.auth.Subject.doAs(Subject.java:422)    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1761)    at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1071)Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'Flink Streaming Job'.    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1951)    at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:128)    at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76)    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1822)    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1804)    at com.huawei.bigdata.flink.examples.ReadFromKafka.main(ReadFromKafka.java:71)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:498)    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:316)    ... 11 moreCaused by: org.apache.flink.runtime.client.JobSubmissionException: Failed to submit JobGraph.    at org.apache.flink.client.program.rest.RestClusterClient.lambda$submitJob$7(RestClusterClient.java:366)    at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:884)    at java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:866)    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)    at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1609)    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)    at java.lang.Thread.run(Thread.java:748)Caused by: org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException: Unhandled error in ZooKeeperLeaderRetrievalDriver:Background exception was not retry-able or retry gave up    at org.apache.flink.runtime.leaderretrieval.ZooKeeperLeaderRetrievalDriver.unhandledError(ZooKeeperLeaderRetrievalDriver.java:178)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl$6.apply(CuratorFrameworkImpl.java:713)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl$6.apply(CuratorFrameworkImpl.java:709)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.listen.ListenerContainer$1.run(ListenerContainer.java:100)    at org.apache.flink.shaded.curator4.org.apache.curator.shaded.com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.listen.ListenerContainer.forEach(ListenerContainer.java:92)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl.logError(CuratorFrameworkImpl.java:708)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl.handleBackgroundOperationException(CuratorFrameworkImpl.java:924)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl.performBackgroundOperation(CuratorFrameworkImpl.java:1001)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl.backgroundOperationsLoop(CuratorFrameworkImpl.java:943)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl.access$300(CuratorFrameworkImpl.java:66)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl$4.call(CuratorFrameworkImpl.java:346)    at java.util.concurrent.FutureTask.run(FutureTask.java:266)    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)    ... 3 moreCaused by: org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /flink_base/flink/default    at org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException.create(KeeperException.java:125)    at org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException.create(KeeperException.java:54)    at org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:1573)    at org.apache.flink.shaded.curator4.org.apache.curator.utils.ZKPaths.mkdirs(ZKPaths.java:308)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CreateBuilderImpl$9.performBackgroundOperation(CreateBuilderImpl.java:801)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.OperationAndData.callPerformBackgroundOperation(OperationAndData.java:84)    at org.apache.flink.shaded.curator4.org.apache.curator.framework.imps.CuratorFrameworkImpl.performBackgroundOperation(CuratorFrameworkImpl.java:965)

本地idea如何构建本机的开发环境，flink读取sasl_plaintext的kafka？然后本机idea启动，直接消费kafka数据？在flink提交任务前加了if (LoginUtil.isSecurityModel()) { try { LOG.info("Securitymode start."); //!!注意，安全认证时，需要用户手动修改为自己申请的机机账号 LoginUtil.securityPrepare(USER_PRINCIPAL, USER_KEYTAB_FILE); } catch (IOException e) { LOG.error("Security prepare failure."); LOG.error("The IOException occured : {}.", e); return; } LOG.info("Security prepare success.");}这部分认证，但是这个好像并没有提交到本地的flinkclient中。

场景描述CDC是变更数据捕获（Change Data Capture）技术的缩写，它可以将源数据库的增量变动记录，同步到一个或多个数据目的中。CDC在数据同步过程中，还可以对数据进行一定的处理，例如分组（GROUP BY）、多表的关联（JOIN）等。本示例通过创建Postgres CDC源表来监控Postgres的数据变化，并将变化的数据信息插入到DWS数据库中。前提条件已创建RDS Postgres实例，具体步骤可参考：RDS PostgreSQL快速入门。本示例创建的RDS Postgres数据库版本选择为：11。说明：创建的RDS Postgres数据库版本不能低于11。已创建DWS实例，具体创建DWS集群的操作可以参考创建DWS集群。本示例创建的DWS集群版本为：8.1.1.205。整体作业开发流程整体作业开发流程参考图1。图1 作业开发流程步骤1：创建队列：创建DLI作业运行的队列。步骤2：创建RDS Postgres数据库：创建RDS Postgres的数据库和表。步骤3：创建DWS数据库和表：创建用于接收数据的DWS数据库和表。步骤4：创建增强型跨源连接：DLI上创建连接RDS和DWS的跨源连接，打通网络。步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。步骤6：发送数据和查询结果：RDS Postgres的表上插入数据，在DWS上查看运行结果。步骤1：创建队列登录DLI管理控制台，在左侧导航栏单击“资源管理 > 队列管理”，可进入队列管理页面。在队列管理界面，单击界面右上角的“购买队列”。在“购买队列”界面，填写具体的队列配置参数，具体参数填写参考如下。计费模式：选择“包年/包月”或“按需计费”。本示例选择“按需计费”。区域和项目：保持默认值即可。名称：填写具体的队列名称。说明：新建的队列名称，名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。长度限制：1~128个字符。队列名称不区分大小写，系统会自动转换为小写。类型：队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”。AZ策略、CPU架构、规格：保持默认即可。企业项目：当前选择为“default”。高级选项：选择“自定义”。网段：配置队列网段。例如，当前配置为10.0.0.0/16。注意：队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合，否则后续创建跨源连接会失败。其他参数根据需要选择和配置。图2 创建队列参数配置完成后，单击“立即购买”，确认配置信息无误后，单击“提交”完成队列创建。步骤2：创建RDS Postgres数据库登录RDS管理控制台，在“实例管理”界面，选择已创建的RDS Postgres实例，选择操作列的“更多 > 登录”，进入数据管理服务实例登录界面。输入实例登录的用户名和密码。单击“登录”，即可进入RDS Postgres数据库并进行管理。在数据库实例界面，单击“新建数据库”，数据库名定义为：testrdsdb，字符集保持默认即可。在testrdsdb数据库下，单击“新建Schema”，Schema名称输入为：test。在test的Schema所在行，单击“操作”列的“打开Schema”。单击“SQL查询”，输入以下创建表语句，创建RDS Postgres表。create table test.cdc_order( order_id VARCHAR, order_channel VARCHAR, order_time VARCHAR, pay_amount FLOAT8, real_pay FLOAT8, pay_time VARCHAR, user_id VARCHAR, user_name VARCHAR, area_id VARCHAR, primary key(order_id));在Postgre中执行下列SQL语句。ALTER TABLE test.cdc_order REPLICA IDENTITY FULL;步骤3：创建DWS数据库和表参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。执行以下命令连接DWS集群的默认数据库“gaussdb”：gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -rgaussdb：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 dws_order; CREATE TABLE dws_order ( order_id VARCHAR, order_channel VARCHAR, order_time VARCHAR, pay_amount FLOAT8, real_pay FLOAT8, pay_time VARCHAR, user_id VARCHAR, user_name VARCHAR, area_id VARCHAR );步骤4：创建增强型跨源连接创建DLI连接RDS的增强型跨源连接在RDS管理控制台，选择“实例管理”，单击对应的RDS实例名称，进入到RDS的基本信息页面。在“基本信息”的“连接信息”中获取该实例的“内网地址”、“数据库端口”、“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_rds。弹性资源池：选择步骤1：创建队列中已经创建的队列。虚拟私有云：选择RDS的虚拟私有云。子网：选择RDS的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的RDS连接信息，地址栏输入“RDS内网地址:RDS数据库端口”，单击“测试”测试DLI到RDS网络是否可达。创建DLI连接DWS的增强型跨源连接在DWS管理控制台，选择“集群管理”，单击已创建的DWS集群名称，进入到DWS的基本信息页面。在“基本信息”的“数据库属性”中获取该实例的“内网IP”、“端口”，“基本信息”页面的“网络”中获取“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。说明：本示例默认RDS和DWS实例分别在两个VPC和子网下，所以要分别创建增强型跨源连接打通网络。如果RDS和DWS实例属于同一VPC和子网下，则创建增强型跨源一次即可，4和5不需要再执行。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_dws。弹性资源池：选择步骤1：创建队列中已经创建的队列。虚拟私有云：选择DWS的虚拟私有云。子网：选择DWS的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的DWS连接信息，地址栏输入“DWS内网IP:DWS端口”，单击“测试”测试DLI到DWS网络是否可达。步骤5：运行作业在DLI管理控制台，单击“作业管理 > Flink作业”，在Flink作业管理界面，单击“创建作业”。在创建队列界面，类型选择“Flink OpenSource SQL”，名称填写为：FlinkCDCPostgreDWS。单击“确定”，跳转到Flink作业编辑界面。在Flink OpenSource SQL作业编辑界面，配置如下参数。所属队列：选择步骤1：创建队列中创建的队列。Flink版本：选择1.12。保存作业日志：勾选。OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。开启Checkpoint：勾选。Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。SQL中加粗的参数需要根据实际情况修改。说明：本示例使用的Flink版本为1.12，故Flink OpenSource SQL语法也是1.12。本示例数据源是Kafka，写入结果数据到Elasticsearch，故请参考Flink OpenSource SQL 1.12创建Postgres CDC源表和Flink OpenSource SQL 1.12创建DWS结果表。create table PostgreCdcSource( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id STRING, primary key (order_id) not enforced ) with ( 'connector' = 'postgres-cdc', 'hostname' = '192.168.15.153',--IP替换为RDS Postgres的实例IP 'port' = '5432',--端口替换为RDS Postgres的实例端口 'username' = 'xxxxx',--RDS Postgres实例的数据库用户名 'password' = 'xxxxx',-RDS Postgres实例的数据库用户密码 'database-name' = 'testrdsdb',--RDS Postgres实例的数据库名 'schema-name' = 'test',--RDS Postgres数据库下的schema 'table-name' = 'cdc_order'--RDS Postgres数据库下的表名 ); create table dwsSink( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id STRING, primary key(order_id) not enforced ) with ( 'connector' = 'gaussdb', 'driver' = 'com.huawei.gauss200.jdbc.Driver', 'url' = 'jdbc:gaussdb://192.168.168.16:8000/testdwsdb', ---192.168.168.16:8000替换为DWS的内网IP和端口，testdwsdb为创建的DWS数据库名 'table-name' = 'test\".\"dws_order', ---test为创建的DWS表的schema，dws_order为对应的DWS表名 'username' = 'xxxxx',--替换为DWS实例的用户名 'password' = 'xxxxx',--替换为DWS实例的用户密码 'write.mode' = 'insert' ); insert into dwsSink select * from PostgreCdcSource where pay_amount > 100;单击“语义校验”确保SQL语义校验成功。单击“保存”，保存作业。单击“启动”，启动作业，确认作业参数信息，单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。步骤6：发送数据和查询结果登录RDS管理控制台，在“实例管理”界面，选择已创建的RDS Postgres实例，选择操作列的“更多 > 登录”，进入数据管理服务实例登录界面。输入实例登录的用户名和密码。单击“登录”，即可进入RDS Postgres数据库并进行管理。在已创建的数据库的操作列，单击“SQL查询”，输入以下创建表语句，插入测试数据。insert into test.cdc_order values ('202103241000000001','webShop','2021-03-24 10:00:00','50.00','100.00','2021-03-24 10:02:03','0001','Alice','330106'), ('202103251606060001','appShop','2021-03-24 12:06:06','200.00','180.00','2021-03-24 16:10:06','0002','Jason','330106'), ('202103261000000001','webShop','2021-03-24 14:03:00','300.00','100.00','2021-03-24 10:02:03','0003','Lily','330106'), ('202103271606060001','appShop','2021-03-24 16:36:06','99.00','150.00','2021-03-24 16:10:06','0001','Henry','330106');参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。执行以下命令连接DWS集群的默认数据库“testdwsdb”：gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r执行以下语句，查询DWS的表数据。select * from test.dws_order;查询结果参考如下：order_channel order_channel order_time pay_amount real_pay pay_time user_id user_name area_id 202103251606060001 appShop 2021-03-24 12:06:06 200.0 180.0 2021-03-24 16:10:06 0002 Jason 330106 202103261000000001 webShop 2021-03-24 14:03:00 300.0 100.0 2021-03-24 10:02:03 0003 Lily 330106

场景描述CDC是变更数据捕获（Change Data Capture）技术的缩写，它可以将源数据库的增量变动记录，同步到一个或多个数据目的中。CDC在数据同步过程中，还可以对数据进行一定的处理，例如分组（GROUP BY）、多表的关联（JOIN）等。本示例通过创建MySQL CDC源表来监控MySQL的数据变化，并将变化的数据信息插入到DWS数据库中。前提条件已创建RDS MySQL实例，具体步骤可参考：RDS MySQL快速入门。本示例创建的RDS MySQL数据库版本选择为：8.0。已创建DWS实例，具体创建DWS集群的操作可以参考创建DWS集群。本示例创建的DWS集群版本为：8.1.1.205。整体作业开发流程整体作业开发流程参考图1。图1 作业开发流程步骤1：创建队列：创建DLI作业运行的队列。步骤2：创建RDS MySQL数据库和表：创建RDS MySQL的数据库和表。步骤3：创建DWS数据库和表：创建用于接收数据的DWS数据库和表。步骤4：创建增强型跨源连接：DLI上创建连接RDS和DWS的跨源连接，打通网络。步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。步骤6：发送数据和查询结果：RDS MySQL的表上插入数据，在DWS上查看运行结果。步骤1：创建队列登录DLI管理控制台，在左侧导航栏单击“资源管理 > 队列管理”，可进入队列管理页面。在队列管理界面，单击界面右上角的“购买队列”。在“购买队列”界面，填写具体的队列配置参数，具体参数填写参考如下。计费模式：选择“包年/包月”或“按需计费”。本示例选择“按需计费”。区域和项目：保持默认值即可。名称：填写具体的队列名称。说明：新建的队列名称，名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。长度限制：1~128个字符。队列名称不区分大小写，系统会自动转换为小写。类型：队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”。AZ策略、CPU架构、规格：保持默认即可。企业项目：当前选择为“default”。高级选项：选择“自定义”。网段：配置队列网段。例如，当前配置为10.0.0.0/16。注意：队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合，否则后续创建跨源连接会失败。其他参数根据需要选择和配置。图2 创建队列参数配置完成后，单击“立即购买”，确认配置信息无误后，单击“提交”完成队列创建。步骤2：创建RDS MySQL数据库和表登录RDS管理控制台，在“实例管理”界面，选择已创建的RDS MySQL实例，选择操作列的“更多 > 登录”，进入数据管理服务实例登录界面。输入实例登录的用户名和密码。单击“登录”，即可进入RDS MySQL数据库并进行管理。在数据库实例界面，单击“新建数据库”，数据库名定义为：testrdsdb，字符集保持默认即可。在已创建的数据库的操作列，单击“SQL查询”，输入以下创建表语句，创建RDS MySQL表。CREATE TABLE mysqlcdc ( `order_id` VARCHAR(64) NOT NULL, `order_channel` VARCHAR(32) NOT NULL, `order_time` VARCHAR(32), `pay_amount` DOUBLE, `real_pay` DOUBLE, `pay_time` VARCHAR(32), `user_id` VARCHAR(32), `user_name` VARCHAR(32), `area_id` VARCHAR(32) ) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8mb4;步骤3：创建DWS数据库和表参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。执行以下命令连接DWS集群的默认数据库“gaussdb”：gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -rgaussdb：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 );步骤4：创建增强型跨源连接创建DLI连接RDS的增强型跨源连接在RDS管理控制台，选择“实例管理”，单击对应的RDS实例名称，进入到RDS的基本信息页面。在“基本信息”的“连接信息”中获取该实例的“内网地址”、“数据库端口”、“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_rds。弹性资源池：选择步骤1：创建队列中已经创建的队列。虚拟私有云：选择RDS的虚拟私有云。子网：选择RDS的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的RDS连接信息，地址栏输入“RDS内网地址:RDS数据库端口”，单击“测试”测试DLI到RDS网络是否可达。创建DLI连接DWS的增强型跨源连接在DWS管理控制台，选择“集群管理”，单击已创建的DWS集群名称，进入到DWS的基本信息页面。在“基本信息”的“数据库属性”中获取该实例的“内网IP”、“端口”，“基本信息”页面的“网络”中获取“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。说明：本示例默认RDS和DWS实例分别在两个VPC和子网下，所以要分别创建增强型跨源连接打通网络。如果RDS和DWS实例属于同一VPC和子网下，则创建增强型跨源一次即可，4和5不需要再执行。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_dws。弹性资源池：选择步骤1：创建队列中已经创建的队列。虚拟私有云：选择DWS的虚拟私有云。子网：选择DWS的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的DWS连接信息，地址栏输入“DWS内网IP:DWS端口”，单击“测试”测试DLI到DWS网络是否可达。步骤5：运行作业在DLI管理控制台，单击“作业管理 > Flink作业”，在Flink作业管理界面，单击“创建作业”。在创建队列界面，类型选择“Flink OpenSource SQL”，名称填写为：FlinkCDCMySQLDWS。单击“确定”，跳转到Flink作业编辑界面。在Flink OpenSource SQL作业编辑界面，配置如下参数。所属队列：选择步骤1：创建队列中创建的队列。Flink版本：选择1.12。保存作业日志：勾选。OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。开启Checkpoint：勾选。Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。SQL中加粗的参数需要根据实际情况修改。说明：本示例使用的Flink版本为1.12，故Flink OpenSource SQL语法也是1.12。本示例数据源是Kafka，写入结果数据到Elasticsearch，故请参考Flink OpenSource SQL 1.12创建MySQL CDC源表和Flink OpenSource SQL 1.12创建DWS结果表。create table mysqlCdcSource( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id STRING ) with ( 'connector' = 'mysql-cdc', 'hostname' = '192.168.12.148',--IP替换为RDS MySQL的实例IP 'port' = '3306',--端口替换为RDS MySQL的实例端口 'username' = 'xxx',--RDS MySQL实例的数据库用户名 'password' = 'xxx',--RDS MySQL实例的数据库用户密码 'database-name' = 'testrdsdb',--RDS MySQL实例的数据库名 'table-name' = 'mysqlcdc'--RDS MySQL实例的数据库下的表名 ); create table dwsSink( order_channel string, pay_amount double, real_pay double, primary key(order_channel) not enforced ) with ( 'connector' = 'gaussdb', 'driver' = 'com.huawei.gauss200.jdbc.Driver', 'url' = 'jdbc:gaussdb://192.168.168.16:8000/testdwsdb', ---192.168.168.16:8000替换为DWS的内网IP和端口，testdwsdb为创建的DWS数据库名 'table-name' = 'test\".\"dwsresult', ---test为创建的DWS表的schema，dwsresult为对应的DWS表名 'username' = 'xxx',--替换为DWS实例的用户名 'password' = 'xxx',--替换为DWS实例的用户密码 'write.mode' = 'insert' ); insert into dwsSink select order_channel, sum(pay_amount),sum(real_pay) from mysqlCdcSource group by order_channel;单击“语义校验”确保SQL语义校验成功。单击“保存”，保存作业。单击“启动”，启动作业，确认作业参数信息，单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。步骤6：发送数据和查询结果登录RDS管理控制台，在“实例管理”界面，选择已创建的RDS MySQL实例，选择操作列的“更多 > 登录”，进入数据管理服务实例登录界面。输入实例登录的用户名和密码。单击“登录”，即可进入RDS MySQL数据库并进行管理。在已创建的数据库的操作列，单击“SQL查询”，输入以下创建表语句，插入测试数据。insert into mysqlcdc values ('202103241000000001','webShop','2021-03-24 10:00:00','100.00','100.00','2021-03-24 10:02:03','0001','Alice','330106'), ('202103241206060001','appShop','2021-03-24 12:06:06','200.00','180.00','2021-03-24 16:10:06','0002','Jason','330106'), ('202103241403000001','webShop','2021-03-24 14:03:00','300.00','100.00','2021-03-24 10:02:03','0003','Lily','330106'), ('202103241636060001','appShop','2021-03-24 16:36:06','200.00','150.00','2021-03-24 16:10:06','0001','Henry','330106');参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。执行以下命令连接DWS集群的默认数据库“testdwsdb”：gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r执行以下命令，查询DWS的表数据。select * from test.dwsresult;查询结果参考如下：order_channel pay_amount real_pay appShop 400.0 330.0 webShop 400.0 200.0

场景描述本示例场景对汽车驾驶的实时数据信息进行分析，将满足特定条件的数据结果进行汇总。数据的输入源为Kafka，结果输出到DWS中。例如，如下样例输入数据：{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"} {"car_id":"3028", "car_owner":"hanmeimei", "car_age":"6", "average_speed":"92", "total_miles":"17000"} {"car_id":"3029", "car_owner":"zhangsan", "car_age":"10", "average_speed":"81", "total_miles":"230000"}预期输出：average_speed <= 90 和 total_miles <= 200000的车辆，即：{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"}前提条件已创建DMS Kafka实例，具体步骤可参考：DMS Kafka入门指引。注意：创建DMS Kafka实例时，不能开启Kafka SASL_SSL。已创建Elasticsearch类型的CSS集群。具体创建CSS集群的操作可以参考创建CSS集群。本示例创建的CSS集群版本为：7.6.2，集群为非安全集群。整体作业开发流程整体作业开发流程参考图1。图1 作业开发流程步骤1：创建队列：创建DLI作业运行的队列。步骤2：创建Kafka的Topic：创建Kafka生产消费数据的Topic。步骤3：创建Elasticsearch搜索索引：创建Elasticsearch搜索索引用于接收结果数据。步骤4：创建增强型跨源连接：DLI上创建连接Kafka和CSS的跨源连接，打通网络。步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。步骤6：发送数据和查询结果：Kafka上发送流数据，在CSS上查看运行结果。步骤1：创建队列登录DLI管理控制台，在左侧导航栏单击“资源管理 > 队列管理”，可进入队列管理页面。在队列管理界面，单击界面右上角的“购买队列”。在“购买队列”界面，填写具体的队列配置参数，具体参数填写参考如下。计费模式：选择“包年/包月”或“按需计费”。本示例选择“按需计费”。区域和项目：保持默认值即可。名称：填写具体的队列名称。说明：新建的队列名称，名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。长度限制：1~128个字符。队列名称不区分大小写，系统会自动转换为小写。类型：队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”。AZ策略、CPU架构、规格：保持默认即可。企业项目：当前选择为“default”。高级选项：选择“自定义”。网段：配置队列网段。例如，当前配置为10.0.0.0/16。注意：队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合，否则后续创建跨源连接会失败。其他参数根据需要选择和配置。图2 创建队列参数配置完成后，单击“立即购买”，确认配置信息无误后，单击“提交”完成队列创建。步骤2：创建Kafka的Topic在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。单击“Topic管理 > 创建Topic”，创建一个Topic。Topic配置参数如下：Topic名称。本示例输入为：testkafkatopic。分区数：1。副本数：1其他参数保持默认即可。步骤3：创建Elasticsearch搜索索引登录CSS管理控制台，选择“集群管理 > Elasticsearch”。在集群管理界面，在已创建的CSS集群的“操作”列，单击“Kibana”访问集群。在Kibana的左侧导航中选择“Dev Tools”，进入到Console界面。在Console界面，执行如下命令创建索引“shoporders”。PUT /shoporders { "settings": { "number_of_shards": 1 }, "mappings": { "properties": { "order_id": { "type": "text" }, "order_channel": { "type": "text" }, "order_time": { "type": "text" }, "pay_amount": { "type": "double" }, "real_pay": { "type": "double" }, "pay_time": { "type": "text" }, "user_id": { "type": "text" }, "user_name": { "type": "text" }, "area_id": { "type": "text" } } } }步骤4：创建增强型跨源连接创建DLI连接Kafka的增强型跨源连接在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。在“连接信息”中获取该Kafka的“内网连接地址”，在“基本信息”的“网络”中获取获取该实例的“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“网络”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_kafka。弹性资源池：选择步骤1：创建队列中已经创建的队列。虚拟私有云：选择Kafka的虚拟私有云。子网：选择Kafka的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中添加的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据中获取的Kafka连接信息，地址栏输入“Kafka内网地址:Kafka数据库端口”，单击“测试”测试DLI到Kafka网络是否可达。创建DLI连接CSS的增强型跨源连接在CSS管理控制台，选择“集群管理”，单击已创建的CSS集群名称，进入到CSS的基本信息页面。在“基本信息”中获取CSS的“内网访问地址”、“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。说明：本示例默认Kafka和CSS实例分别在两个VPC和子网下，所以要分别创建增强型跨源连接打通网络。如果Kafka和CSS实例属于同一VPC和子网下，则创建增强型跨源一次即可，4和5不需要再执行。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_css。弹性资源池：选择步骤1：创建队列中已经创建的队列。虚拟私有云：选择CSS的虚拟私有云。子网：选择CSS的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中添加的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2获取的CSS连接信息，地址栏输入“CSS内网地址:CSS内网端口”，单击“测试”测试DLI到CSS网络是否可达。步骤5：运行作业在DLI管理控制台，单击“作业管理 > Flink作业”，在Flink作业管理界面，单击“创建作业”。在创建队列界面，类型选择“Flink OpenSource SQL”，名称填写为：FlinkKafkaES。单击“确定”，跳转到Flink作业编辑界面。在Flink OpenSource SQL作业编辑界面，配置如下参数。所属队列：选择步骤1：创建队列中创建的队列。Flink版本：选择1.12。保存作业日志：勾选。OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。开启Checkpoint：勾选。Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。SQL中加粗的参数需要根据实际情况修改。说明：本示例使用的Flink版本为1.12，故Flink OpenSource SQL语法也是1.12。本示例数据源是Kafka，写入结果数据到Elasticsearch，故请参考Flink OpenSource SQL 1.12创建Kafka源表和Flink OpenSource SQL 1.12创建Elasticsearch结果表。CREATE TABLE kafkaSource ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) with ( "connector" = "kafka", "properties.bootstrap.servers" = "10.128.0.120:9092,10.128.0.89:9092,10.128.0.83:9092",--替换为kafka的内网连接地址和端口 "properties.group.id" = "click", "topic" = "testkafkatopic", --创建的Kafka Topic "format" = "json", "scan.startup.mode" = "latest-offset" ); CREATE TABLE elasticsearchSink ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'elasticsearch-7', 'hosts' = '192.168.168.125:9200', --替换为CSS集群的内网地址和端口 'index' = 'shoporders' --创建的Elasticsearch搜索引擎 ); --将Kafka数据写入到Elasticsearch索引中 insert into elasticsearchSink select * from kafkaSource;单击“语义校验”确保SQL语义校验成功。单击“保存”，保存作业。单击“启动”，启动作业，确认作业参数信息，单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。步骤6：发送数据和查询结果使用Kafka客户端向步骤2：创建Kafka的Topic中的Topic发送数据，模拟实时数据流。Kafka生产和发送数据的方法请参考：DMS - 连接实例生产消费信息。发送样例数据如下：{"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0002", "user_name":"Jason", "area_id":"330106"}发送成功后，在CSS集群的Kibana中执行下述语句并查看相应结果：GET shoporders/_search查询结果返回如下：{ "took" : 0, "timed_out" : false, "_shards" : { "total" : 1, "successful" : 1, "skipped" : 0, "failed" : 0 }, "hits" : { "total" : { "value" : 2, "relation" : "eq" }, "max_score" : 1.0, "hits" : [ { "_index" : "shoporders", "_type" : "_doc", "_id" : "6fswzIAByVjqg3_qAyM1", "_score" : 1.0, "_source" : { "order_id" : "202103241000000001", "order_channel" : "webShop", "order_time" : "2021-03-24 10:00:00", "pay_amount" : 100.0, "real_pay" : 100.0, "pay_time" : "2021-03-24 10:02:03", "user_id" : "0001", "user_name" : "Alice", "area_id" : "330106" } }, { "_index" : "shoporders", "_type" : "_doc", "_id" : "6vs1zIAByVjqg3_qyyPp", "_score" : 1.0, "_source" : { "order_id" : "202103241606060001", "order_channel" : "appShop", "order_time" : "2021-03-24 16:06:06", "pay_amount" : 200.0, "real_pay" : 180.0, "pay_time" : "2021-03-24 16:10:06", "user_id" : "0002", "user_name" : "Jason", "area_id" : "330106" } } ] } }

场景描述该场景为对汽车驾驶的实时数据信息进行分析，将满足特定条件的数据结果进行汇总。数据的输入源为Kafka，结果输出到DWS中。例如，如下样例输入数据：{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"} {"car_id":"3028", "car_owner":"hanmeimei", "car_age":"6", "average_speed":"92", "total_miles":"17000"} {"car_id":"3029", "car_owner":"zhangsan", "car_age":"10", "average_speed":"81", "total_miles":"230000"}预期输出：average_speed <= 90 和 total_miles <= 200000的车辆，即：{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"}前提条件已创建DMS Kafka实例，具体步骤可参考：创建Kafka实例。注意：创建DMS Kafka实例时，不能开启Kafka SASL_SSL。已创建DWS实例，具体创建DWS集群的操作可以参考创建DWS集群。本示例创建的DWS集群版本为：8.1.1.205。整体作业开发流程整体作业开发流程参考图1。图1 作业开发流程步骤1：创建队列：创建DLI作业运行的队列。步骤2：创建Kafka的Topic：创建Kafka生产消费数据的Topic。步骤3：创建DWS数据库和表：创建DWS数据库和表信息。步骤4：创建增强型跨源连接：DLI上创建连接Kafka和DWS的跨源连接，打通网络。步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。步骤6：发送数据和查询结果：Kafka上发送流数据，在RDS上查看运行结果。步骤1：创建队列登录DLI管理控制台，在左侧导航栏单击“资源管理 > 队列管理”，可进入队列管理页面。在队列管理界面，单击界面右上角的“购买队列”。在“购买队列”界面，填写具体的队列配置参数，具体参数填写参考如下。计费模式：选择“包年/包月”或“按需计费”。本示例选择“按需计费”。区域和项目：保持默认值即可。名称：填写具体的队列名称。说明：新建的队列名称，名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。长度限制：1~128个字符。队列名称不区分大小写，系统会自动转换为小写。类型：队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”。AZ策略、CPU架构、规格：保持默认即可。企业项目：当前选择为“default”。高级选项：选择“自定义”。网段：配置队列网段。例如，当前配置为10.0.0.0/16。注意：队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合，否则后续创建跨源连接会失败。其他参数根据需要选择和配置。图2 创建队列参数配置完成后，单击“立即购买”，确认配置信息无误后，单击“提交”完成队列创建。步骤2：创建Kafka的Topic在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。单击“Topic管理 > 创建Topic”，创建一个Topic。Topic配置参数如下：Topic名称。本示例输入为：testkafkatopic。分区数：1。副本数：1其他参数保持默认即可。步骤3：创建DWS数据库和表参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。执行以下命令连接DWS集群的默认数据库“gaussdb”：gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -rgaussdb：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 qualified_cars; CREATE TABLE qualified_cars ( car_id VARCHAR, car_owner VARCHAR, car_age INTEGER , average_speed FLOAT8, total_miles FLOAT8 );步骤4：创建增强型跨源连接创建DLI连接Kafka的增强型跨源连接在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。在“连接信息”中获取该Kafka的“内网连接地址”，在“基本信息”的“网络”中获取获取该实例的“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“网络”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_kafka。弹性资源池：选择步骤1：创建队列中已经创建的队列名。虚拟私有云：选择Kafka的虚拟私有云。子网：选择Kafka的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的Kafka连接信息，地址栏输入“Kafka内网地址:Kafka数据库端口”，单击“测试”测试DLI到Kafka网络是否可达。创建DLI连接DWS的增强型跨源连接在DWS管理控制台，选择“集群管理”，单击已创建的DWS集群名称，进入到DWS的基本信息页面。在“基本信息”的“数据库属性”中获取该实例的“内网IP”、“端口”，“基本信息”页面的“网络”中获取“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。说明：本示例默认Kafka和DWS实例分别在两个VPC和子网下，所以要分别创建增强型跨源连接打通网络。如果Kafka和DWS实例属于同一VPC和子网下，则创建增强型跨源一次即可，4和5不需要再执行。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_dws。弹性资源池：选择步骤1：创建队列中已经创建的队列名。虚拟私有云：选择DWS的虚拟私有云。子网：选择DWS的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的DWS连接信息，地址栏输入“DWS内网IP:DWS端口”，单击“测试”测试DLI到DWS网络是否可达。步骤5：运行作业在DLI管理控制台，单击“作业管理 > Flink作业”，在Flink作业管理界面，单击“创建作业”。在创建队列界面，类型选择“Flink OpenSource SQL”，名称填写为：FlinkKafkaDWS。单击“确定”，跳转到Flink作业编辑界面。在Flink OpenSource SQL作业编辑界面，配置如下参数。所属队列：选择步骤1：创建队列中创建的队列。Flink版本：选择1.12。保存作业日志：勾选。OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。开启Checkpoint：勾选。Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。SQL中加粗的参数需要根据实际情况修改。说明：本示例使用的Flink版本为1.12，故Flink OpenSource SQL语法也是1.12。本示例数据源是Kafka，写入结果数据到DWS，故请参考Flink OpenSource SQL 1.12创建Kafka源表和Flink OpenSource SQL 1.12创建DWS结果表（RDS连接）。create table car_infos( car_id STRING, car_owner STRING, car_age INT, average_speed DOUBLE, total_miles DOUBLE ) with ( "connector" = "kafka", "properties.bootstrap.servers" = "10.128.0.120:9092,10.128.0.89:9092,10.128.0.83:9092",--替换为kafka的内网连接地址和端口 "properties.group.id" = "click", "topic" = "testkafkatopic",--创建的Kafka的Topic "format" = "json", "scan.startup.mode" = "latest-offset" ); create table qualified_cars ( car_id STRING, car_owner STRING, car_age INT, average_speed DOUBLE, total_miles DOUBLE ) WITH ( 'connector' = 'gaussdb', 'driver' = 'com.huawei.gauss200.jdbc.Driver', 'url' = 'jdbc:gaussdb://192.168.168.16:8000/testdwsdb', ---192.168.168.16:8000替换为DWS的内网IP和端口，testdwsdb为创建的DWS数据库名 'table-name' = 'test\".\"qualified_cars', ---test为创建的DWS表的schema，qualified_cars为对应的DWS表名 'username' = 'xxxx',--替换为DWS实例的用户名 'password' = 'xxxx',--替换为DWS实例的用户密码 'write.mode' = 'insert' ); /** 将合格车辆信息输出 **/ INSERT INTO qualified_cars SELECT * FROM car_infos where average_speed <= 90 and total_miles <= 200000;单击“语义校验”确保SQL语义校验成功。单击“保存”，保存作业。单击“启动”，启动作业，确认作业参数信息，单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。步骤6：发送数据和查询结果使用Kafka客户端向步骤2：创建Kafka的Topic中的Topic发送数据，模拟实时数据流。Kafka生产和发送数据的方法请参考：DMS - 连接实例生产消费信息。发送样例数据如下：{"car_id":"3027", "car_owner":"lilei", "car_age":"7", "average_speed":"76", "total_miles":"15000"} {"car_id":"3028", "car_owner":"hanmeimei", "car_age":"6", "average_speed":"92", "total_miles":"17000"} {"car_id":"3029", "car_owner":"zhangsan", "car_age":"10", "average_speed":"81", "total_miles":"230000"}参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。执行以下命令连接DWS集群的默认数据库“testdwsdb”：gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r查询DWS的表数据。select * from test.qualified_cars;查询结果参考如下：car_id car_owner car_age average_speed total_miles 3027 lilei 7 76.0 15000.0

场景描述该场景为根据商品的实时点击量，获取每小时内点击量最高的3个商品及其相关信息，数据的输入源为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前提条件已创建DMS Kafka实例，具体步骤可参考：创建Kafka实例。注意：创建DMS Kafka实例时，不能开启Kafka SASL_SSL。已创建RDS MySQL实例，具体步骤可参考：RDS MySQL快速入门。本示例创建的RDS MySQL数据库版本选择为：8.0。整体作业开发流程整体作业开发流程参考图1。图1 作业开发流程步骤1：创建队列：创建DLI作业运行的队列。步骤2：创建Kafka的Topic：创建Kafka生产消费数据的Topic。步骤3：创建RDS数据库和表：创建RDS MySQL数据库和表信息。步骤4：创建增强型跨源连接：DLI上创建连接Kafka和RDS的跨源连接，打通网络。步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。步骤6：发送数据和查询结果：Kafka上发送流数据，在RDS上查看运行结果。步骤1：创建队列登录DLI管理控制台，在左侧导航栏单击“资源管理 > 队列管理”，可进入队列管理页面。在队列管理界面，单击界面右上角的“购买队列”。在“购买队列”界面，填写具体的队列配置参数，具体参数填写参考如下。计费模式：选择“包年/包月”或“按需计费”。本示例选择“按需计费”。区域和项目：保持默认值即可。名称：填写具体的队列名称。说明：新建的队列名称，名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。长度限制：1~128个字符。队列名称不区分大小写，系统会自动转换为小写。类型：队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”。AZ策略、CPU架构、规格：保持默认即可。企业项目：当前选择为“default”。高级选项：选择“自定义”。网段：配置队列网段。例如，当前配置为10.0.0.0/16。注意：队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合，否则后续创建跨源连接会失败。其他参数根据需要选择和配置。图2 创建队列参数配置完成后，单击“立即购买”，确认配置信息无误后，单击“提交”完成队列创建。步骤2：创建Kafka的Topic登录Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka实例名称，进入到Kafka实例的基本信息页面。单击“Topic管理 > 创建Topic”，创建一个Topic。Topic配置参数如下：Topic名称。本示例输入为：testkafkatopic。分区数：1。副本数：1其他参数保持默认即可。步骤3：创建RDS数据库和表登录RDS管理控制台，在“实例管理”界面，选择已创建的RDS MySQL实例，选择操作列的“更多 > 登录”，进入数据管理服务实例登录界面。输入实例登录的用户名和密码。单击“登录”，即可进入RDS MySQL数据库并进行管理。在数据库实例界面，单击“新建数据库”，数据库名定义为：testrdsdb，字符集保持默认即可。在已创建的数据库的操作列，单击“SQL查询”，输入以下创建表语句，创建RDS MySQL表。CREATE TABLE clicktop ( `range_time` VARCHAR(64) NOT NULL, `product_id` VARCHAR(32) NOT NULL, `product_name` VARCHAR(32), `event_count` VARCHAR(32), PRIMARY KEY (`range_time`,`product_id`) ) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8mb4;步骤4：创建增强型跨源连接创建DLI连接Kafka的增强型跨源连接在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。在“连接信息”中获取该Kafka的“内网连接地址”，在“基本信息”的“网络”中获取该实例的“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“网络”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_kafka。弹性资源池：选择步骤1：创建队列中已经创建的队列名称。虚拟私有云：选择Kafka的虚拟私有云。子网：选择Kafka的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的Kafka连接信息，地址栏输入“Kafka内网地址:Kafka数据库端口”，单击“测试”测试DLI到Kafka网络是否可达。创建DLI连接RDS的增强型跨源连接在RDS管理控制台，选择“实例管理”，单击对应的RDS实例名称，进入到RDS的基本信息页面。在“基本信息”的“连接信息”中获取该实例的“内网地址”、“数据库端口”、“虚拟私有云”和“子网”信息，方便后续操作步骤使用。单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选为：1，策略选为：允许，协议选择：TCP，端口值不填，类型：IPV4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。说明：本示例默认Kafka和RDS实例分别在两个VPC和子网下，所以要分别创建增强型跨源连接打通网络。如果Kafka和RDS实例属于同一VPC和子网下，则创建增强型跨源一次即可，4和5不需要再执行。在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。连接名称：设置具体的增强型跨源名称。本示例输入为：dli_rds。弹性资源池：选择步骤1：创建队列中已经创建的队列名称。虚拟私有云：选择RDS的虚拟私有云。子网：选择RDS的子网。其他参数可以根据需要选择配置。参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。在“测试连通性”界面，根据2中获取的RDS连接信息，地址栏输入“RDS内网地址:RDS数据库端口”，单击“测试”测试DLI到RDS网络是否可达。步骤5：运行作业在DLI管理控制台，单击“作业管理 > Flink作业”，在Flink作业管理界面，单击“创建作业”。在创建作业界面，作业类型选择“Flink OpenSource SQL”，名称填写为：FlinkKafkaRds。单击“确定”，跳转到Flink作业编辑界面。在Flink OpenSource SQL作业编辑界面，配置如下参数。所属队列：选择步骤1：创建队列中创建的队列。Flink版本：选择1.12。保存作业日志：勾选。OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。开启Checkpoint：勾选。Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。SQL中加粗的参数需要根据实际情况修改。说明：本示例使用的Flink版本为1.12，故Flink OpenSource SQL语法也是1.12。本示例数据源是Kafka，写入结果数据到RDS，故请参考Flink OpenSource SQL 1.12创建Kafka源表和Flink OpenSource SQL 1.12创建JDBC结果表（RDS连接）。create table click_product( user_id string, --点击用户的id user_name string, --用户名称 event_time string, --点击时间 product_id string, --商品id product_name string --商品名称 ) with ( "connector" = "kafka", "properties.bootstrap.servers" = "10.128.0.120:9092,10.128.0.89:9092,10.128.0.83:9092",--替换为kafka的内网连接地址和端口 "properties.group.id" = "click", "topic" = "testkafkatopic",--创建的Kafka Topic名称 "format" = "json", "scan.startup.mode" = "latest-offset" ); --结果表 create table top_product ( range_time string, --计算的时间范围 product_id string, --商品id product_name string, --商品名称 event_count bigint, --点击次数 primary key (range_time, product_id) not enforced ) with ( "connector" = "jdbc", "url" = "jdbc:mysql://192.168.12.148:3306/testrdsdb",--testrdsdb为创建的RDS的数据库名，IP和端口替换为RDS MySQL的实例IP和端口 "table-name" = "clicktop", "username" = "xxxxx", --替换为RDS MySQL的实例的用户名 "password" = "xxxxx", --替换为RDS MySQL的实例的用户密码 "sink.buffer-flush.max-rows" = "1000", "sink.buffer-flush.interval" = "1s" ); create view current_event_view as select product_id, product_name, count(1) as click_count, concat(substring(event_time, 1, 13), ":00:00") as min_event_time, concat(substring(event_time, 1, 13), ":59:59") as max_event_time from click_product group by substring (event_time, 1, 13), product_id, product_name; insert into top_product select concat(min_event_time, " - ", max_event_time) as range_time, product_id, product_name, click_count from ( select *, row_number() over (partition by min_event_time order by click_count desc) as row_num from current_event_view ) where row_num <= 3单击“语义校验”确保SQL语义校验成功。单击“保存”，保存作业。单击“启动”，启动作业，确认作业参数信息，单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。步骤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表数据已写入成功。

fusioninsight opensource flink 1.12 sql 作业中，怎么把kafka的数据接进来写入postgres中，尝试好多，一直sql校验失败。查资料没有示例

什么是微认证？          华为云微认证是基于线上学习与在线实践，快速获得场景化技能提升的认证。微认证清单 - 大数据使用DLI Flink SQL进行电商实时业务数据分析开发     课程简介：电商通常有web,小程序等多种接入方式,为掌握其实时变化，需统计各平台的实时访问量、订单数等，从而针对性地调整营销策略。     课程结构：电商实时业务应用场景介绍8认识电商常用的实时业务特点及应用电商实业业务对应大数据技术组件的原理47了解实现电商网站数据实时计算的相关大数据技术特性及原理华为云实时流计算Flink及解决方案7掌握华为云实时流计算Flink及解决方案及相应应用华为云实战案例15掌握华为云实时流计算Flink验流程及开发思路     1、电商实时业务应用场景介绍          电商从2009年发展至今,当前线上购物无疑是最火热的购物方式,而电商平台则又可以以多种方式接入,例如通过web方式访问、通过app的方式访问、通过微信小程序的方式访问等等。电商平台则需要每天统计各平台的实时访问数据量、订单数、访问人数等等          指标,从而能在显示大屏上实时展示相关数据,方便及时了解数据变化,有针对性地调整营销策略。这些高效快捷地统计指标是如何获得的呢?这是我们这次课程及实验所需要理解学习的          当前有很多电商的大数据平台会将每个商品的订单信息实时写入Kafka中,这些信息包括订单ID、订单生成的渠道(即web方式、app方式等)、订单时间、订单金额、折扣后实际支付金额、支付时间、用户ID、用户姓名、订单地区ID等信息。          针对业务场景,我们在大数据分析业务需要做的,就是根据当前可以获取到的业务数据,实时统计每种渠道的相关指标,输出存储到数据库中,并进行大屏展示。     2、电商实时业务对应大数据技术组件的原理        （1）流计算                概述         流式计算就像汽车过收费站,每一个车在通过闸口时都要收费。流式计算中每个实时产生的数据都要被实时的处理。        流计算秉承一个基本理念,即数据的价值随着时间的流逝而降低,如用户点击流。因此,当事件出现时就应该立即进行处理,而不是缓存起来进行离线处理。为了及时处理流数据,就需要一个低延迟、可扩展、高可靠的处理引擎。                应用场景                主要框架       Kafka        Kafka是最初由Linkedin公司开发,是一个分布式、分区的、多副本的、多订阅者,基于zookeepert办调的分布式日志系统。       主要应用场景是:日志收集系统和消息系统。        分布式消息传递基于可靠的消息队列,在客户端应用和消息系统之间异步传递消息。有两种主要的消息传递模式:点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。       Kafka就是一种发布-订阅模式。 Spark                Spark简介          2009年诞生于美国加州大学伯克利分校AMP实验室。          Apache Spark是一种基于内存的快速、通用、可扩展的大数据计算引擎。          Spark 是一站式解决方案，集批处理（Spark Core ）、实时流处理（Spark Streaming ）、交互式查询（Spark SQL ）、图计算（GraphX ）与机器学习（MLLib ）于一体。                  Spark应用场景           批处理可用于ETL (抽取、转换、加载)。          机器学习可用于自动判断淘宝的买家评论是好评还是差评。          交互式分析可用于查询Hive数据仓库。          流处理可用于页面点击流分析，推荐系统，舆情分析等实时业务。                 Spark架构                Spark特点                       SparkStreaming          Spark Streaming的基本原理是将实时输入数据流以时间片（秒级）为单位进行拆分，然后经Spark引擎以类似批处理的方式处理每个时间片数据。        Flink华为云数据可视化DLI          产品概述数据湖探索(Data Lake Insight,简称DLI)是完全兼容Apache Spark和Apache Flink生态,实现批流一体的Serverless大数据计算分析服务。DLI支持多模引擎,企业仅需使用SQL或程序就可轻松完成异构数据源的批处理、流处理、内存计算、机器学习等,挖掘和探索数据价值。          特点          应用场景：电商行业数据可视化          概述            广义:指一切能够把抽象、枯燥或难以理解的内容,包括看似毫无意义的数据、信息、知识等以一种容易理解的视觉方式展示出来的技术。            狭义:利用计算机图形学和图像处理技术,将数据转换为图形或图像在屏幕上显示出来,并进行各种交互处理的理论、方法和技术。          发展          工具        华为云数据可视化DLV          概述         数据可视化(Data Lake Visualization,简称DLV)是一站式数据可视化开发平台,适配云上云下多种数据源,提供丰富多样的2D、3D可视化组件,采用拖搜式自由布局。          特点          应用场景：某企业安全态势感知     3、华为云实时流计算Flink及解决方案        基于实时流计算的可视化解决方案        解决方案应用场景之智慧城市          智慧城市是通过对大量实时数据的监控、采集和处理,为复杂问题做出快速响应。智慧城市涉及范围很广,智慧城市建设主要包括政务、交通、企业、民生等方面。         解决方案应用场景之实时推荐          根据用户行为数据(包含历史数据和实时数据),通过构建的推荐模型对用户行为秒级调整并生成对应的推荐列表,分钟级更新候选集。          实时推荐主要包括广告推荐、商品推荐、视频推荐、游戏推荐等。     动手实验：         流程介绍 实验单独学习链接：华为云原生大数据serverless服务DLI_在线课程_华为云开发者学堂_云计算培训-华为云 (huaweicloud.com)

SQL 数据类型在介绍完一些基本概念之后，我们来认识一下，Flink SQL 中的数据类型。Flink SQL 内置了很多常见的数据类型，并且也为用户提供了自定义数据类型的能力。总共包含 3 部分：原子数据类型。复合数据类型。用户自定义数据类型。一、原子数据类型1、字符串类型：CHAR、CHAR(n)：定长字符串，就和 Java 中的 Char 一样，n 代表字符的定长，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。VARCHAR、VARCHAR(n)、STRING：可变长字符串，就和 Java 中的 String 一样，n 代表字符的最大长度，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。STRING 等同于 VARCHAR(2147483647)。2、二进制字符串类型：BINARY、BINARY(n)：定长二进制字符串，n 代表定长，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。VARBINARY、VARBINARY(n)、BYTES：可变长二进制字符串，n 代表字符的最大长度，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。BYTES 等同于 VARBINARY(2147483647)。3、 精确数值类型：DECIMAL、DECIMAL(p)、DECIMAL(p, s)、DEC、DEC(p)、DEC(p, s)、NUMERIC、NUMERIC(p)、NUMERIC(p, s)：固定长度和精度的数值类型，就和 Java 中的 BigDecima一样，p 代表数值位数(长度)，取值范围 [1, 38];s 代表小数点后的位数(精度)，取值范围 [0, p]。如果不指定，p 默认为 10，s 默认为 0。TINYINT：-128 到 127 的 1 字节大小的有符号整数，就和 Java 中的 byte 一样。SMALLINT：-32,768 to 32,767 的 2 字节大小的有符号整数，就和 Java 中的 short 一样。INT、INTEGER：-2,147,483,648 to 2,147,483,647 的 4 字节大小的有符号整数，就和 Java 中的 int 一样。BIGINT：-9,223,372,036,854,775,808 to 9,223,372,036,854,775,807 的 8 字节大小的有符号整数，就和 Java 中的 long 一样。4、有损精度数值类型：FLOAT：4 字节大小的单精度浮点数值，就和 Java 中的 float 一样。DOUBLE、DOUBLE PRECISION：8 字节大小的双精度浮点数值，就和 Java 中的 double 一样。关于 FLOAT 和 DOUBLE 的区别可见 https://www.runoob.com/w3cnote/float-and-double-different.html。5、布尔类型：BOOLEAN。6、NULL 类型：NULL。7、Raw 类型：RAW('class', 'snapshot') 。只会在数据发生网络传输时进行序列化，反序列化操作，可以保留其原始数据。以 Java 举例，class 参数代表具体对应的 Java 类型，snapshot 代表类型在发生网络传输时的序列化器。8、日期、时间类型：DATE：由 年-月-日 组成的 不带时区含义 的日期类型，取值范围 [0000-01-01, 9999-12-31]TIME、TIME(p)：由 小时：分钟：秒[.小数秒] 组成的 不带时区含义 的的时间的数据类型，精度高达纳秒，取值范围 [00:00:00.000000000到23:59:59.9999999]。其中 p 代表小数秒的位数，取值范围 [0, 9]，如果不指定 p，默认为 0。TIMESTAMP、TIMESTAMP(p)、TIMESTAMP WITHOUT TIME ZONE、TIMESTAMP(p) WITHOUT TIME ZONE：由 年-月-日 小时：分钟：秒[.小数秒] 组成的 不带时区含义 的时间类型，取值范围 [0000-01-01 00:00:00.000000000, 9999-12-31 23:59:59.999999999]。其中 p 代表小数秒的位数，取值范围 [0, 9]，如果不指定 p，默认为 6。TIMESTAMP WITH TIME ZONE、TIMESTAMP(p) WITH TIME ZONE：由 年-月-日 小时：分钟：秒[.小数秒] 时区 组成的 带时区含义 的时间类型，取值范围 [0000-01-01 00:00:00.000000000 +14:59, 9999-12-31 23:59:59.999999999 -14:59]。其中 p 代表小数秒的位数，取值范围 [0, 9]，如果不指定 p，默认为 6。TIMESTAMP_LTZ、TIMESTAMP_LTZ(p)：由 年-月-日 小时：分钟：秒[.小数秒] 时区 组成的 带时区含义 的时间类型，取值范围 [0000-01-01 00:00:00.000000000 +14:59, 9999-12-31 23:59:59.999999999 -14:59]。其中 p 代表小数秒的位数，取值范围 [0, 9]，如果不指定 p，默认为 6。TIMESTAMP_LTZ 与 TIMESTAMP WITH TIME ZONE 的区别在于：TIMESTAMP WITH TIME ZONE 的时区信息是携带在数据中的，举例：其输入数据应该是 2022-01-01 00:00:00.000000000 +08:00;TIMESTAMP_LTZ 的时区信息不是携带在数据中的，而是由 Flink SQL 任务的全局配置决定的，我们可以由 table.local-time-zone 参数来设置时区。INTERVAL YEAR TO MONTH、 INTERVAL DAY TO SECOND：interval 的涉及到的种类比较多。INTERVAL 主要是用于给 TIMESTAMP、TIMESTAMP_LTZ 添加偏移量的。举例，比如给 TIMESTAMP 加、减几天、几个月、几年。二、复合数据类型数组类型：ARRAY、t ARRAY。数组最大长度为 2,147,483,647。t 代表数组内的数据类型。举例 ARRAY、ARRAY，其等同于 INT ARRAY、STRING ARRAY。Map 类型：MAP。Map 类型就和 Java 中的 Map 类型一样，key 是没有重复的。举例 Map、Map。集合类型：MULTISET、t MULTISET。就和 Java 中的 List 类型，一样，运行重复的数据。举例 MULTISET，其等同于 INT MULTISET。对象类型：ROW、ROW、ROW(n0 t0, n1 t1, ...>、ROW(n0 t0 'd0', n1 t1 'd1', ...)。就和 Java 中的自定义对象一样。举例：ROW(myField INT, myOtherField BOOLEAN)，其等同于 ROW。三、用户自定义数据类型用户自定义类型就是运行用户使用 Java 等语言自定义一个数据类型出来。但是目前数据类型不支持使用 CREATE TABLE 的 DDL 进行定义，只支持作为函数的输入输出参数。