chunjun1.12 SQL使用自定义UDF函数 无法创建临时函数
- Dominant language
- Java
- Stars
- 4.1k
- Forks
- 1.7k
- PR merge metrics
- No merged PRs in 30d
Description
### Search before asking
- [X] I had searched in the [issues](https://github.com/DTStack/chunjun/issues) and found no similar issues.
### What happened
create temporary function myToTimestamp as 'com.biandou.functions.MyToTimestamp';
CREATE TABLE dwd_tb_data_relate_device
(
portNumber BIGINT,
deviceIndex INT,
deviceId int,
ts bigint,
WQP Double,
WPP Double,
`offset` BIGINT METADATA VIRTUAL
-- WATERMARK FOR tss as tss - INTERVAL '-1' MINUTES
) WITH (
'connector' = 'kafka-x'
,'topic' = 'dwd_tb_data_relate_device'
,'properties.bootstrap.servers' = '172.18.8.203:9092'
,'properties.group.id' = 'dwd'
,'scan.startup.mode' = 'earliest-offset'
,'format' = 'json'
,'json.timestamp-format.standard' = 'SQL'
,'json.ignore-parse-errors' = 'true'
);
CREATE TABLE dws_tb_data_energy
(
portNumber BIGINT,
deviceIndex INT,
deviceId int,
ts bigInt,
tss timestamp,
WQP Double,
WPP Double,
`offset` BIGINT
) WITH (
'connector' = 'stream-x',
'print' = 'true'
-- 'connector' = 'kafka-x'
-- ,'topic' = 'dwd_tb_data_relate_device'
-- ,'properties.bootstrap.servers' = '172.18.8.203:9092'
-- ,'format' = 'json'
-- ,'sink.parallelism' = '2'
-- ,'json.timestamp-format.standard' = 'SQL'
);
insert into dws_tb_data_energy
select portNumber,
deviceIndex,
deviceId,
ts,
myToTimestamp(ts) as tss,
WQP,
WPP,
'offset'
from dwd_tb_data_relate_device;
在chunjun1.12中执行:

在flink1.12中执行成功

### What you expected to happen
像flink本身一样 正常执行
package com.biandou.core.dwd;
import com.biandou.common.utils.MyKafkaUtil;
import com.biandou.functions.MyRowToColumn;
import com.biandou.functions.MyToTimestamp;
import com.biandou.functions.MyYmdH;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class Test002 {
public static void main(String[] args) {
//TODO 1.获取执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1); //生产环境中设置为Kafka主题的分区数
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
tableEnv.createFunction("myToTimestamp", MyToTimestamp.class);
String sourceSql = "CREATE TABLE dwd_tb_data_relate_device\n" +
"(\n" +
" portNumber BIGINT,\n" +
" deviceIndex INT,\n" +
" deviceId int,\n" +
" ts bigint,\n" +
" tss as myToTimestamp(ts),\n" +
" WQP Double,\n" +
" WPP Double,\n" +
" `offset` BIGINT METADATA VIRTUAL )"
+ MyKafkaUtil.getKafkaDDL("dwd_tb_data_relate_device", "dwd");
tableEnv.executeSql(sourceSql);
String querySQL = "select * from dwd_tb_data_relate_device ";
tableEnv.sqlQuery(querySQL).execute().print();
}
}
### How to reproduce
create temporary function myToTimestamp as 'com.biandou.functions.MyToTimestamp';
CREATE TABLE dwd_tb_data_relate_device
(
portNumber BIGINT,
deviceIndex INT,
deviceId int,
ts bigint,
WQP Double,
WPP Double,
`offset` BIGINT METADATA VIRTUAL
-- WATERMARK FOR tss as tss - INTERVAL '-1' MINUTES
) WITH (
'connector' = 'kafka-x'
,'topic' = 'dwd_tb_data_relate_device'
,'properties.bootstrap.servers' = '172.18.8.203:9092'
,'properties.group.id' = 'dwd'
,'scan.startup.mode' = 'earliest-offset'
,'format' = 'json'
,'json.timestamp-format.standard' = 'SQL'
,'json.ignore-parse-errors' = 'true'
);
CREATE TABLE dws_tb_data_energy
(
portNumber BIGINT,
deviceIndex INT,
deviceId int,
ts bigInt,
tss timestamp,
WQP Double,
WPP Double,
`offset` BIGINT
) WITH (
'connector' = 'stream-x',
'print' = 'true'
-- 'connector' = 'kafka-x'
-- ,'topic' = 'dwd_tb_data_relate_device'
-- ,'properties.bootstrap.servers' = '172.18.8.203:9092'
-- ,'format' = 'json'
-- ,'sink.parallelism' = '2'
-- ,'json.timestamp-format.standard' = 'SQL'
);
insert into dws_tb_data_energy
select portNumber,
deviceIndex,
deviceId,
ts,
myToTimestamp(ts) as tss,
WQP,
WPP,
'offset'
from dwd_tb_data_relate_device;
### Anything else
_No response_
### Version
1.12_release
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
Assessment
This issue has not been assessed yet.