扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
flink的时间及时区问题怎样解决，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。
武清网站制作公司哪家好，找创新互联！从网页设计、网站建设、微信开发、APP开发、自适应网站建设等网站项目制作，到程序开发，运营维护。创新互联2013年开创至今到现在10年的时间，我们拥有了丰富的建站经验和运维经验，来保证我们的工作的顺利进行。专注于网站建设就选创新互联。
1.时间纪元
所谓的”时间纪元”就是1970年1月1日0时0分0秒，指的是开始的时间。比如Java类代码：
Date date = new Date(0);
System.out.println(date);
打印出来的结果：
Thu Jan 01 08:00:00 CST 1970
也是1970年1月1日，实际上时分秒是0点0分0秒，这里打印出来的时间是8点而非0点，原因是存在系统时间和本地时间的问题，其实系统时间依然是0点，只不过我们的电脑时区设置为东8区，故打印的结果是8点。
只需要将时区设置为GMT+0，即可打印出0点0分0秒
System.setProperty("user.timezone","GMT+0");
实际上时区问题都是在此时间纪元基础上加/减一定的offset。
2.Flink时间
说java纪元跟本文将的flink时间问题有啥关系呢？
Flink在使用时间的这个概念的时候就是基于时间纪元这个概念的。比如首先，我们的时区是东八区，在我们的视野中UTC-0时间应该加8小时的offset，才是我们看到的时间，所以在使用flink的窗口的时候往往比我们当前的时间少8小时。
还有flink的窗口对其，也是基于纪元时间的。比如下面的有三个窗口函数的例子
1).5min滚动窗口
14:16:391启动的窗口，滚动窗口时间是5min，会发现并不是等待五分钟之后才有结果输出，而是到了14:20:00.0的时候就直接输出结果了。
2).30min滚动窗口
14:27:11启动的滚动窗口，是在14:30:00的时候就直接输出了，而不是等待半小时。
3).1hour滚动窗口
15:54:48启动的一小时的滚动窗口，输出时间是16点整。
时间上差了八小时，但是对齐是基于时间纪元的整数单位。
3.解决差八小时问题
实际在使用的时候flink输出的时差很令人反感，但是没办法flink目前不支持配置时区，但是blink支持，等待着合并吧。
其实，时区问题解决方案比较多吧，要想不伤筋动骨，主要介绍以下三种：
flink端不做处理。也即是在读取数据的时候加上8小时的offset。
使用udf等算子给时间戳加上8小时的offset。
sink内部做处理。
1).Udf实现
sink端处理
import org.apache.flink.table.functions.ScalarFunction;import java.sql.Timestamp;import java.text.ParseException;import java.text.SimpleDateFormat;import java.util.Date;import java.util.TimeZone;public class UTC2Local extends ScalarFunction {public Timestamp eval(Timestamp s) {long timestamp = s.getTime() + 28800000;return new Timestamp(timestamp);}}
注册udf
tEnv.registerFunction("utc2local",new UTC2Local()); 使用udf
Table table1 = tEnv.sqlQuery("select count(number),utc2local(TUMBLE_END(proctime, INTERVAL '1' HOUR)) from res group by TUMBLE(proctime, INTERVAL '1' HOUR)");
2). sink内部支持
sink端的实现也比较简单，主要是判断输出字段类型，然后加上8小时offset即可。可以参考blink的printtablesink的实现。
override def invoke(in: JTuple2[JBool, Row]): Unit = {val sb = new StringBuilderval row = in.f1for (i <- 0 to row.getArity - 1) {if (i > 0) sb.append(",")val f = row.getField(i)if (f.isInstanceOf[Date]) {sb.append(DateTimeFunctions.dateFormat(f.asInstanceOf[JDate].getTime, "yyyy-MM-dd", tz))} else if (f.isInstanceOf[Time]) {sb.append(DateTimeFunctions.dateFormat(f.asInstanceOf[JDate].getTime, "HH:mm:ss", tz))} else if (f.isInstanceOf[Timestamp]) {sb.append(DateTimeFunctions.dateFormat(f.asInstanceOf[JDate].getTime,"yyyy-MM-dd HH:mm:ss.SSS", tz))} else {sb.append(StringUtils.arrayAwareToString(f))}}if (in.f0) {System.out.println(prefix + "(+)" + sb.toString())} else {System.out.println(prefix + "(-)" + sb.toString())}}
看完上述内容，你们掌握flink的时间及时区问题怎样解决的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注创新互联行业资讯频道，感谢各位的阅读！
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
微信二维码
移动版官网