华为云用户手册

操作流程 图1 操作流程图 表1 资源规划 资源名称 资源类型 说明 ELB-Test 独享型ELB 独享型ELB支持高级转发策略，因此需创建独享型ELB实例。 Server_Group-Test01 后端服务器组 用于管理部署了旧版本业务的E CS 。 Server_Group-Test02 后端服务器组 用于管理部署了新版本业务的ECS。 ECS01 弹性云服务器 上面部署了旧版本业务，添加至Server_Group-Test01。 ECS02 弹性云服务器 上面部署了旧版本业务，添加至Server_Group-Test01。 ECS03 弹性云服务器 上面部署了旧版本业务，添加至Server_Group-Test01。 ECS04 弹性云服务器 上面部署了新版本业务，添加至Server_Group-Test02。 ECS05 弹性云服务器 上面部署了新版本业务，添加至Server_Group-Test02。 ECS06 弹性云服务器 上面部署了新版本业务，添加至Server_Group-Test02。 本最佳实践中，独享型ELB和ECS均在同一VPC中。在实际应用中，如果您的ECS和ELB不在同一VPC中，可以跨VPC添加ECS至ELB的后端服务器组中，详细请参考跨VPC添加服务器至负载均衡器。

创建后端服务器组并添加后端服务器 单击上述创建的独享型负载均衡ELB-Test名称。 在监听器页签，单击右上角的“创建后端服务器组”。 名称：Server_Group-Test01 后端协议：HTTP 其他参数根据需要设置。 参考步骤2再添加后端服务器组Server_Group-Test02。 图4 添加后端服务器组 单击后端服务器组Server_Group-Test01名称，添加ECS01、ECS02、ECS03至Server_Group-Test01。 单击后端服务器组Server_Group-Test02名称，添加ECS04、ECS05、ECS06至Server_Group-Test02。

基于HTTP请求方法实现新旧版本应用平滑过渡 通过配置转发规则为“HTTP请求方法”的高级转发策略，实现将来自客户端的GET和DELETE请求转发至旧版本应用上，将来自客户端的POST和PUT请求转发至新版本应用上。以此种方式运行一段时间后，确认新版本应用无问题后，再将所有请求全部切换至新版本应用。 图5 基于HTTP请求方法实现新旧版本应用平滑过渡 单击上述创建的独享型负载均衡ELB-Test名称。 在“监听器”页签，单击上述创建的HTTP监听器名称。 切换至右边的“转发策略”页面，单击“添加转发策略”。 转发至旧版本应用：在下拉列表中选择“HTTP请求方法”，选择“GET”和“DELETE”，动作选择“转发至后端服务器组”，后端服务器组选择“Server_Group-Test01”。 图6 基于HTTP请求方法将部分请求转发至旧版本应用上 单击“保存”。 参考步骤3和步骤4再添加一个转发策略，实现将请求转发至新版本应用上。 转发至新版本应用：在下拉列表中选择“HTTP请求方法”，选择“PUT”和“POST”，动作选择“转发至后端服务器组”，后端服务器组选择“Server_Group-Test02”。 图7 基于HTTP请求方法将部分请求转发至新版本应用上

步骤七：验证通过跨VPC后端功能添加同VPC后端服务器组是否成功 单击上述创建的独享型负载均衡ELB-Test操作列的“更多”。 选择“绑定IPv4公网IP”，给ELB-Test绑定一个弹性公网IP（EIP-Test：120.46.131.153）。 图10 ELB绑定EIP 使用浏览器访问“http://120.46.131.153/”，显示如下页面，说明本次访问请求被ELB实例转发到后端服务器“ECS-Test”上，“ECS-Test”正常处理请求并返回请求的页面。 图11 验证通过跨VPC后端功能添加同VPC后端服务器是否成功

资源和成本规划 资源的实际费用以华为云管理控制台显示为准。 表1 资源规划 资源 资源名称 资源说明 数量 每月费用（元） VPC vpc-peering 创建独享型负载均衡（ELB-Test）和ECS-Test所在VPC： 规划网段：10.1.0.0/16 1 00.00 对等连接 Peering-Test 在ELB所在的VPC和云上其他VPC之间建立对等连接 本端VPC网段：10.1.0.0/16 对端VPC网段：任选 1 00.00 路由表 Route-VPC-Test-01 创建对等连接路由，所属VPC：VPC-Test-01 目的地址：10.1.0.0/16 1 00.00 ELB ELB-Test 独享型负载均衡（ELB-Test） 私网IP：10.1.0.9 1 400 本例使用的是华东-上海一， 应用型(HTTP/HTTPS) | 小型 I 网络型(TCP/UDP) | 小型 I EIP EIP-Test 用于给ELB-Test绑定的弹性公网IP 120.46.131.153 1 115 本例使用的是华东-上海一，带宽5M。 ECS ECS-Test ECS所属VPC：vpc-peering 私网IP：10.1.0.56 1 267.78 本例使用的是华东-上海一，c7.large.2，CentOS7.6操作系统的云服务器。包含系统盘价格。

步骤三：添加对等连接路由 在虚拟私有云控制台单击左侧“路由表”。 单击右上角的“创建路由表”。 根据表1创建路由表Route-VPC-Test-01，完成后单击“确定”。详见《虚拟私有云用户指南》。 路由表名称：Route-VPC-Test-01 所属VPC：vpc-peering 目的地址：10.1.0.0/16 下一跳类型：对等连接 下一跳：Peering-Test 图5 创建路由表Route-VPC-Test-01

步骤六：将ECS添加至ELB后端服务器组 单击上述创建的独享型负载均衡ELB-Test名称。 切换到“监听器”页签，单击上述所创建的HTTP监听器。 切换至右边的“后端服务器组”页签，单击“跨VPC后端”。 单击“添加跨VPC后端”，设置相关参数，完成后单击“确定”。详见《弹性负载均衡用户指南》。 跨VPC后端IP：10.1.0.56（ECS-Test的私网IP） 后端端口：填写业务端口 权重：根据需要设置 图9 添加跨VPC后端

步骤七：放行独享引擎回源IP 网站以“独享模式”成功接入WAF后，所有网站访问请求将先经过负载均衡器然后流转到独享引擎实例进行监控，经独享引擎实例过滤后再返回到源站服务器，流量经独享引擎实例返回源站的过程称为回源。 在服务器看来，接入WAF后所有源IP都会变成独享引擎实例的回源IP（即独享引擎实例对应的子网IP），以防止源站IP暴露后被黑客直接攻击。 源站服务器上的安全软件很容易认为独享引擎的回源IP是恶意IP，有可能触发屏蔽WAF回源IP的操作。一旦WAF的回源IP被屏蔽，WAF的请求将无法得到源站的正常响应，因此，网站以“独享模式”接入WAF防护后，您需要在源站服务器上设置放行创建的独享引擎实例对应的子网IP，不然可能会出现网站打不开或打开极其缓慢等情况。 详细操作步骤请参考回源到ELB。

操作场景 HTTPS是加密数据传输协议，安全性高，如果您需要保证业务建立安全连接，可以通过负载均衡的HTTP重定向功能，将HTTP访问重定向至HTTPS。 该功能可以满足您如下需求，PC、手机浏览器等以HTTP请求访问Web服务，配置了HTTP访问重定向至HTTPS后，后端服务器返回HTTPS的响应。默认强制以HTTPS访问网页。 因为HTTP标准协议只支持GET和HEAD方法的重定向，所以设置了HTTP重定向至HTTPS后，POST和其他方法会被改为GET方法，这是客户端浏览器的行为，而非ELB修改的。如果您需要实现除GET和HEAD方法以外的访问方式，建议直接使用HTTPS方式进行访问。 HTTP重定向至HTTPS是指所有的HTTP请求都将转给HTTPS监听器处理为HTTPS请求，但HTTPS请求是通过HTTP被发送给后端服务器的。 HTTP监听器重定向至HTTPS监听器，HTTPS监听器所关联的后端服务器上不能再安装证书，否则会引起HTTPS请求不生效。

使用OpenSSL制作CA证书 登录到任意一台安装有openssl工具的Linux机器。 创建工作目录并进入该目录。 mkdir ca cd ca 创建CA证书的openssl配置文件ca_cert.conf，内容如下： [ req ] distinguished_name = req_distinguished_name prompt = no [ req_distinguished_name ] O = ELB 创建CA证书私钥文件ca.key。 openssl genrsa -out ca.key 2048 图1 生成CA证书私钥文件 创建CA证书的csr请求文件ca.csr。 openssl req -out ca.csr -key ca.key -new -config ./ca_cert.conf 创建自签名的CA证书ca.crt。 openssl x509 -req -in ca.csr -out ca.crt -sha1 -days 5000 -signkey ca.key 图2 创建自签名CA证书

使用CA证书签发客户端证书 登录到生成CA证书的服务器。 创建与CA平级的目录，并进入该目录。 mkdir client cd client 创建客户端证书的openssl配置文件client_cert.conf，内容如下： [ req ] distinguished_name = req_distinguished_name prompt = no [ req_distinguished_name ] O = ELB CN = www.test.com CN字段可以根据需求改为对应的 域名 、IP地址。 创建客户端证书私钥文件client.key。 openssl genrsa -out client.key 2048 图4 创建客户端证书私钥文件 创建客户端证书的csr请求文件client.csr。 openssl req -out client.csr -key client.key -new -config ./client_cert.conf 图5 创建客户端证书csr文件 使用CA证书签发客户端证书client.crt。 openssl x509 -req -in client.csr -out client.crt -sha1 -CAcreateserial -days 5000 -CA ../ca/ca.crt -CAkey ../ca/ca.key 图6 签发客户端证书 把客户端证书格式转为浏览器可识别的p12格式。 openssl pkcs12 -export -clcerts -in client.crt -inkey client.key -out client.p12 该命令执行时需要输入导出密码，请输入并记住该密码，在证书导入浏览器时需要使用。

使用CA证书签发服务器证书 用户可以用权威CA签发的证书或者自签名的证书，这里以自签名证书为例说明如何创建服务器证书。 登录到生成CA证书的服务器。 创建与CA平级的目录，并进入该目录。 mkdir server cd server 创建服务器证书的openssl配置文件server_cert.conf，内容如下： [ req ] distinguished_name = req_distinguished_name prompt = no [ req_distinguished_name ] O = ELB CN = www.test.com CN字段可以根据需求改为服务器对应的域名、IP地址。 创建服务器证书私钥文件server.key。 openssl genrsa -out server.key 2048 创建服务器证书的csr请求文件server.csr。 openssl req -out server.csr -key server.key -new -config ./server_cert.conf 使用CA证书签发服务器证书server.crt。 openssl x509 -req -in server.csr -out server.crt -sha1 -CAcreateserial -days 5000 -CA ../ca/ca.crt -CAkey ../ca/ca.key 图3 签发服务器证书

使用场景 一般的HTTPS业务场景只对服务器做认证，因此只需要配置服务器的证书即可。某些关键业务（如银行支付），需要对通信双方的身份都要做认证，即双向认证，以确保业务的安全性。 此时，除了配置服务器的证书之外，还需要配置客户端的证书，以实现通信双方的双向认证功能。 本章节以自签名证书为例，介绍如何配置HTTPS双向认证。但是自签名证书存在安全隐患，建议客户使用 云证书管理服务 购买证书、或购买其他权威机构颁发的证书。

数据类型映射 表5 数据类型映射 DWS数据类型 Flink SQL数据类型 BOOLEAN BOOLEAN SMALLINT(INT2) SMALLSERIAL(SERIAL2) SMALLINT INTEGER SERIAL INTEGER BIGINT BIGSERIAL BIGINT REAL FLOAT4 FLOAT DOUBLE FLOAT8 DOUBLE CHAR(n) CHAR VARCHAR(n) VARCHAR DATE DATE TIMESTAMP[(p)] [WITHOUT TIME ZONE] TIMESTAMP NUMERIC[(p[,s])] DECIMAL[(p[,s])] DECIMAL

键处理 当写入数据到外部数据库时，Flink 会使用 DDL 中定义的主键。如果定义了主键，则连接器将以 upsert 模式工作，否则连接器将以 append 模式工作。 在 upsert 模式下，Flink 将根据主键判断插入新行或者更新已存在的行，这种方式可以确保幂等性。为了确保输出结果是符合预期的，推荐为表定义主键并且确保主键是底层数据库中表的唯一键或主键。在 append 模式下，Flink 会把所有记录解释为 INSERT 消息，如果违反了底层数据库中主键或者唯一约束，INSERT 插入可能会失败。

Lookup Cache 该连接器可以用在时态表关联中作为一个可 lookup 的 source (又称为维表)，当前只支持同步的查找模式。 默认情况下，lookup cache 是未启用的，你可以设置 lookup.cache.max-rows and lookup.cache.ttl 参数来启用。 lookup cache 的主要目的是用于提高时态表关联该 连接器的性能。默认情况下，lookup cache 不开启，所以所有请求都会发送到外部数据库。 当 lookup cache 被启用时，每个进程（即 TaskManager）将维护一个缓存。Flink 将优先查找缓存，只有当缓存未查找到时才向外部数据库发送请求，并使用返回的数据更新缓存。 当缓存命中最大缓存行 lookup.cache.max-rows 或当行超过最大存活时间 lookup.cache.ttl 时，缓存中最先添加的条目将被标记为过期。 缓存中的记录可能不是最新的，用户可以将 lookup.cache.ttl 设置为一个更小的值以获得更好的刷新数据，但这可能会增加发送到数据库的请求数。所以要做好吞吐量和正确性之间的平衡。 默认情况下，flink 会缓存主键的空查询结果，你可以通过将 lookup.cache.caching-missing-key 设置为 false 来切换行为。

示例 该示例是从kafka数据源中读取数据，并将DWS的表作为维表，然后将结果写入到DWS的另一张表中。 在DWS中创建相应的表，表名为area_info作为flink维表，SQL语句参考如下。 create table public.area_info( area_id VARCHAR, area_province_name VARCHAR, area_city_name VARCHAR, area_county_name VARCHAR, area_street_name VARCHAR, region_name VARCHAR); 在DWS中执行以下SQL语句，向dws_order表中插入数据。 insert into public.area_info (area_id, area_province_name, area_city_name, area_county_name, area_street_name, region_name) values ('330102', 'a1', 'b1', 'c1', 'd1', 'e1'), ('330106', 'a1', 'b1', 'c2', 'd2', 'e1'), ('330108', 'a1', 'b1', 'c3', 'd3', 'e1'), ('330110', 'a1', 'b1', 'c4', 'd4', 'e1'); 在DWS中创建相应的表，表名为order_detail作为flink结果表，SQL语句参考如下。 create table public.order_detail( 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, area_province_name VARCHAR, area_city_name VARCHAR, area_county_name VARCHAR, area_street_name VARCHAR, region_name VARCHAR); 参考增强型跨源连接，根据DWS所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置DWS的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根据DWS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本。 如下脚本中的加粗参数请根据实际环境修改。 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, proctime as Proctime() ) WITH ( 'connector' = 'kafka', 'topic' = 'kafkaTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort', 'properties.group.id' = 'GroupId', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ); CREATE TABLE dwsLookUp ( area_id VARCHAR, area_province_name VARCHAR, area_city_name VARCHAR, area_county_name VARCHAR, area_street_name VARCHAR, region_name VARCHAR ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:gaussdb://DWSIP:DWSPort/DWSdbName', 'table-name' = 'area_info', 'username' = 'xx', 'password' = 'xx', -- dew存储的密码的key，其值为dws密码 'dew.endpoint' = 'kms.xx.myhuaweicloud.com', --使用的DEW服务所在的endpoint信息 'dew.csms.secretName' = 'xx', --DEW服务通用凭据的凭据名称 'dew.csms.decrypt.fields' = 'password', --其中password字段值需要利用DEW凭证管理,进行解密替换 'dew.csms.version' = 'v1' ); 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, area_province_name string, area_city_name string, area_county_name string, area_street_name string, region_name string ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:gaussdb://DWSIP:DWSPort/DWSdbName', 'table-name' = 'order_detail', 'username' = 'xx', 'password' = 'xx', -- dew存储的密码的key，其值为dws密码 'dew.endpoint' = 'kms.xx.myhuaweicloud.com', --使用的DEW服务所在的endpoint信息 'dew.csms.secretName' = 'xx', --DEW服务通用凭据的凭据名称 'dew.csms.decrypt.fields' = 'password', --其中password字段值需要利用DEW凭证管理,进行解密替换 'dew.csms.version' = 'v1' ); insert into dwsSink select orders.order_id, orders.order_channel, orders.order_time, orders.pay_amount, orders.real_pay, orders.pay_time, orders.user_id, orders.user_name, area.area_id, area.area_province_name, area.area_city_name, area.area_county_name, area.area_street_name, area.region_name from kafkaSource as orders left join dwsLookUp for system_time as of orders.proctime as area on orders.area_id = area.area_id; 向Kafka的源表的topic中发送如下数据: {"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":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"} 登录DWS，执行如下SQL查询结果表中的数据： select * from public.order_detail; 数据结果参考如下： 202103241606060001 appShop 2021-03-24 16:06:06 200.0 180.0 2021-03-24 16:10:06 0001 Alice 330106 a1 b1 c2 d2 e1 202103251202020001 miniAppShop 2021-03-25 12:02:02 60.0 60.0 2021-03-25 12:03:00 0002 Bob 330110 a1 b1 c4 d4 e1

常见问题 Q：作业运行失败，运行日志中有如下报错信息，应该怎么解决？ java.io.IOException: unable to open JDBC writer ... Caused by: org.postgresql.util.PSQLException: The connection attempt failed. ... Caused by: java.net.SocketTimeoutException: connect timed out A：应考虑是跨源没有绑定，或者跨源没有绑定成功。 参考增强型跨源连接章节，重新配置跨源。参考 DLI 跨源连接DWS失败进行问题排查。

分区扫描 为了在并行 Source task 实例中加速读取数据，Flink 提供了分区扫描的特性。 如果下述分区扫描参数中的任一项被指定，则下述所有的分区扫描参数必须都被指定。这些参数描述了在多个 task 并行读取数据时如何对表进行分区。 scan.partition.column 必须是相关表中的数字、日期或时间戳列。注意，scan.partition.lower-bound 和 scan.partition.upper-bound 用于决定分区的起始位置和过滤表中的数据。如果是批处理作业，也可以在提交 flink 作业之前获取最大值和最小值。 scan.partition.column：输入用于进行分区的列名。 scan.partition.num：分区数。 scan.partition.lower-bound：第一个分区的最小值。 scan.partition.upper-bound：最后一个分区的最大值。

维表参数说明 表4 维表参数说明 参数 是否必选 默认值 数据类型 说明 connector 是 无 String 指定使用什么类型的连接器，这里应该是'jdbc'。 url 是 无 String DWS连接地址。“url”参数中的ip地址请使用DWS的内网地址。 jdbc:gaussdb://${ip}:${port}/${dbName}。 table-name 是 无 String 操作的DWS表名。如果该DWS表在某schema下，则具体可以参考如果该DWS表在某schema下的说明。 driver 否 无 String 用于连接到此 URL 的 DWS 驱动类名，如果不设置，将自动从 URL 中推导。 可配置com.huawei.gauss200.jdbc.Driver。 username 否 无 String DWS数据库认证用户名，需要和'password'参数一起配置。 password 否 无 String DWS数据库认证密码，需要和'username'参数一起配置。 connection.max-retry-timeout 否 60s Duration 最大重试超时时间，以秒为单位且不应该小于 1 秒。 scan.partition.column 否 无 String 用于对输入进行分区的列名。 scan.partition.column 否 无 Integer 分区数。 scan.partition.lower-bound 否 无 Integer 第一个分区的最小值。 scan.partition.upper-bound 否 无 Integer 最后一个分区的最大值。 scan.fetch-size 否 0 Integer 每次循环读取时应该从数据库中获取的行数。如果指定的值为 '0'，则该配置项会被忽略。 scan.auto-commit 否 true Boolean 在 JDBC 驱动程序上设置 auto-commit 标志， 它决定了每个语句是否在事务中自动提交。Postgres可能需要将此设置为 false 以便流化结果。 lookup.cache.max-rows 否 无 Integer lookup cache 的最大行数，如果超过该值，缓存中最先添加的条目将被标记为过期。 默认情况下，lookup cache 是未开启的。请参阅下面的Lookup Cache 部分了解更多详情。 lookup.cache.ttl 否 无 Duration lookup cache 中每一行记录的最大存活时间，如果超过该时间，缓存中最先添加的条目将被标记为过期。 默认情况下，lookup cache 是未开启的。请参阅下面的Lookup Cache部分了解更多详情。 lookup.cache.caching-missing-key 否 true Boolean 标记缓存丢失的键，默认为true lookup.max-retries 否 3 Integer 查询数据库失败的最大重试时间。

结果表参数说明 表3 结果表参数说明 参数 是否必选 默认值 数据类型 说明 connector 是 无 String 指定使用什么类型的连接器，这里应该是'jdbc'。 url 是 无 String DWS连接地址。“url”参数中的ip地址请使用DWS的内网地址。 jdbc:gaussdb://${ip}:${port}/${dbName}。 table-name 是 无 String 操作的DWS表名。如果该DWS表在某schema下，则具体可以参考如果该DWS表在某schema下的说明。 driver 否 无 String 用于连接到此 URL 的 DWS 驱动类名，如果不设置，将自动从 URL 中推导。 可配置com.huawei.gauss200.jdbc.Driver。 username 否 无 String DWS数据库认证用户名，需要和'password'参数一起配置。 password 否 无 String DWS数据库认证密码，需要和'username'参数一起配置。 connection.max-retry-timeout 否 60s Duration 最大重试超时时间，以秒为单位且不应该小于 1 秒。 sink.buffer-flush.max-rows 否 100 Integer flush 前缓存记录的最大值，可以设置为 '0' 来禁用它。 sink.buffer-flush.interval 否 1s Duration flush 间隔时间，超过该时间后异步线程将 flush 数据。可以设置为 '0' 来禁用它。注意, 为了完全异步地处理缓存的 flush 事件，可以将 'sink.buffer-flush.max-rows' 设置为 '0' 并配置适当的 flush 时间间隔。 sink.max-retries 否 3 Integer 写入记录到数据库失败后的最大重试次数。 sink.parallelism 否 无 Integer 用于定义sink 算子的并行度。默认情况下，并行度是由框架决定：使用与上游链式算子相同的并行度。

前提条件 请务必确保您的账户下已在 数据仓库 服务（DWS）里创建了DWS集群。 如何创建DWS集群，请参考创建集群。 请确保已创建DWS数据库表。 该场景作业需要运行在DLI的独享队列上，因此要与DWS集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《 数据湖探索 用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

语法格式 1 2 3 4 5 6 7 8 9 10 11 12 13 create table dwsTable ( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) (',' watermark for rowtime_column_name as watermark-strategy_expression) ) with ( 'connector' = 'gaussdb', 'url' = '', 'table-name' = '', 'username' = '', 'password' = '' );

源表参数说明 表2 源表参数说明 参数 是否必选 默认值 数据类型 说明 connector 是 无 String 指定使用什么类型的连接器，这里应该是'jdbc'。 url 是 无 String DWS连接地址。“url”参数中的ip地址请使用DWS的内网地址。 jdbc:gaussdb://${ip}:${port}/${dbName}。 table-name 是 无 String 操作的DWS表名。如果该DWS表在某schema下，则具体可以参考如果该DWS表在某schema下的说明。 driver 否 无 String 用于连接到此 URL 的 DWS 驱动类名，如果不设置，将自动从 URL 中推导。 可配置com.huawei.gauss200.jdbc.Driver。 username 否 无 String DWS数据库认证用户名，需要和'password'参数一起配置。 password 否 无 String DWS数据库认证密码，需要和'username'参数一起配置。 connection.max-retry-timeout 否 60s Duration 最大重试超时时间，以秒为单位且不应该小于 1 秒。 scan.partition.column 否 无 String 用于对输入进行分区的列名。 scan.partition.column 否 无 Integer 分区数。 scan.partition.lower-bound 否 无 Integer 第一个分区的最小值。 scan.partition.upper-bound 否 无 Integer 最后一个分区的最大值。 scan.fetch-size 否 0 Integer 每次循环读取时应该从数据库中获取的行数。如果指定的值为 '0'，则该配置项会被忽略。 scan.auto-commit 否 true Boolean 在 JDBC 驱动程序上设置 auto-commit 标志， 它决定了每个语句是否在事务中自动提交。Postgres可能需要将此设置为 false 以便流化结果。

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 with参数中字段只能使用单引号，不能使用双引号。

在PV中设置挂载参数 在PV中设置挂载参数可以通过mountOptions字段实现，如下所示，mountOptions支持挂载的字段请参见极速文件存储挂载参数。 使用kubectl连接集群，详情请参见通过kubectl连接集群。 在PV中设置挂载参数，示例如下： apiVersion: v1 kind: PersistentVolume metadata: annotations: pv.kubernetes.io/provisioned-by: everest-csi-provisioner name: pv-sfsturbo # PV的名称 spec: accessModes: - ReadWriteMany # 访问模式，极速文件存储必须为ReadWriteMany capacity: storage: 500Gi # 极速文件存储容量大小 csi: driver: sfsturbo.csi.everest.io # 挂载依赖的存储驱动 fsType: nfs volumeHandle: {your_volume_id} # 极速文件存储的ID volumeAttributes: everest.io/share-export-location: {your_location} # 极速文件存储的共享路径 everest.io/enterprise-project-id: {your_project_id} # 极速文件存储的项目ID storage.kubernetes.io/csiProvisionerIdentity: everest-csi-provisioner persistentVolumeReclaimPolicy: Retain # 回收策略 storageClassName: csi-sfsturbo # SFS Turbo存储类名称 mountOptions: # 挂载参数 - vers=3 - nolock - timeo=600 - hard PV创建后，可以创建PVC关联PV，然后在工作负载的容器中挂载，具体操作步骤请参见通过静态存储卷使用已有极速文件存储。 验证挂载参数是否生效。 本例中将PVC挂载至使用nginx:latest镜像的工作负载，并通过mount -l命令查看挂载参数是否生效。 查看已挂载文件存储的Pod，本文中的示例工作负载名称为web-sfsturbo。 kubectl get pod | grep web-sfsturbo 回显如下： web-sfsturbo-*** 1/1 Running 0 23m 执行以下命令查看挂载参数，其中web-sfsturbo-***为示例Pod。 kubectl exec -it web-sfsturbo-*** -- mount -l | grep nfs 若回显中的挂载信息与设置的挂载参数一致，说明挂载参数设置成功。 {您的挂载地址} on /data type nfs (rw,relatime,vers=3,rsize=1048576,wsize=1048576,namlen=255,hard,nolock,noresvport,proto=tcp,timeo=600,retrans=2,sec=sys,mountaddr=**.**.**.**,mountvers=3,mountport=20048,mountproto=tcp,local_lock=all,addr=**.**.**.**)

极速文件存储挂载参数 CCE Autopilot在挂载极速文件存储时默认设置了如表1所示的参数。 表1 极速文件存储挂载参数 参数 参数值 描述 keep-original-ownership 无需填写 表示是否保留文件挂载点的ownership。 默认为不添加该参数，此时挂载极速文件存储时将会默认把挂载点的ownership修改为root:root。 如添加该参数，挂载极速文件存储时将保持文件系统原有的ownership。 vers 3 文件系统版本，目前只支持NFSv3。取值：3 nolock 无需填写 选择是否使用NLM协议在服务器上锁文件。当选择nolock选项时，锁对于同一主机的应用有效，对不同主机不受锁的影响。 timeo 600 NFS客户端重传请求前的等待时间(单位为0.1秒)。建议值：600。 hard/soft 无需填写 挂载方式类型。 取值为hard，即使用硬连接方式，若NFS请求超时，则客户端一直重新请求直至成功。 取值为soft，即软挂载方式挂载系统，若NFS请求超时，则客户端向调用程序返回错误。 默认为hard。 除了以上参数外，您还可以设置其他的文件存储挂载参数，具体请参见挂载NFS文件系统到云服务器（Linux）。

使用场景 根据使用场景不同，对象存储支持以下挂载方式： 通过静态存储卷使用已有对象存储：即静态创建的方式，需要先使用已有的对象存储创建PV，然后通过PVC在工作负载中挂载存储。适用于已有可用的底层存储或底层存储需要包周期的场景。 通过动态存储卷使用对象存储：即动态创建的方式，无需预先创建对象存储，在创建PVC时通过指定存储类（StorageClass），即可自动创建对象存储和对应的PV对象。适用于无可用的底层存储，需要新创建的场景。

对象存储规格 对象存储提供了多种存储类别，从而满足客户业务对存储性能、成本的不同诉求。 并行文件系统（推荐使用）：并行文件系统（Parallel File System）是OBS提供的一种经过优化的高性能文件系统，提供毫秒级别访问时延，以及TB/s级别带宽和百万级别的IOPS，能够快速处理高性能计算（HPC）工作负载。相较于对象桶，并行文件系统在稳定性、性能上更具优势。详细介绍请参见什么是并行文件系统。 对象桶（不推荐使用）： 标准存储：访问时延低和吞吐量高，因而适用于有大量热点文件（平均一个月多次）或小文件（小于1MB），且需要频繁访问数据的业务场景，例如：大数据、移动应用、热点视频、社交图片等场景。 低频访问存储：适用于不频繁访问（平均一年少于12次）但在需要时也要求快速访问数据的业务场景，例如：文件同步/共享、企业备份等场景。与标准存储相比，低频访问存储有相同的数据持久性、吞吐量以及访问时延，且成本较低，但是可用性略低于标准存储。 关于对象存储的详细介绍，请以对象存储类别为准。

对象存储介绍 对象存储服务 （Object Storage Service，OBS）提供海量、安全、高可靠、低成本的数据存储能力，可供用户存储任意类型和大小的数据。适合企业备份/归档、 视频点播 、视频监控等多种数据存储场景。 标准接口：具备标准Http Restful API接口，用户必须通过编程或第三方工具访问对象存储。 数据共享：服务器、嵌入式设备、IOT设备等所有调用相同路径，均可访问共享的对象存储数据。 公共/私有网络：对象存储数据允许在公网访问，满足互联网应用需求。 容量与性能：容量无限制，性能较高（IO读写时延10ms级）。 应用场景：适用于（基于OBS界面、OBS工具、OBS SDK等）的一次上传共享多读（ReadOnlyMany）的各种工作负载（Deployment/StatefulSet）和普通任务（Job）使用，主要面向大数据分析、静态网站托管、在线视频点播、基因测序、智能视频监控、备份归档、企业云盘（网盘）等场景。