DWS对接DLI Flink实现实时数据接入
当前实时数据的接入和处理在DWS承载的业务中所占的比例越来越大，DWS可以和DLI Flink结合，实现实时数据处理结果的报表展示。同时，DWS可以作为Flink维表引擎，存储维表数据，共Flink关联时使用。下面介绍DWS如何与DLI Flink相结合，实现实时数据的存储和查询。
一、创建DLI Flink作业
- 在DLI管理控制台的左侧导航栏中，单击“作业管理”>“Flink作业”，进入“Flink作业”页面。
- 在“Flink作业”页面右上角单击“创建作业”，弹出“创建作业”对话框。
图1 创建Flink SQL作业
3. 配置作业信息。
| 表1 作业配置信息 |
|
| 参数 |
参数说明 |
| 类型 |
选择“Flink SQL”：用户通过编辑SQL语句来启动作业。 |
| 名称 |
作业名称，只能由字母、中文、数字、中划线和下划线组成，并且长度为1～57字节。 说明： 作业名称必须是唯一的。 |
| 描述 |
作业的相关描述，长度为0～512字节。 |
| 模板名称 |
当编辑器选择“SQL编辑器”时，该参数有效。 用户可以选择样例模板或自定义的作业模板。关于模板的详细信息，请参见Flink模板管理。 |
| 标签 |
使用标签标识云资源。包括“标签键”和“标签值”。如果您需要使用同一标签标识多种云资源，即所有服务均可在标签输入框下拉选择同一标签，建议在标签管理服务（TMS）中创建预定义标签。具体请参考《标签管理服务用户指南》。 说明： o 最多支持10个标签。 o 一个“键”只能添加一个“值”。 o 标签键：在输入框中输入标签键名称。 说明： § 标签键的最大长度为36个字符 ，不能包含“=”,“*”,“,”,“<”,“>”,“\”,“|”,“/”，且首尾字符不能为空格。 § 若有预定义标签，可在输入框的下拉列表中进行选择。 o 标签值：在输入框中输入标签值。 说明： § 标签值的最大长度为43个字符，不能包含“=”,“*”,“,”,“<”,“>”,“\”,“|”,“/”，且首尾字符不能为空格。 § 若有预定义标签，可在输入框的下拉列表中进行选择。 |
4. 单击“确定”，进入作业“编辑”页面。
5. 编辑SQL作业。
在SQL语句编辑区域，输入详细的SQL语句。相关SQL语句请参考《数据湖探索SQL语法参考》。
6. 单击“语义校验”，确保语义校验成功。
-
- 只有语义校验成功后，才可以执行“调试”或“启动”作业的操作。
- 如果校验成功，提示“SQL语义校验成功”。
- 如果校验失败，会在错误的SQL语句前面显示红色的“X”记号，鼠标移动到“X”号上可查看详细错误，请根据错误提示修改SQL语句。
7. 设置作业运行参数。
图2 设置Flink SQL作业运行参数
| 表2 作业运行参数说明 |
|
| 参数 |
参数说明 |
| CU数量 |
CU数量为DLI的计算单元数量和管理单元数量总和，CU也是DLI的计费单位，1CU=1核4G。 |
| 管理单元 |
管理单元CU数量。 |
| 最大并行数 |
最大并行数是指同时运行Flink SQL作业的最大任务数。 说明： 最大并行数不能大于计算单元（CU数量-管理单元）的4倍。 |
| TaskManager配置 |
用于设置TaskManager资源参数。 勾选后需配置下列参数： o “单TM所占CU数”：每个TaskManager占用的资源数量。 o “单TM Slot”：每个TaskManager包含的Slot数量。 |
| 保存作业日志 |
设置是否将作业运行时的日志信息保存到OBS。日志信息的保存路径为：“桶名/jobs/logs/作业id开头的目录”。在作业列表中，单击对应的作业名称，在“运行日志”页签，可以单击页面提供的OBS链接跳转至对应的路径下。 勾选后需配置下列参数： “OBS桶”：选择OBS桶用于保存用户作业日志信息。如果选择的OBS桶是未授权状态，需要单击“OBS授权”。 说明： 如果同时勾选了“开启Checkpoint”和“保存作业日志”，OBS授权一次即可。 |
| 作业异常告警 |
设置是否将作业异常告警信息，如作业出现运行异常或者欠费情况，以SMN的方式通知用户。 勾选后需配置下列参数： “SMN主题”： 选择一个自定义的SMN主题。如何自定义SMN主题，请参见《消息通知服务用户指南》中“创建主题”章节。 |
| 开启Checkpoint |
设置是否开启作业快照，开启后可基于Checkpoint（一致性检查点）恢复作业。 勾选后需配置下列参数： o “Checkpoint间隔”：Checkpoint的时间间隔，单位为秒，输入范围 1~999999，默认值为10s。 o “Checkpoint 模式”：支持如下两种模式： § AtLeastOnce：事件至少被处理一次。 § ExactlyOnce：事件仅被处理一次。 o “OBS桶”：选择OBS桶用于保存用户Checkpoint。如果选择的OBS桶是未授权状态，需要单击“OBS授权”。 说明： 如果同时勾选了“开启Checkpoint”和“保存作业日志”，OBS授权一次即可。 |
| 异常自动重启 |
设置是否启动异常自动重启功能，当作业异常时将自动重启并恢复作业。 勾选后需配置下列参数： o “异常重试最大次数”：配置异常重试最大次数。单位为“次/小时”。 § 无限：无限次重试。 § 有限：自定义重试次数。 o “从Checkpoint恢复”：需要同时勾选“开启Checkpoint”才可配置该参数。 |
| 空闲状态保留时长 |
用于清除GroupBy或Window经过最大保留时间后仍未更新的中间状态，默认设置为1小时。 |
| 脏数据策略 |
选择处理脏数据的策略。支持如下三种策略：“忽略”，“抛出异常”和“保存”。 说明： “保存”是指将脏数据保存到OBS桶中。 |
| 脏数据转储地址 |
“脏数据策略”选择“保存”时，配置该参数。单击地址框选择保存脏数据的OBS路径。 |
| 所属队列 |
默认选择“共享队列”，用户也可以选择自定义的独享队列。 选择“独享队列”时需配置以下参数： “UDF Jar”：用户自定义UDF文件，在选择UDF Jar之前需要将对应的jar包上传至OBS桶中，并在“数据管理>程序包管理”中创建程序包，具体操作请参考创建程序包。 用户可以在SQL中调用插入Jar包中的自定义函数。 说明： o 当子用户在创建作业时，子用户只能选择已经被分配的队列。 o 当所选择队列的剩余容量不能满足作业需求时，系统会自动扩容，将按照增加的容量计费。当队列空闲时，系统也会自动缩容。 |
- （可选）根据需要调试参数。作业调试功能只用于验证SQL逻辑，不会有数据写入操作。 具体操作请参见调试作业。
- 单击“保存”，保存作业和相关参数。
二. 编写flink sql读取和写入数据
1. 以flink从kafka读取数据写入DWS为例，sql语句如下：
第一步：创建数据源,一般是kafka源。
第二步：创建维表
Flink还支持以GaussDB(DWS)作为维表引擎，实现flink表和维表的关联操作，再将数据sink到目标表。DWS维表创建语法如下。
create table dimtion_tbl (id bigint,
name varchar,
address varchar)
with (
"connector.type" = "gaussdb",
"connector.url" = "jdbc:postgresql://xx.xx.xx.xx:8000/db_name",
"connector.table" = "table_name",
"connector.username" = "user_name",
"connector.password" = "password",
"connector.lookup.cache.max-rows" = "200000",
"connector.lookup.cache.ttl" = "1h"
);
其中需要注意的是：
connector.lookup.cache.max-rows需要根据flink的维表cache大小来确定，即用cache大小除以表数据每行平均大小，即为缓存行数。维表一般都较小，因此尽量将所有数据都缓存在flink中，减小对DWS的压力。
connector.lookup.cache.ttl：根据实际情况设置，一般设置为与原平台维表一致即可。
此外，Flink会对维表进行高并发点查，需要在维表的关联字段上建立B-tree索引。
第三步：创建目标表
参数解释：
| 参数 |
是否必选 |
说明 |
| connector.type |
是 |
connector类型，对于dws(gaussdb)，需配置为'gaussdb' |
| connector.url |
是 |
jdbc连接地址，格式为：jdbc:postgresql://${ip}:${port}/${dbName} |
| connector.table |
是 |
操作的表名 |
| connector.username |
否 |
数据库认证用户名，需和'connector.password'一起配置 |
| connector.password |
否 |
数据库认证密码，需和'connector.username'一起配置 |
| connector.driver |
否 |
jdbc连接驱动，默认为: org.postgresql.Driver |
| connector.read.partition.column |
否 |
source配置，用于对输入进行分区的列名 |
| connector.read.partition.num |
否 |
source配置，输入分区的个数 |
| connector.read.partition.lower-bound |
否 |
source配置，第一个分区的最小值 |
| connector.read.partition.upper-bound |
否 |
source配置，最后一个分区的最大值 |
| connector.read.fetch-size |
否 |
source配置，每次从数据库拉取数据的行数。默认值为0，表示忽略该提示 |
| connector.lookup.cache.max-rows |
否 |
维表配置，缓存的最大行数，超过该值时，老的数据会被踢除。-1表示不使用缓存 |
| connector.lookup.cache.ttl |
否 |
维表配置，缓存超时时间，超过该时间的数据会被剔除。格式为：{length value}{time unit label}，如123ms, 321s，支持的时间单位包括: d,h,min,s,ms等，默认为ms |
| connector.lookup.max-retries |
否 |
维表配置，数据拉取最大重试次数，默认为3 |
| connector.write.mode |
否 |
sink配置，数据写入模式，支持: copy, insert以及upsert三种。与'primary key'配合使用。 未配置'primary key'时：支持copy及insert两种追加写入，其中copy方式为批量写入，insert为单条写入。 配置'primary key'时：，支持copy&merge、upsert以及insert&update三种更新写入。注意：由于dws不支持更新分布列，因而配置的更新主键必须包含dws表中定义的所有分布列 |
| connector.write.flush.max-rows |
否 |
sink配置，数据flush大小，超过该值将触发写入flush。默认为5000 |
| connector.write.flush.interval |
否 |
sink配置，数据flush周期，周期性触发写入flush。格式为：{length value}{time unit label}，如123ms, 321s，支持的时间单位包括: d,h,min,s,ms等，默认为ms |
| connector.write.max-retries |
否 |
sink配置，写入最大重试次数，默认为3 |
| connector.write.merge.filter-key |
否 |
sink配置，copy&merge模式下，merge时过滤列名 |
| connector.write.escape-string-value |
否 |
sink配置，是否对string类型值进行转义，默认为false |
其中标黄的三个参数会影响写入DWS的方式和性能。
第四步：插入数据语句
插入语句一般是insert into 目标表 select 业务逻辑 from源表 join 维表。
三. 启动作业
单击“启动”，进入“启动Flink作业”页面，确认作业规格和费用后，单击“立即启动”，启动作业。
启动作业后，系统将自动跳转到Flink作业管理页面，新创建的作业将显示在作业列表中，在“状态”列中可以查看作业状态。作业提交成功后，状态将由“提交中”变为“运行中”。运行完成后显示“已完成”。
如果作业状态为“提交失败”或“运行异常”，表示作业提交或运行失败。用户可以在作业列表中的“状态”列中，将鼠标移动到状态图标上查看错误信息，单击 可以复制错误信息。根据错误信息解决故障后，重新提交。
- 点赞
- 收藏
- 关注作者
评论（0）