DTStack / DTStack/chunjun

chunjun1.12 SQL使用自定义UDF函数 无法创建临时函数

Open
#1,475 2 comments 0 reactions 0 assignees View on GitHub
bug
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中执行:
![1673245902(1)](https://user-images.githubusercontent.com/34857750/211251870-21f0ce36-deb5-45aa-a4d6-d1c0b717b1ed.png)
在flink1.12中执行成功
![1673245928(1)](https://user-images.githubusercontent.com/34857750/211251917-6ca14896-e5f2-4e4f-9710-088ff2cdc4f0.png)

### 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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.