Flink这个是什么原因呀？

从你写的这个结构看，没有使用 catalog
那点调试的前提是你这个脚本要完整
完整的 source 的 DDL 完整的 sink DDL 完整的 DML
如下所示：
CREATE TEMPORARY TABLE datagenSource (
input varchar
)with (
'connector' = 'datagen'
);

CREATE TEMPORARY TABLE printSink (
b1 bigint
)with (
'connector' = 'print',
'limit' = '100'
);

INSERT INTO printSink
SELECT
cast(substring(cast(CURRENT_TIMESTAMP as varchar),-3,3)as bigint)+unix_timestamp(CURRENT_TIMESTAMP)*1000
FROM
datagenSource;
source sink 的 temporary 关键词、DML 的INSERT INTO 。此回答整理自钉群“实时计算Flink产品交流群”