云服务器内容精选

  • FlinkServer作业重启策略介绍 Flink支持不同的重启策略,以在发生故障时控制作业是否重启以及如何重启。如果不指定重启策略,集群会使用默认的重启策略。用户也可以在提交作业时指定一个重启策略,可参考如何创建FlinkServer作业在作业开发界面配置( MRS 3.1.0及以后版本)。 重启策略也可以通过Flink的配置文件“客户端安装目录/Flink/flink/conf/flink-conf.yaml”中的参数“restart-strategy”指定,为全局配置,还可以在应用代码中动态指定,会覆盖全局配置,重启策略包括失败率(failure-rate)和两种默认策略,默认策略为如下: 无重启(No restart):如果没有启用CheckPoint,默认使用该策略。 固定间隔(fixed-delay):如果启用了CheckPoint,但没有配置重启策略,默认使用该策略。
  • fixed-delay策略 发生故障时会尝试重启作业固定次数,如果超过了最大的尝试次数,作业最终会失败。并且在两次连续重启尝试之间,重启策略会等待固定的时间。 以配置如果重启失败了3次则认为该Job失败,重试时间间隔为10s为例,参数配置为: restart-strategy: fixed-delay restart-strategy.fixed-delay.attempts: 3 restart-strategy.fixed-delay.delay: 10 s
  • failure-rate策略 在作业失败后会直接重启,但超过设置的失败率后,作业会被认定为失败。在两个连续的重启尝试之间,重启策略会等待一个固定的时间。 以配置10分钟内如果重启失败了3次则认为该作业失败,重试时间间隔为10s为例,参数配置为: restart-strategy: failure-rate restart-strategy.failure-rate.max-failures-per-interval: 3 restart-strategy.failure-rate.failure-rate-interval: 10 min restart-strategy.failure-rate.delay: 10 s
  • 如何选择重启策略 如果用户在作业失败后,不希望重试,则推荐使用No restart策略。 如果用户在作业失败后,希望对作业进行重试,推荐使用failure-rate策略。因为fixed-delay策略可能会因为网络、内存等硬件故障导致用户作业失败次数达到最大重试次数,从而导致作业失败。 为了防止在failure-rate策略下的无限重启,推荐如下参数配置: restart-strategy: failure-rate restart-strategy.failure-rate.max-failures-per-interval: 3 restart-strategy.failure-rate.failure-rate-interval: 10 min restart-strategy.failure-rate.delay: 10 s
  • UDF java代码及SQL样例 UDF java使用样例 package com.xxx.udf; import org.apache.flink.table.functions.ScalarFunction; public class UdfClass_UDF extends ScalarFunction { public int eval(String s) { return s.length(); } } UDF SQL使用样例 CREATE TEMPORARY FUNCTION udf as 'com.xxx.udf.UdfClass_UDF'; CREATE TABLE udfSource (a VARCHAR) WITH ('connector' = 'datagen','rows-per-second'='1'); CREATE TABLE udfSink (a VARCHAR,b int) WITH ('connector' = 'print'); INSERT INTO udfSink SELECT a, udf(a) FROM udfSource;
  • 上传UDF至FlinkServer 准备UDF jar文件,大小不能超过200MB。 访问Flink WebUI,请参考访问FlinkServer WebUI界面。 单击“UDF管理”进入UDF管理页面。 单击“添加UDF”,在“本地Jar文件”参数后选择并上传本地已准备好的UDF jar文件。 填写UDF名称以及描述信息后,单击“确定”。 “UDF名称”最多可添加10项,“名称”可自定义,“类名”需与上传的UDF jar文件中UDF函数全限定类名一一对应。 上传UDF jar文件后,服务器默认保留5分钟,5分钟内单击确定则完成UDF创建,超时后单击确定则创建UDF失败并弹出错误提示:本地UDF文件路径有误。 在UDF列表中,可查看当前应用内所有的UDF信息。可在对应UDF信息的“操作”列编辑或删除UDF信息(只能删除未被使用的UDF项)。 (可选)如果需要立即运行或开发作业,可在“作业管理”进行相关作业配置,可参考如何创建FlinkServer作业。
  • 上传UDF 访问Flink WebUI,请参考访问FlinkServer WebUI界面。 单击“UDF管理”进入UDF管理页面。 单击“添加UDF”,在“本地Jar文件”参数后选择并上传本地已准备好的UDF jar文件。 填写UDF名称以及描述信息后,单击“确定”。 “UDF名称”最多可添加10项,“名称”可自定义,“类名”需与上传的UDF jar文件中UDF函数全限定类名一一对应。 上传UDF jar文件后,服务器默认保留5分钟,5分钟内单击确定则完成UDF创建,超时后单击确定则创建UDF失败并弹出错误提示:本地UDF文件路径有误。 在UDF列表中,可查看当前应用内所有的UDF信息。可在对应UDF信息的“操作”列编辑或删除UDF信息(只能删除未被使用的UDF项)。 (可选)如果需要立即运行或开发作业,可在“作业管理”进行相关作业配置,可参考创建FlinkServer作业。
  • UDF java代码及SQL样例 UDF java使用样例 package com.xxx.udf; import org.apache.flink.table.functions.ScalarFunction; public class UdfClass_UDF extends ScalarFunction { public int eval(String s) { return s.length(); } } UDF SQL使用样例 CREATE TEMPORARY FUNCTION udf as 'com.xxx.udf.UdfClass_UDF'; CREATE TABLE udfSource (a VARCHAR) WITH ('connector' = 'datagen','rows-per-second'='1'); CREATE TABLE udfSink (a VARCHAR,b int) WITH ('connector' = 'print'); INSERT INTO udfSink SELECT a, udf(a) FROM udfSource;