Calibre | 将EPUB转换为微软正黑体

-- 建表语句，用于接收 Sink 端数据
CREATE TABLE `udf_output` (
  `id`        int(10) NOT NULL,
  `len_name`  int(10) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

// 类名：StringLengthUdf
package demos.UDF;
​
import org.apache.flink.table.functions.ScalarFunction;
​
public class StringLengthUdf extends ScalarFunction {
​
    public long eval(String a) {
        return a == null ? 0 : a.length();
    }
​
    public long eval(String b, String c) {
        return eval(b) + eval(c);
    }
}

mvn clean package

CREATE TEMPORARY SYSTEM FUNCTION StringLengthUdf AS 'demos.UDF.StringLengthUdf';

CREATE TABLE random_source ( 
  id INT, 
  name1 VARCHAR, 
  name2 VARCHAR 
  ) WITH ( 
  'connector' = 'datagen', 
  'rows-per-second'='1',          -- 每秒产生的数据条数
  'fields.id.kind'='sequence',    -- 无界的随机数
  'fields.id.start'='1',          -- 随机数的最小值
  'fields.id.end'='5',            -- 随机数的最大值
  'fields.name1.length'='10',     -- 随机字符串的长度
  'fields.name2.length'='10'      -- 随机字符串的长度
);

CREATE TABLE `jdbc_upsert_sink_table` (
    `id` INT,
    `len_name` INT,
    PRIMARY KEY(id) NOT ENFORCED
) WITH (
    -- 指定数据库连接参数
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://xx.xx.xx.xx:xxxx/testdb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',         -- 请替换为您的实际 MySQL 连接参数
    'table-name' = 'udf_output',           -- 需要写入的数据表
    'username' = 'root',                   -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'xxxxxxxxx',              -- 数据库访问的密码
    'sink.buffer-flush.max-rows' = '200',  -- 批量输出的条数
    'sink.buffer-flush.interval' = '2s'    -- 批量输出的间隔
);

INSERT INTO jdbc_upsert_sink_table
SELECT
id,
CAST(StringLengthUdf(name1,name2) AS INT) AS `len_name`
FROM random_source;