Flink 1.10 UDF 的一個小問題

在使用 Flink 1.10 的 SQL 的時候，遇到個小問題： 一個返回當前時間的函數返回的結果是啟動的時間，並且保持不變。

比如下面這個UDF，獲取當前時間的 時分秒（HH:mm:ss 格式）

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.table.functions.ScalarFunction;

import java.text.SimpleDateFormat;

/**
 * Sysdate 返回當前時間的 HH:mm:ss 格式的字符串
 */
public class Systime extends ScalarFunction {

    private static SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");

    public Systime() {
    }

    public String eval() throws Exception {
        // 當前時間的毫秒值，轉為 HH:mm:ss 
        return sdf.format(System.currentTimeMillis());
    }


    public TypeInformation<?> getResultType(Class<?>[] signature) {
        return Types.STRING;
    }
}

最開始用的時候，直接寫了這么個UDF，還自我感覺很溜。。。

在測試的時候，也沒有留意到返回的值都是相同的，在線上跑了幾天才發現。。。

如sql ：

INSERT INTO user_log_sink
SELECT user_id, systime(), item_id
FROM user_log a;

返回值如下：

全是一樣的

在代碼里面添加 日志信息，只有在啟動的時候輸出了一次日志信息（在客戶端），在 taskmanager.log 里沒有對應日志，說明確實沒有執行

對比其他的UDF 除了沒有參數，其他好像沒有什么不一樣，都是繼承的 ScalarFunction 、也都在客戶端注冊。。

在其他UDF 中也添加日志信息，測試發現：     

都是在注冊的時候，調用了對應的構造方法，但是無參的 UDF，在任務初始化階段，還調用了一次，而需要參數的UDF 並沒有；數據進入的時候，無參的UDF 並沒有調用 eval 方法，有參的UDF 就正常調用了。

直接debug 無參的UDF，從日志中發現除了在注冊的時候 執行了 構造方法，還調用了有一次  eval 方法，所以在 eval 中添加斷點：

 

 於是就找到這個類：

package org.apache.flink.table.planner.codegen
/**
  * Evaluates constant expressions with code generator.   計算常量表達式
  *
  * @param allowChangeNullability If the reduced expr's nullability can be changed, e.g. a null
  *                               literal is definitely nullable and the other literals are
  *                               not null.
  */
class ExpressionReducer(
    config: TableConfig,
    allowChangeNullability: Boolean = false)
  extends RexExecutor

從注釋就可以看出，Flink 在計算常量表達式，而我的 systime() 就被認為是常量表達式了，在客戶端執行一次，得到結果，之后的函數，直接使用對應常量，而不再進函數計算返回了。

 

 隨后的獲取到的SQL輸出結果：

 看到這里，問題就很清楚了，一些常量的表達式，Flink 在客戶端初始化的時候，直接執行一次，緩存了結果，之后就直接返回這個結果，而不是去執行表達式

 解決也很簡單，直接給 UDF 添加一個參數（注：必須是SQL的字段，如果是常量也會被Flink 優化）

public String eval(String input) throws Exception {
        // 當前時間的毫秒值，轉為 HH:mm:ss
        return sdf.format(System.currentTimeMillis());
    }

使用的SQL：

INSERT INTO user_log_sink
SELECT user_id, systime(), systime_param(user_id)
FROM user_log a;

systime_param 並不會再初始化的時候執行

返回結果如下：

 

 搞定

歡迎關注Flink菜鳥公眾號，會不定期更新Flink（開發技術）相關的推文