创建FlinkServer作业-华为云

MAPREDUCE服务 MRS-创建FlinkServer作业对接DWS表:操作场景

操作场景 FlinkServer支持对接8.1.x及之后版本的 GaussDB （DWS）数据库，本章节介绍GaussDB（DWS）作为Source表、Sink表以及维表的DDL定义，以及创建表时使用的WITH参数和代码示例，并指导如何在FlinkServer作业管理页面操作。FlinkSQL与GaussDB（DWS）数据类型对应关系如下表所示。本示例以安全模式FlinkServer、Kafka为例，对接安全模式GaussDB（DWS）。根据安全需求，FlinkServer界面回显FlinkSQL时，SQL中的“password”字段将显示为空，在回显状态下需要将密码信息补齐后再提交作业。本章节仅适用于 MRS 3.2.0至MRS 3.3.1版本及集群，MRS 3.3.1及之后的版本请参考创建FlinkServer作业写入数据至数据仓库服务(DWS)。

MAPREDUCE服务 MRS 创建FlinkServer作业

MAPREDUCE服务 MRS-创建FlinkServer作业对接DWS表:前提条件

前提条件需确保FlinkServer所在集群和GaussDB（DWS）所在集群网络互通，确保“可用区”、“虚拟私有云”、“安全组”配置相同。 FlinkServer所在集群（安全模式）：集群中已安装HDFS、Yarn、Kafka、ZooKeeper和Flink服务。包含Kafka服务的客户端已安装，安装路径如：/opt/client。参考创建FlinkServer权限角色创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：flinkuser。待对接的GaussDB（DWS）所在集群（安全模式）：可参考如下命令连接数据库并创建接受数据的表： gsql -d postgres -h IP -U username –p port -W password –r postgres：需要连接的数据库名称。 IP：GaussDB(DWS) 集群地址。如果通过公网地址连接，请指定为集群“公网访问域名 ”，如果通过内网地址连接，请指定为集群“内网访问域名”。如果通过弹性负载均衡连接，请指定为“弹性负载均衡地址”。 username和password：连接数据库的用户名及密码。命令中如果携带认证密码信息可能存在安全风险，在执行命令前建议关闭系统的history命令记录功能，避免信息泄露。 port ：Coordinator的端口号，请根据实际情况替换，可使用gs_om -t status --detail查询Coordinator数据路径，在该路径下的“postgresql.conf”文件中查看端口号信息。创建用于接受数据的空表，如表“customer_t1”： CREATE TABLE customer_t1 ( c_customer_sk INTEGER, c_customer_name VARCHAR(32) ) with (orientation = column,compression=middle) distribute by hash (c_customer_name);

MAPREDUCE服务 MRS 创建FlinkServer作业

MAPREDUCE服务 MRS-创建FlinkServer作业写入数据至Hive表:前提条件

前提条件集群已安装HDFS、Yarn、Kafka、Flink和Hive（且服务名称必须为Hive）等服务。包含Hive服务的客户端已安装，安装路径如：/opt/client。 Flink支持1.12.2及以后版本，Hive支持3.1.0及以后版本。参考创建FlinkServer权限角色创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：flink_admin。参考创建FlinkServer集群连接步骤中的“说明”获取访问Flink WebUI用户的客户端配置文件及用户凭据。

MAPREDUCE服务 MRS 创建FlinkServer作业

MAPREDUCE服务 MRS-创建FlinkServer流表源:新建FlinkServer流表步骤

新建FlinkServer流表步骤访问Flink WebUI，请参考访问FlinkServer WebUI界面。单击“流表管理”进入流表管理页面。单击“新建流表”，在新建流表页面参考表1填写信息，单击“确定”，完成流表创建。创建完成后，可在对应流表的“操作”列对流表进行编辑、删除等操作。图1 新建流表表1 新建流表信息参数名称参数描述备注流/表名称流/表的名称。例如：flink_sink 描述流/表的描述信息。 - 映射表类型 Flink SQL本身不带有数据存储功能，所有涉及表创建的操作，实际上均是对于外部数据表、存储的引用映射。类型包含Kafka、HDFS。 - 类型包含数据源表Source，数据结果表Sink。不同映射表类型包含的表如下所示。 Kafka：Source、Sink HDFS：Source、Sink - 数据连接选择数据连接。 - Topic 读取的Kafka的topic，支持从多个Kakfa topic中读取，topic之间使用英文分隔符进行分隔。 “映射表类型”选择“Kafka”时存在此参数。 - 文件路径要传输的HDFS目录或单个文件路径。 “映射表类型”选择“HDFS”时存在此参数。例如： “/user/sqoop/ ”或“/user/sqoop/example.csv” 编码选择不同“映射表类型”对应的编码如下： Kafka： CS V、JSON HDFS：CSV - 前缀 “映射表类型”选择“Kafka”，且“类型”选择“Source”，“编码”选择“JSON”时含义为：多层嵌套json的层级前缀，使用英文逗号(,)进行分隔。例如：data,info表示取嵌套json中data，info下的内容，作为json格式数据输入分隔符选择不同“映射表类型”对应的含义为：用于指定CSV字段分隔符。当数据“编码”为“CSV”时存在此参数。例如：“,” 行分隔符文件中的换行符，包含“\r”、“\n”、“\r\n”。 “映射表类型”选择“HDFS”时存在此参数。 - 列分隔符文件中的字段分隔符。 “映射表类型”选择“HDFS”时存在此参数。例如：“,” 流/表结构填写流/表结构，包含名称，类型。 - Proctime 指系统时间，与数据本身的时间戳无关，即在Flink算子内计算完成的时间。 “类型”选择“Source”时存在此参数。 - Event Time 指事件产生的时间，即数据产生时自带时间戳。 “类型”选择“Source”时存在此参数。 -

MAPREDUCE服务 MRS 创建FlinkServer作业前准备

MAPREDUCE服务 MRS-创建FlinkServer流表源:新建流表

新建流表访问Flink WebUI，请参考访问FlinkServer WebUI界面。单击“流表管理”进入流表管理页面。单击“新建流表”，在新建流表页面参考表1填写信息，单击“确定”，完成流表创建。创建完成后，可在对应流表的“操作”列对流表进行编辑、删除等操作。图1 新建流表表1 新建流表信息参数名称参数描述备注流/表名称流/表的名称。例如：flink_sink 描述流/表的描述信息。 - 映射表类型 Flink SQL本身不带有数据存储功能，所有涉及表创建的操作，实际上均是对于外部数据表、存储的引用映射。类型包含Kafka、HDFS。 - 类型包含数据源表Source，数据结果表Sink。不同映射表类型包含的表如下所示。 Kafka：Source、Sink HDFS：Source、Sink - 数据连接选择数据连接。 - Topic 读取的Kafka的topic，支持从多个Kakfa topic中读取，topic之间使用英文分隔符进行分隔。 “映射表类型”选择“Kafka”时存在此参数。 - 文件路径要传输的HDFS目录或单个文件路径。 “映射表类型”选择“HDFS”时存在此参数。例如： “/user/sqoop/ ”或“/user/sqoop/example.csv” 编码选择不同“映射表类型”对应的编码如下： Kafka：CSV、JSON HDFS：CSV - 前缀 “映射表类型”选择“Kafka”，且“类型”选择“Source”，“编码”选择“JSON”时含义为：多层嵌套json的层级前缀，使用英文逗号(,)进行分隔。例如：data,info表示取嵌套json中data，info下的内容，作为json格式数据输入分隔符选择不同“映射表类型”对应的含义为：用于指定CSV字段分隔符。当数据“编码”为“CSV”时存在此参数。例如：“,” 行分隔符文件中的换行符，包含“\r”、“\n”、“\r\n”。 “映射表类型”选择“HDFS”时存在此参数。 - 列分隔符文件中的字段分隔符。 “映射表类型”选择“HDFS”时存在此参数。例如：“,” 流/表结构填写流/表结构，包含名称，类型。 - Proctime 指系统时间，与数据本身的时间戳无关，即在Flink算子内计算完成的时间。 “类型”选择“Source”时存在此参数。 - Event Time 指事件产生的时间，即数据产生时自带时间戳。 “类型”选择“Source”时存在此参数。 -

MAPREDUCE服务 MRS 创建FlinkServer作业前准备

MAPREDUCE服务 MRS-创建FlinkServer作业对接DWS表:前提条件

前提条件需确保FlinkServer所在集群和GaussDB（DWS）所在集群网络互通，确保“可用区”、“虚拟私有云”、“安全组”配置相同。 FlinkServer所在集群（安全模式）：集群中已安装HDFS、Yarn、Kafka、ZooKeeper和Flink服务。包含Kafka服务的客户端已安装，安装路径如：/opt/client。参考创建FlinkServer权限角色创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：flinkuser。待对接的GaussDB（DWS）所在集群（安全模式）：可参考如下命令连接数据库并创建接受数据的表： gsql -d postgres -h IP -U username –p port -W password –r postgres：需要连接的数据库名称。 IP：GaussDB(DWS) 集群地址。如果通过公网地址连接，请指定为集群“公网访问域名”，如果通过内网地址连接，请指定为集群“内网访问域名”。如果通过弹性负载均衡连接，请指定为“弹性负载均衡地址”。 username和password：连接数据库的用户名及密码。命令中如果携带认证密码信息可能存在安全风险，在执行命令前建议关闭系统的history命令记录功能，避免信息泄露。 port ：Coordinator的端口号，请根据实际情况替换，可使用gs_om -t status --detail查询Coordinator数据路径，在该路径下的“postgresql.conf”文件中查看端口号信息。创建用于接受数据的空表，如表“customer_t1”： CREATE TABLE customer_t1 ( c_customer_sk INTEGER, c_customer_name VARCHAR(32) ) with (orientation = column,compression=middle) distribute by hash (c_customer_name);

MAPREDUCE服务 MRS 创建FlinkServer作业

MAPREDUCE服务 MRS-创建FlinkServer作业写入数据至ClickHouse表:FlinkSQL与ClickHouse数据类型对应关系

FlinkSQL与ClickHouse数据类型对应关系 FlinkSQL数据类型 ClickHouse数据类型 BOOLEAN UInt8 TINYINT Int8 SMALLINT Int16 INTEGER Int32 BIGINT Int64 FLOAT Float32 DOUBLE Float64 CHAR String VARCHAR String VARBINARY FixedString DATE Date TIMESTAMP DateTime DECIMAL Decimal

MAPREDUCE服务 MRS 创建FlinkServer作业

云服务器内容精选

创建FlinkServer作业

7*24

备案

专业服务

退订

建议反馈

售前咨询热线