watermark报类型不匹配，但是已经转成timestamp了

[xiazhifu]

定义watermark时报类型不匹配，但是已经通过TO_TIMESTAMP函数转成timestamp类型了，系统提示出来的却是localDateTime类型，求大佬解答，如下图

[simenliuxing]

生成watermark的字段目前只支持long和timestramp类型，TO_TIMESTAMP返回的不是这两种类型。你可以按如下方式来定义：
1.ctime是long或者timestramp类型，不需要TO_TIMESTAMP转换了，直接定义watermark就行。
2.自定义一个udf函数，如下：
public class TimeParse extends ScalarFunction {
public static Map<String, SimpleDateFormat> dateTimeFormatterMap = new ConcurrentHashMap<String, SimpleDateFormat>();

/**
 * 统一返回该格式 long格式的timestramp
 *
 * @param timStr
 * @param format
 * @return
 */
public Long eval(String timStr, String format) {
    SimpleDateFormat sdf;
    long parseTime = 0L;
    try {
        if (dateTimeFormatterMap.containsKey(format)) {
            sdf = dateTimeFormatterMap.get(format);
        } else {
            sdf = new SimpleDateFormat(format);
            dateTimeFormatterMap.put(format, sdf);
        }
        parseTime = sdf.parse(timStr).getTime();
    } catch (ParseException e) {
        e.printStackTrace();
        // TODO log
    }
    return parseTime;
}

}
--sql中使用方式
CREATE scala FUNCTION totimestramp WITH TimeParse;
...
totimestramp(ctime,'yyyyMMddHHmmss') as ts,
WATERMARK FOR ts AS withOffset(ts, 0)