Flink SQL如何使用自定义格式解析TIMESTAMP？

Great question! While Flink SQL doesn’t let you directly pass a DateTimeFormatter instance into your queries, you absolutely can parse timestamp strings in non-standard formats using two main approaches: built-in functions and custom user-defined functions (UDFs).

1. Use Flink's Built-in TO_TIMESTAMP Function #

Flink SQL provides the TO_TIMESTAMP(string, format) function, which supports Java's DateTimeFormatter pattern syntax to parse custom timestamp strings. This is the simplest approach for most common custom formats.

Example Usage #

Suppose you have timestamp strings in the format yyyy/MM/dd HH:mm:ss (e.g., 2024/05/20 14:30:00). You can parse them like this:

SELECT TO_TIMESTAMP('2024/05/20 14:30:00', 'yyyy/MM/dd HH:mm:ss') AS parsed_timestamp;

Common Pattern Symbols #

Here are some frequently used pattern elements matching Java's DateTimeFormatter:

• yyyy: 4-digit year
• MM: 2-digit month (01-12)
• dd: 2-digit day (01-31)
• HH: 24-hour format hour (00-23)
• mm: 2-digit minute (00-59)
• ss: 2-digit second (00-59)
• SSS: 3-digit millisecond (000-999)

2. Create a Custom Scalar UDF for Special Formats #

If you need to handle extremely unique or complex timestamp formats that the built-in function can’t cover (e.g., yyyyMMddHHmmss without separators), you can create a custom scalar function that wraps DateTimeFormatter directly.

Step 1: Implement the UDF (Java Example) #

import org.apache.flink.table.functions.ScalarFunction;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

public class CustomTimestampParser extends ScalarFunction {
    // Define your custom formatter once (thread-safe)
    private static final DateTimeFormatter CUSTOM_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");

    // Evaluation method: takes a string and returns a LocalDateTime
    public LocalDateTime eval(String timestampStr) {
        return LocalDateTime.parse(timestampStr, CUSTOM_FORMATTER);
    }
}

Step 2: Register and Use the UDF in Flink SQL #

First, register the UDF in your Flink session:

CREATE TEMPORARY FUNCTION parse_custom_ts AS 'com.your.package.CustomTimestampParser';

Then use it in your query:

SELECT parse_custom_ts('20240520143000') AS parsed_timestamp;

Key Note #

Flink SQL’s default TIMESTAMP type parsing is fixed to the yyyy-MM-dd HH:mm:ss.fff format, but the above approaches let you bypass this restriction and handle any timestamp format you need—no direct DateTimeFormatter instance required in your SQL queries.

内容的提问来源于stack exchange，提问作者Rakesh Chalasani