其他
真实案例 | Flink实时计算处理脏数据问题
The following article is from 暴走大数据 Author 群主王知无
这是我最近在调试一个Flink任务中出现的问题。
问题描述
我们线上的一个任务今天报错，业务场景是：Flink消费消息队列中的消息然后做简单的维表联合查询。今天报警发现类似如下错误：任务fail-over重试几次，然后失败。
报错很明显，出现了NumberFormatException，null不能转为Long。
解决办法
解决办法更简单。
这个问题在Spark和Flink中都会存在，最直接的办法就是过滤掉。
阿里云上的Blink同样给出了文档，如下：
使用：
select avg(id) from SOURCEwhere test_id IS NOT NULL
select avg(case when id IS Nullthen 0else test_idend as test_id)from SOURCE
自定义一个UDF
按照上面的处理办法，在SQL中处理当然没有问题，但是我们在实际环境中会遇到非常多的这种情况，我个人建议自定义一个UDF，这个UDF的作用就是专门处理null或者空串或者其他各种异常情况的。
官方给出的一个经典的UDF案例如下：
public class HashCode extends ScalarFunction {
private int factor = 12;
public HashCode(int factor) {
this.factor = factor;
}
public int eval(String s) {
return s.hashCode() * factor;
}
}
BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);
// register the function
tableEnv.registerFunction("hashCode", new HashCode(10));
// use the function in Java Table API
myTable.select("string, string.hashCode(), hashCode(string)");
// use the function in SQL API
tableEnv.sqlQuery("SELECT string, hashCode(string) FROM MyTable");照着实现一个自己的处理逻辑即可！欢迎点赞+收藏