[SQL] NumberFormatException in Flink

I am running some aggregations in Flink SQL, Flink 1.18, and i am getting this error:

Caused by: java.lang.NumberFormatException: Character N is neither a decimal digit number, decimal point, nor "e" notation exponential mark.
at java.base/java.math.BigDecimal.<init>(Unknown Source)
at java.base/java.math.BigDecimal.<init>(Unknown Source)
at java.base/java.math.BigDecimal.<init>(Unknown Source)
at java.base/java.math.BigDecimal.valueOf(Unknown Source)
at org.apache.flink.table.runtime.functions.SqlFunctionUtils.sround(SqlFunctionUtils.java:896)
at StreamExecCalc$1286.processElement_0_3_rewriteGroup5_9_rewriteGroup78_split251(Unknown Source)
at StreamExecCalc$1286.processElement_0_3_rewriteGroup5_9_rewriteGroup78(Unknown Source)
at StreamExecCalc$1286.processElement_0_3_rewriteGroup80(Unknown Source)
at StreamExecCalc$1286.processElement_split240(Unknown Source)
at StreamExecCalc$1286.processElement(Unknown Source)


This seems to be due to some invalid values that exist in a MySQL table that i read and use in the aggregations to update them. How can these values be data cleaned? I have tried with REGEXP but i get a message that it is not supported in Flink although in the documentation it says it does. My current implementation of data cleaning does this:

CASE WHEN TRY_CAST(col1 AS BIGINT) IS NOT NULL THEN col1 ELSE NULL END AS col1


For all the columns that are involved in ROUND functions, but it does not seem to fix the issue.

Continue reading...