[Flink] Version 1.0.2 Flink reports an error when running: java.io.FileNotFoundException job.sql cannot be found

[miaoze8]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

不确认是不是BUG，请帮忙看看

1.0.2版本Flink Per Job模式运行时候报错，File does not exist: /data1/dinky-1.0.2/tmp/5827f547-c821-4d20-9feb-1091dbc230c4/job.sql

但部署目录的tmp目录是存在此文件并且有权限访问

2024-05-08 10:46:31.341 WARN org.apache.flink.yarn.configuration.YarnLogConfigUtil(73): The configuration directory ('/data1/flink-1.19.0/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
2024-05-08 10:46:31.552 INFO org.apache.flink.yarn.YarnClusterDescriptor(267): No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
java.lang.RuntimeException: java.lang.RuntimeException: java.io.FileNotFoundException: File does not exist: /data1/dinky-1.0.2/tmp/5827f547-c821-4d20-9feb-1091dbc230c4/job.sql
at org.dinky.gateway.yarn.YarnPerJobGateway.submitJobGraph(YarnPerJobGateway.java:86)
at org.dinky.job.builder.JobTransBuilder.submitByGateway(JobTransBuilder.java:201)
at org.dinky.job.builder.JobTransBuilder.processWithGateway(JobTransBuilder.java:109)
at org.dinky.job.builder.JobTransBuilder.handleStatementSet(JobTransBuilder.java:81)
at org.dinky.job.builder.JobTransBuilder.run(JobTransBuilder.java:70)
at org.dinky.job.JobManager.executeSql(JobManager.java:361)
at org.dinky.service.task.FlinkSqlTask.execute(FlinkSqlTask.java:68)
at org.dinky.service.impl.TaskServiceImpl.executeJob(TaskServiceImpl.java:191)
at org.dinky.service.impl.TaskServiceImpl$$FastClassBySpringCGLIB$$22087f7c.invoke()
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:89)
at org.dinky.aop.ProcessAspect.processStepAround(ProcessAspect.java:110)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:634)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:624)
at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:72)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:175)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:97)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:708)
at org.dinky.service.impl.TaskServiceImpl$$EnhancerBySpringCGLIB$$cefcf291.executeJob()
at org.dinky.service.impl.TaskServiceImpl.submitTask(TaskServiceImpl.java:319)
at org.dinky.service.impl.TaskServiceImpl$$FastClassBySpringCGLIB$$22087f7c.invoke()
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:89)
at org.dinky.aop.UdfClassLoaderAspect.round(UdfClassLoaderAspect.java:55)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:634)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:624)
at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:72)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:97)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:708)
at org.dinky.service.impl.TaskServiceImpl$$EnhancerBySpringCGLIB$$cefcf291.submitTask()
at org.dinky.controller.TaskController.submitTask(TaskController.java:82)
at org.dinky.controller.TaskController$$FastClassBySpringCGLIB$$5b68bd32.invoke()
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.aspectj.AspectJAfterThrowingAdvice.invoke(AspectJAfterThrowingAdvice.java:64)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:175)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.framework.adapter.AfterReturningAdviceInterceptor.invoke(AfterReturningAdviceInterceptor.java:57)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:175)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:89)
at org.dinky.aop.ProcessAspect.processAround(ProcessAspect.java:73)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:634)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:624)
at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:72)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:175)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.framework.adapter.AfterReturningAdviceInterceptor.invoke(AfterReturningAdviceInterceptor.java:57)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:97)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:708)
at org.dinky.controller.TaskController$$EnhancerBySpringCGLIB$$e60139d5.submitTask()
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:205)
at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:150)
at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:117)
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:895)
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808)
at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1072)
at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:965)
at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006)
at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:529)
at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:623)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:209)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:51)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at org.springframework.web.filter.CorsFilter.doFilterInternal(CorsFilter.java:91)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:117)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at com.alibaba.druid.support.http.WebStatFilter.doFilter(WebStatFilter.java:124)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:117)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at cn.dev33.satoken.filter.SaPathCheckFilterForServlet.doFilter(SaPathCheckFilterForServlet.java:55)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:117)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.doFilterInternal(WebMvcMetricsFilter.java:96)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:117)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:117)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:178)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:153)
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:167)
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:90)
at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:481)
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:130)
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:93)
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:74)
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343)
at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:389)
at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:63)
at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:926)
at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1791)
at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:52)
at org.apache.tomcat.util.threads.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1191)
at org.apache.tomcat.util.threads.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:659)
at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: java.io.FileNotFoundException: File does not exist: /data1/dinky-1.0.2/tmp/5827f547-c821-4d20-9feb-1091dbc230c4/job.sql
at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321)
at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.tryAdvance(ArrayList.java:1359)
at java.util.stream.ReferencePipeline.forEachWithCancel(ReferencePipeline.java:126)
at java.util.stream.AbstractPipeline.copyIntoWithCancel(AbstractPipeline.java:498)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:485)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
at java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:230)
at java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:196)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.anyMatch(ReferencePipeline.java:449)
at org.apache.flink.yarn.YarnClusterDescriptor.isUsrLibDirIncludedInShipFiles(YarnClusterDescriptor.java:1913)
at org.apache.flink.yarn.YarnClusterDescriptor.addShipFiles(YarnClusterDescriptor.java:349)
at org.dinky.executor.ClusterDescriptorAdapterImpl.addShipFiles(ClusterDescriptorAdapterImpl.java:43)
at org.dinky.gateway.yarn.YarnPerJobGateway.submitJobGraph(YarnPerJobGateway.java:73)
... 145 more
Caused by: java.io.FileNotFoundException: File does not exist: /data1/dinky-1.0.2/tmp/5827f547-c821-4d20-9feb-1091dbc230c4/job.sql
at org.apache.hadoop.hdfs.DistributedFileSystem$27.doCall(DistributedFileSystem.java:1441)
at org.apache.hadoop.hdfs.DistributedFileSystem$27.doCall(DistributedFileSystem.java:1434)
at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1434)
at org.apache.flink.yarn.YarnClusterDescriptor.getFileStatus(YarnClusterDescriptor.java:386)
at org.apache.flink.yarn.YarnClusterDescriptor.lambda$isUsrLibDirIncludedInShipFiles$12(YarnClusterDescriptor.java:1910)
at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:73)
... 159 more

Process Step SUBMIT_EXECUTE exit with status:FAILED

Application模式也报错. 但部署目录的tmp目录是不存在此文件
java.lang.RuntimeException: java.lang.RuntimeException: java.io.FileNotFoundException: File does not exist: /data1/dinky-1.0.2/tmp/1d0023cd-6b8c-4d13-b444-926f2aaa9cf9/job.sql

[dinky] 2024-05-08 11:01:34 CST WARN org.apache.flink.yarn.configuration.YarnLogConfigUtil 73 discoverLogConfigFile - The configuration di rectory ('/data1/flink-1.19.0/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
[dinky] 2024-05-08 11:01:34 CST INFO org.apache.flink.yarn.YarnClusterDescriptor 267 getLocalFlinkDistPath - No path for the flink jar pas sed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
[dinky] 2024-05-08 11:01:34 CST INFO org.dinky.service.impl.TaskServiceImpl 192 executeJob - execute job finished,status is FAILED
[dinky] 2024-05-08 11:01:34 CST ERROR org.dinky.aop.exception.UnKnownExceptionHandler 39 unknownException - java.lang.RuntimeException: java.lang.RuntimeException: java.io.FileNotFoundException: File does not exist: /data1/dinky-1.0.2/tmp/1d0023cd-6b8c-4d13-b444-926f2aaa9cf9/job.sql
at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321)
at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.tryAdvance(ArrayList.java:1359)
at java.util.stream.ReferencePipeline.forEachWithCancel(ReferencePipeline.java:126)
at java.util.stream.AbstractPipeline.copyIntoWithCancel(AbstractPipeline.java:498)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:485)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
at java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:230)
at java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:196)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.anyMatch(ReferencePipeline.java:449)
at org.apache.flink.yarn.YarnClusterDescriptor.isUsrLibDirIncludedInShipFiles(YarnClusterDescriptor.java:1913)
at org.apache.flink.yarn.YarnClusterDescriptor.addShipFiles(YarnClusterDescriptor.java:349)
at org.dinky.executor.ClusterDescriptorAdapterImpl.addShipFiles(ClusterDescriptorAdapterImpl.java:43)
at org.dinky.gateway.yarn.YarnApplicationGateway.submitJar(YarnApplicationGateway.java:99)

What you expected to happen

Flink任务正常执行

How to reproduce

创建flink sql任务 使用官网demo代码
https://www.dinky.org.cn/docs/next/practical_guide/Integration_practice/dinky_integration_flink#%E5%88%9B%E5%BB%BA-datagen-%E4%BB%BB%E5%8A%A1

使用Per Job方式时执行报错
java.lang.RuntimeException: java.lang.RuntimeException: java.io.FileNotFoundException: File does not exist: /data1/dinky-1.0.2/tmp/5827f547-c821-4d20-9feb-1091dbc230c4/job.sql

Anything else

No response

Version

dev

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[github-actions]

Hello @miaoze8, this issue is about CDC/CDCSOURCE, so I assign it to @aiwenmo. If you have any questions, you can comment and reply.

你好 @miaoze8, 这个 issue 是关于 CDC/CDCSOURCE 的，所以我把它分配给了 @aiwenmo。如有任何问题，可以评论回复。

[Zzm0809]

在环境正常情况下 无法复现
以下为复现步骤:

DROP TABLE IF EXISTS source_table3;
CREATE TABLE IF NOT EXISTS source_table3(
--订单id
`order_id` BIGINT,
--产品

`product` BIGINT,
--金额
`amount` BIGINT,

--支付时间
`order_time` as CAST(CURRENT_TIMESTAMP AS TIMESTAMP(3)), -- `在这里插入代码片`
--WATERMARK
WATERMARK FOR order_time AS order_time - INTERVAL '2' SECOND
) WITH(
'connector' = 'datagen',
 'rows-per-second' = '1',
 'fields.order_id.min' = '1',
 'fields.order_id.max' = '2',
 'fields.amount.min' = '1',
 'fields.amount.max' = '10',
 'fields.product.min' = '1',
 'fields.product.max' = '2'
);

-- SELECT * FROM source_table3 LIMIT 10;

DROP TABLE IF EXISTS sink_table5;
CREATE TABLE IF NOT EXISTS sink_table5(
--产品
`product` BIGINT,
--金额
`amount` BIGINT,
--支付时间
`order_time` TIMESTAMP(3),
--1分钟时间聚合总数
`one_minute_sum` BIGINT
) WITH(
'connector'='print'
);

INSERT INTO sink_table5
SELECT
product,
amount,
order_time,
SUM(amount) OVER(
PARTITION BY product
ORDER BY order_time
-- 标识统计范围是1个 product 的最近 1 分钟的数据
RANGE BETWEEN INTERVAL '1' MINUTE PRECEDING AND CURRENT ROW
) as one_minute_sum
FROM source_table3;

[miaoze8]

1.0.1版本同样配置perjob能成功递交 并且发现在dinky部署目录tmp里面也没有类似这种临时文件/5827f547-c821-4d20-9feb-1091dbc230c4/job.sql

请问下正常情况递交flink任务时候是需要在dinky部署目录tmp里面先生成 xxxxx/job.sql的文件么?

这个功能在1.0.2版本有调整么?

======
另外1.0.2报错里面有这个
Caused by: java.io.FileNotFoundException: File does not exist: /data1/dinky-1.0.2/tmp/a71e4cd7-452c-4d71-baf0-4fadc9756d6e/job.sql
at org.apache.hadoop.hdfs.DistributedFileSystem$29.doCall(DistributedFileSystem.java:1749)

是不是获取到hdfs里面去了?

[miaoze8]

@Zzm0809

[Zzm0809]

测下来 1.19 的支持可能存在问题, 建议先使用 1.19 以下版本

[gaoyan1998]

link #3490