flink 函数

🧙 Questions

☄️ Ideas

CAST(data AS STRING)

IF(nullable_field = '', '', nullable_field)
IF(nullable_field is NULL, '123', nullable_field)

常用

DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyyMMddHHmmss')   -> etl_batch_id
CURRENT_TIMESTAMP                                  -> etl_load_time

salary 字段 RSA 非对称加密（Flink 1.12.7）

salary 是短敏感字段，可在写入下游前使用 RSA-OAEP with SHA-256 加密。函数只持有公钥，私钥仅由需要解密的一方保管；每次加密的结果都会不同，这是 OAEP 随机填充的正常行为。

下面的 UDF 输入为 STRING，输出为 Base64 文本。NULL 仍返回 NULL，且异常信息中不包含工资明文。

package com.isxcode.flink.udf;

import org.apache.flink.table.functions.ScalarFunction;

import javax.crypto.Cipher;
import javax.crypto.spec.OAEPParameterSpec;
import javax.crypto.spec.PSource;
import java.nio.charset.StandardCharsets;
import java.security.GeneralSecurityException;
import java.security.KeyFactory;
import java.security.PublicKey;
import java.security.interfaces.RSAPublicKey;
import java.security.spec.MGF1ParameterSpec;
import java.security.spec.X509EncodedKeySpec;
import java.util.Base64;

/**
 * 使用 RSA/OAEP(SHA-256) 加密 salary 等短文本字段。
 * 构造参数必须是 X.509 PEM 格式的 RSA 公钥。
 */
public final class SalaryRsaEncryptFunction extends ScalarFunction {

    private static final long serialVersionUID = 1L;
    private static final int SHA_256_LENGTH_BYTES = 32;

    private final String publicKeyPem;
    private transient volatile RSAPublicKey publicKey;

    public SalaryRsaEncryptFunction(String publicKeyPem) {
        if (publicKeyPem == null || publicKeyPem.trim().isEmpty()) {
            throw new IllegalArgumentException("RSA public key must not be blank");
        }
        this.publicKeyPem = publicKeyPem;
    }

    public String eval(String salary) {
        if (salary == null) {
            return null;
        }

        try {
            RSAPublicKey key = getPublicKey();
            byte[] plaintext = salary.getBytes(StandardCharsets.UTF_8);
            int maxPlaintextLength = (key.getModulus().bitLength() + 7) / 8
                    - 2 * SHA_256_LENGTH_BYTES - 2;
            if (plaintext.length > maxPlaintextLength) {
                throw new IllegalArgumentException("salary is too long for the configured RSA key");
            }

            Cipher cipher = Cipher.getInstance("RSA/ECB/OAEPWithSHA-256AndMGF1Padding");
            // 显式指定 MGF1 使用 SHA-256，避免不同 JCA Provider 的默认值不一致。
            OAEPParameterSpec oaep = new OAEPParameterSpec(
                    "SHA-256", "MGF1", MGF1ParameterSpec.SHA256, PSource.PSpecified.DEFAULT);
            cipher.init(Cipher.ENCRYPT_MODE, key, oaep);
            return Base64.getEncoder().encodeToString(cipher.doFinal(plaintext));
        } catch (GeneralSecurityException e) {
            throw new IllegalStateException("RSA-OAEP encryption failed", e);
        }
    }

    private RSAPublicKey getPublicKey() throws GeneralSecurityException {
        RSAPublicKey cached = publicKey;
        if (cached != null) {
            return cached;
        }

        synchronized (this) {
            if (publicKey == null) {
                String base64 = publicKeyPem
                        .replace("-----BEGIN PUBLIC KEY-----", "")
                        .replace("-----END PUBLIC KEY-----", "")
                        .replaceAll("\\s", "");
                PublicKey parsed = KeyFactory.getInstance("RSA").generatePublic(
                        new X509EncodedKeySpec(Base64.getDecoder().decode(base64)));
                if (!(parsed instanceof RSAPublicKey)) {
                    throw new IllegalArgumentException("configured key is not an RSA public key");
                }
                publicKey = (RSAPublicKey) parsed;
            }
            return publicKey;
        }
    }
}

生成部署所需的密钥。私钥不要放进 Flink 作业、JAR、Flink SQL 或日志；公钥可以作为作业配置传入。

# 生成 2048 位私钥；生产环境建议由 KMS、HSM 或密钥管理流程保管
openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048 \
  -out salary-rsa-private.pem

# 导出 UDF 所需的 X.509 PEM 公钥
openssl pkey -in salary-rsa-private.pem -pubout \
  -out salary-rsa-public.pem

在 Flink 1.12.7 Java 作业中注册函数实例，并在 SQL 中显式将数值型 salary 转为字符串：

import org.apache.flink.table.api.TableEnvironment;

import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;

String publicKeyPem = new String(
        Files.readAllBytes(Paths.get(System.getenv("SALARY_RSA_PUBLIC_KEY_FILE"))),
        StandardCharsets.UTF_8);

tableEnv.createTemporarySystemFunction(
        "salary_rsa_encrypt",
        new SalaryRsaEncryptFunction(publicKeyPem));

tableEnv.executeSql(
        "INSERT INTO employee_sink "
                + "SELECT employee_id, "
                + "salary_rsa_encrypt(CAST(salary AS STRING)) AS salary, "
                + "department "
                + "FROM employee_source");

2048 位 RSA 在 OAEP-SHA-256 下最多加密 190 字节明文，适合 salary 这类短字段；密文经 Base64 编码后固定为 344 个字符。下游字段建议至少使用 VARCHAR(344)。若要加密较长文本或整行数据，应改用 AES-GCM 加密数据、RSA-OAEP 仅加密 AES 密钥的混合加密方案。

🔗 Links

• flink docs
• flink docs history
• flink UDF 1.12