云服务器内容精选
-
操作场景 在实际作业运行中，由于作业的数据流量变化，导致所需计算资源不同，造成流量较小时计算资源浪费，流量较大时计算资源不足以满足计算所需。 DLI 提供的动态扩缩容功能可以根据当前作业的负载情况，例如：数据输入输出量、数据输入输出速率、反压等情况，动态的调整当前作业所用的计算资源，提升资源利用率。 开启Flink作业动态扩缩容后，系统将根据Flink作业的实际资源需求动态调整资源分配。当弹性资源池中剩余的Pod资源足以支持作业的最小资源需求时，系统将自动减少作业所在节点的数量，确保作业高效运行，同时提高资源的利用效率。 当前仅Flink 1.12版本的作业支持开启动态扩缩容。
-
使用须知 在Flink作业进行动态扩缩容时如果队列资源被抢占，剩余资源不满足作业启动所需资源则可能存在作业无法正常恢复的情况。 在Flink作业进行动态扩缩时后台作业需要停止继而从savepoint恢复，因此未恢复成功前，作业无法处理数据。 因扩缩容过程中需要触发savepoint，因此必须配置obs桶，并保存日志，同时请注意开启checkpoint。 扩缩容检测周期不要设置过小，避免频繁启停作业。 扩缩容作业恢复过程中的时间长短受savepoint的大小影响，如果保存点较大，可能恢复时间较慢。 如果需要调整动态扩缩容的配置项，则需要停止作业进行编辑，并提交运行才能生效。
-
Flink作业权限操作场景 针对不同用户，可以通过权限设置分配不同的作业，不同用户之间的作业效率互不影响，保障作业性能。 管理员用户和作业的所有者拥有所有权限，不需要进行权限设置且其他用户无法修改其作业权限。 给新用户设置作业权限时，该用户所在用户组的所属区域需具有Tenant Guest权限。关于Tenant Guest权限的介绍和开通方法，详细参见《权限策略》和《 统一身份认证 服务用户指南》中的创建用户组。
-
Flink作业权限使用说明 查看作业详情 租户以及admin用户可以查看和操作所有作业。 子用户以及拥有只读权限的用户只能查看自己的作业。 他人赋权给该子用户查看权限外的任意权限，则该作业仅显示在作业列表中，但不支持该子用户查看作业详情。 启动作业 用户需要同时拥有队列的提交作业权限以及作业的启动作业权限。 停止作业 用户需要同时拥有队列的停止作业权限以及作业的停止作业权限。 删除作业 如果作业在可删除状态，则用户拥有作业的删除权限即可。 如果作业在不可删除状态，用户删除作业时，系统会先停止作业，停止作业权限说明可以参考•停止作业，并且用户还需要拥有作业的删除权限。 创建作业 子用户默认不能创建作业。 创建作业时，用户需要拥有创建作业的权限。目前只有admin用户创建作业的权限，同时用户还需要拥有该作业使用的相关程序包组权限或者程序包权限。 编辑作业 编辑作业时，用户需要拥有更新作业的权限，同时用户还需要拥有该作业使用的相关程序包所属组权限或者程序包权限。
-
概述 dws-connector-flink是在dws-client的基础上对接Flink的一个工具，工具为对dws-client的包装，整体入库能力跟dws-client一致。dws-connector-flink为 GaussDB (DWS)团队自研工具，后续将根据GaussDB(DWS)数据库持续优化。 dws-flink-connector的DWS-Connector只支持单并发查询存量数据，暂不支持并行读取。
-
UDF函数DnHashFunction参数说明 参数格式 dn_hash（'dws表名',sink并行度,最大并行度,dws作为分布列的数据在源数据的字段名称{1,}） 参数说明 使用时上游并行度必须不多于sink并行度，DnHashFunction同样是通过进程内获取sink 算子初始化的dws client实例获取到的表元数据，如果当前进程无sin算子就会导致无法获取client实例。 使用后会增加一个hash算子，如果链路有多个算子处理业务，当执行hash算子后不可以再有改变数据分区的算子，否则数据会被再次分区就不能到达指定sink算子。 最大并行度默认flink自动调整的，算法中需要使用，因此自动调整的无法使用，必须通过参数设置固定并把设置额值作为UDF的参数，可以通过参数pipeline.max-parallelism设置或者jar方式通过API设置： StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();evn.setParallelism(1);evn.setMaxParallelism(1024); 如果分布列包含多个字段，分布列的字段顺序需要保持和DWS一致，分布列支持的字段类型和dws client一致参考参数WRITE_PARTITION_POLICY，使用功能同样需要额外配置，不可自行使用。
-
使用flink SQL直连DN入库 该能力依赖flink sql DISTRIBUTEBY能力，mrs有提供此能力，具体请参见Flink SQL语法增强。 connector提供udf函数可根据分布列值计算出下游并并发结合flink sql DISTRIBUTEBY能力实现将数据按DN分区能力，示例： 需要在SQL中引入UDF。 CREATE temporary FUNCTION dn_hash AS 'com.huaweicloud.dws.connectors.flink.partition.DnHashFunction'; 正常写Source SQL。 CREATE TABLE users( id BIGINT, name STRING, age INT, text STRING, created_at TIMESTAMP(3), updated_at TIMESTAMP(3)) WITH ( 'connector' = 'datagen', 'fields.id.kind' = 'sequence', 'fields.id.start' = '1', 'fields.id.end' = '1000', 'fields.name.length' = '10', 'fields.age.min' = '18', 'fields.age.max' = '60', 'fields.text.length' = '5' ) Sink表定义SQL中需要新增一个字段并且要求int类型值用于接收UDF计算的结果，示例中叫dn_hash。 create table dws_users( dn_hash int, id BIGINT, name STRING, age INT, text STRING, created_at TIMESTAMP(3), updated_at TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED) WITH ( 'connector' = 'dws', 'url' = '%s', 'tableName' = 'test.users', 'username' = '%s', 'autoFlushBatchSize' = '50000', 'password' = '%s' ) Insert into sql使用 udf获取数据下游算子信息，同时使用DISTRIBUTEBY对返回结果做数据分区，数据就会按照udf返回信息到下游指定并行度。 insert into dws_users select /*+ DISTRIBUTEBY('dn_hash') */ dn_hash('test.users',10,1024, id) as dn_hash, * from users
-
示例 该示例是从kafka数据源中读取数据，写入DWS结果表中，并指定攒批时间不超过10秒，每批数据最大30000条，其具体步骤如下： 在GaussDB(DWS)数据库中创建表public.dws_order： 1 2 3 4 5 6 7 8 91011 create table public.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 ); 消费Kafka中order_test topic中的数据作为数据源，public.dws_order作为结果表，Kafka数据为JSON格式，并且字段名称和数据库字段名称一一对应： 1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637383940 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', 'topic' = 'order_test', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort', 'properties.group.id' = 'GroupId', 'scan.startup.mode' = 'latest-offset', 'format' = 'json');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) WITH ( 'connector' = 'dws', 'url' = 'jdbc:gaussdb://DWSAddress:DWSPort/DWSdbName', 'tableName' = 'dws_order', 'username' = 'DWSUserName', 'password' = 'DWSPassword', 'autoFlushMaxInterval' = '10s', 'autoFlushBatchSize' = '30000');insert into dwsSink select * from kafkaSource; 给Kafka写入测试数据： 1 {"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"} 等10秒后在GaussDB(DWS)表中查询结果： 1 select * from dws_order 结果如下：
-
常见问题 Q：writeMode参数设置什么值比较合适？ A：根据业务场景分update（只更新存在的数据）和upsert（对于同一主键数据如果存在就更新，不存在就新增一条数据）两个类型，推荐直接使用auto方式即可，该方式下会根据数据量的大小自动选择，如果数据量较大会增大攒批参数autoFlushBatchSize，即可提升入库性能。 Q：autoFlushBatchSize和autoFlushMaxInterval怎么设置比较合适？ A：autoFlushBatchSize参数用于设置最大攒批条数，autoFlushMaxInterval参数用于设置最大攒批间隔，两个参数分别从时间和空间维度管控攒批。 通过autoFlushMaxInterval可保证数据量较小时的时效性，如对时效性无强制要求通常不建议设置的太小，建议不低于3s走默认值即可。 通过autoFlushBatchSize可控制一批数据的最大条数，一般来说攒批量越大，对于整体入库性能会更好，对性能来说通常该参数的设置推荐越大越好，参数的设置根据业务数据的大小以及flink运行内存来设置，保证不内存溢出。 对于大多业务来说无需设置autoFlushMaxInterval，将autoFlushBatchSize设置为50000即可。 Q: 遇到数据库死锁了怎么办？ A：通常出现死锁大致分为行锁死锁和分布式死锁。 行锁：该场景通常为同一主键数据的并发更新造成行锁，该情况可以通过对数据做key by解决，key by必须根据数据库主键做，保证同一个主键数据会在同一个并发中，破坏掉并发更新的条件，无法造成死锁。Flink SQL做key by需要Flink本身支持，对于DLI/ MRS 均能实现，如MRS flink通过增加参数“key-by-before-sink=true”可实现key by。具体怎么使用以实现方为准，对于无法使用的建议使用API方式入库。 分布式死锁：该场景通常为列存表的并发更新造成分布式死锁，暂无法解决，建议使用行存或者hstore。
-
Flink SQL配置参数 Flink SQL中设置的PRIMARY KEY将自动映射到dws-client中的uniqueKeys。参数跟随client版本发布，参数功能与client一致，以下参数说明表示为最新参数。 表1 数据库配置 参数 说明 默认值 connector flink框架区分connector参数，固定为dws。 - url 数据库连接地址。 - username 配置连接用户。 - password 配置密码。 - tableName 对应dws表。 - 表2 连接配置 参数 说明 默认值 connectionSize 初始dws-client时的并发数量。 1 connectionMaxUseTimeSeconds 连接创建多少秒后强制释放（单位：秒）。 3600（一小时） connectionMaxIdleMs 连接最大空闲时间，超过后将释放（单位：毫秒）。 60000（一分钟） dws client参数全量支持在flink sql通过key方式配置，下表参数为兼容1.x版本参数，当同时配置2.x和1.x参数时生效2.x版本参数值： 表3 DWS client写入参数 参数 说明 默认值 conflictStrategy 有主键表数据写入时主键冲突策略： ignore：保持原数据，忽略更新数据。 update：用新数据中非主键列更新原数据中对应列。 replace：用新数据替换原数据。 说明： update和replace在全字段upsert时等效，在部分字段upsert时，replace相当于将数据中不包含的列设置为null。 update writeMode 入库方式： auto：系统自动选择。 copy_merge：当存在主键时使用copy方式入临时表，从临时表merge至目标表；无主键时直接copy至目标表。 copy_upsert：当存在主键时使用copy方式入临时表，从临时表upsert至目标表；无主键时直接copy至目标表。 upsert: 有主键用upsert sql入库；无主键用insert into入库。 UPDATE：使用update where语法更新数据，若原表无主键可选择指定uniqueKeys，指定字段不要求必须是唯一索引，但非唯一索引可能会影响性能。 COPY_UPDATE：数据先通过copy方式入库到临时表，通过临时表加速使用update from where方式更新目标数据。 UPDATE_AUTO：批量小于copyWriteBatchSize使用UPDATE，否则使用COPY_UPDATE。 auto maxFlushRetryTimes 在入库时最大尝试次数，次数内执行成功则不抛出异常，每次重试间隔为 1秒 * 次数。 3 autoFlushBatchSize 自动刷库的批大小（攒批大小）。 5000 autoFlushMaxInterval 自动刷库的最大间隔时间（攒批时长）。 5s copyWriteBatchSize 在“writeMode == auto”下，使用copy的批大小。 5000 metadataCacheSeconds 系统中对元数据的最大缓存时间，例如表定义信息（单位秒）。 180 copyMode copy入库格式： CS V：将数据拼接为CSV格式入库，该方式稳定，但性能略低。 DELIMITER：用分隔符将数据拼接，然后入库，该方式需要数据中不包含分隔符。 CSV createTempTableMode 创建临时表方式： AS LIKE AS numberAsEpochMsForDatetime 如果数据库为时间类型数据源为数字类型，是否将数据当成时间戳转换为对应时间类型。 false stringToDatetimeFormat 如果数据库为时间类型数据源为字符串类型，按该格式转换为时间类型，该参数配置即开启。 null 表4 connector参数 参数 说明 默认值 ignoreDelete 忽略flink任务中的delete。 false (1.0.10前默认true) ignoreNullWhenUpdate 是否忽略flink中字段值为null的更新，只有在“conflictStrategy == update”时有效。 false sink.parallelism flink系统参数用于设置sink并发数量。 跟随上游算子 printDataPk 是否在connector接收到数据时打印数据主键，用于排查问题。 false ignoreUpdateBefore 忽略flink任务中的update_before，在大表局部更新时该参数一定打开，否则有update时会导致数据的其它列被设置为null，因为会先删除再写入数据。 true
-
格式语法 SQL语法格式可能在不同Flink环境下有细微差异，具体以事件环境格式为准，with后面的参数名称及参数值以此文档为准。 1 2 3 4 5 6 7 8 9101112 create table dwsSink ( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED))with ( 'connector' = 'dws', 'url' = '', 'tableName' = '', 'username' = '', 'password' = '');
-
FlinkSQL支持MultiJoin算子使用限制 MultiJoin算子只支持FULL OUTER JOIN和INNER JOIN语法。 MultiJoin算子只支持等值关联。如果有非等值过滤条件，可以使用view提前过滤。 MultiJoin算子多表关联时所有的表关联键要一致。 MultiJoin算子仅支持table.exec.state.ttl，不支持表级TTL以及JTL。 Full outer join到MultiJoin状态不兼容，切换时无法通过快照恢复作业。
-
Flink HA方案介绍 每个Flink集群只有单个JobManager，存在单点失败的情况。Flink有Yarn、Standalone和Local三种模式，其中Yarn和Standalone是集群模式，Local是指单机模式。但Flink对于Yarn模式和Standalone模式提供HA机制，使集群能够从失败中恢复。这里主要介绍Yarn模式下的HA方案。 Flink支持HA模式和Job的异常恢复。这两项功能高度依赖ZooKeeper，在使用之前用户需要在“flink-conf.yaml”配置文件中配置ZooKeeper，配置ZooKeeper的参数如下： high-availability: zookeeperhigh-availability.zookeeper.quorum: ZooKeeperIP地址:2181high-availability.storageDir: hdfs:///flink/recovery Yarn模式 Flink的JobManager与Yarn的Application Master（简称AM）是在同一个进程下。Yarn的ResourceManager对AM有监控，当AM异常时，Yarn会将AM重新启动，启动后，所有JobManager的元数据从HDFS恢复。但恢复期间，旧的业务不能运行，新的业务不能提交。ZooKeeper上还是存有JobManager的元数据，比如运行Job的信息，会提供给新的JobManager使用。对于TaskManager的失败，由JobManager上Akka的DeathWatch机制处理。当TaskManager失败后，重新向Yarn申请容器，创建TaskManager。 Yarn模式的HA方案的更多信息，可参考链接： http://hadoop.apache.org/docs/r3.1.1/hadoop-yarn/hadoop-yarn-site/ResourceManagerHA.html Standalone模式 对于Standalone模式的集群，可以启动多个JobManager，然后通过ZooKeeper选举出Leader作为实际使用的JobManager。该模式下可以配置一个主JobManager（Leader JobManager）和多个备JobManager（Standby JobManager），这能够保证当主JobManager失败后，备的某个JobManager可以承担主的职责。图1为主备JobManager的恢复过程。 图1 恢复过程 TaskManager恢复 对于TaskManager的失败，由JobManager上Akka的DeathWatch机制处理。当TaskManager失败后，由JobManager负责创建一个新TaskManager，并把业务迁移到新的TaskManager上。 JobManager恢复 Flink的JobManager与Yarn的Application Master（简称AM）是在同一个进程下。Yarn的ResourceManager对AM有监控，当AM异常时，Yarn会将AM重新启动，启动后，所有JobManager的元数据从HDFS恢复。但恢复期间，旧的业务不能运行，新的业务不能提交。 Job恢复 Job的恢复必须在Flink的配置文件中配置重启策略。当前包含三种重启策略：fixed-delay、failure-rate和none。只有配置fixed-delay、failure-rate，Job才可以恢复。另外，如果配置了重启策略为none，但Job设置了Checkpoint，默认会将重启策略改为fixed-delay，且重试次数是配置项“restart-strategy.fixed-delay.attempts”的值。 三种策略的具体信息请参考Flink官网：https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/task_failure_recovery.html。配置策略的参考如下： restart-strategy: fixed-delayrestart-strategy.fixed-delay.attempts: 3restart-strategy.fixed-delay.delay: 10 s 以下场景的异常，都会导致Job重新恢复： 当JobManager失败后，所有Job会停止，直到新的JobManager运行后，所有Job恢复。 当某一TaskManager失败后，这个TaskManager上的所有作业都将停止，然后等待有可用资源后重启。 当某个Job的Task失败后，整个Job也会重启。 有关Job的配置重启策略，具体内容请参见https://ci.apache.org/projects/flink/flink-docs-release-1.12/ops/jobmanager_high_availability.html。
-
中小规模数据量维度表可以采用内存维度表（如Hudi） 内存维度表：将维度数据加载到内存当中，每个TM都会加载全量的数据，在内存内实现数据点查关联。若数据量过大，需要给TM分配大的内存空间，否则容易导致作业异常。 外置维度表：将维度数据存在高速的K-V数据库中，通过远程的K-V查询实现点查关联，常用的开源K-V库有HBase。 状态维度表：将维度表数据当做流表，实时读入到流式作业当中，通过数据的回撤流能力实现维度更新和数据不对齐场景下的数据一致性保证。维度表保存时间比较长，当前Flink on Hudi能力可以针对Hudi作为维度表单独设置TTL时长。 表2 维度表实现方式对比 维度 内存维度表（hive/hudi表） 外置维度表（HBase） 状态维度表 性能 非常高（毫秒内） 中（毫秒级） 高（毫秒内~毫秒级） 数据量 小，建议单个TM保持1GB以内 大，TB级 中，GB级 存储资源 内存消耗大，单个TM全量存储 外置存储，无存储资源消耗 各TM分散存储，内存+磁盘存储 时效性 周期性数据加载，时效低 相对高 高 关联数据结果 低 中 -
-
生产环境使用增量Rocksdb作为State Backend Flink提供了三种状态后端：MemoryStateBackend，FsStateBackend，和RocksDBStateBackend。 MemoryStateBackend是将state存储在JobManager的Java堆上，每个状态的大小不能超过akka帧的大小，且总量不能超过JobManager的堆内存大小。所以只适合于本地开发调试，或状态大小有限的一些小状态的场景。 FsStateBackend是文件系统状态后端，正常情况下将state存储在TaskManager堆内存中，当Checkpoint时将state存储在文件系统上，而JobManager内存中存储极少的元数据（高可用场景下存储在ZooKeeper）。因为文件系统的存储空间足够，适合于大状态，长窗口，或大键值状态的有状态处理任务，也适合于高可用方案。 RocksDBStateBackend是内嵌数据库后端，正常情况下state存储在RocksDB数据库中，该数据库数据放在本地磁盘上，在Checkpoint时将state存储在配置的文件系统上而JobManager内存中存储极少的元数据（高可用场景下存储在ZooKeeper），同时是唯一一个可以增量Checkpoint的状态后端，除了适合于FsStateBackend的场景，还适用于超大状态的场景。 表1 Flink状态后端 类别 MemoryStateBackend FsStateBackend RocksDBStateBackend 方式 Checkpoint数据直接返回给Master节点，不落盘 数据写入文件，将文件路径传给Master 数据写入文件，将文件路径传给Master 存储 堆内存 堆内存 Rocksdb（本地磁盘） 性能 相比最好（一般不用） 性能好 性能不好 缺点 数据量小、易丢失 容易OOM风险 需要读写、序列化、IO等耗时 是否支持增量 不支持 不支持 支持 【示例】配置RockDBStateBackend（flink-conf.yaml）： state.backend: rocksdbstate.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints
更多精彩内容
CDN加速
GaussDB
文字转换成语音
免费的服务器
如何创建网站
域名网站购买
私有云桌面
云主机哪个好
域名怎么备案
手机云电脑
SSL证书申请
云点播服务器
免费OCR是什么
电脑云桌面
域名备案怎么弄
语音转文字
文字图片识别
云桌面是什么
网址安全检测
网站建设搭建
国外CDN加速
SSL免费证书申请
短信批量发送
图片OCR识别
云数据库MySQL
个人域名购买
录音转文字
扫描图片识别文字
OCR图片识别
行驶证识别
虚拟电话号码
电话呼叫中心软件
怎么制作一个网站
Email注册网站
华为VNC
图像文字识别
企业网站制作
个人网站搭建
华为云计算
免费租用云托管
云桌面云服务器
ocr文字识别免费版
HTTPS证书申请
图片文字识别转换
国外域名注册商
使用免费虚拟主机
云电脑主机多少钱
鲲鹏云手机
短信验证码平台
OCR图片文字识别
SSL证书是什么
申请企业邮箱步骤
免费的企业用邮箱
云免流搭建教程
域名价格
