Hi Hequn, I think it's obvious when we see the job graph for 200 unions. I have attached the screenshot here with.
I also tried out different approach , which is instead of UNION ALL On Mon, Jan 14, 2019 at 2:57 PM Hequn Cheng <chenghe...@gmail.com> wrote: > Hi dhanuka, > > > I am trying to deploy 200 SQL unions and it seems all the tasks getting > failing after some time. > Would be great if you can show us some information(say exception stack) > about the failure. Is it caused by OOM of job manager? > > > How do i allocate memory for task manager and job manager. What are the > factors need to be considered . > According to your SQL, I guess you need more memory for the job manager[1] > since you unionAll 200 tables, the job graph should be a bit big. As for > the taskmanger, I think it may be ok to use the default memory setting > unless you allocate a lot of memory in your UDFs or you just want to make > better use of the memory(we can discuss more if you like). > > Best, Hequn > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/ops/config.html#jobmanager > > On Mon, Jan 14, 2019 at 9:54 AM dhanuka ranasinghe < > dhanuka.priyan...@gmail.com> wrote: > >> Hi Fabian, >> >> Thanks for the prompt reply and its working 🤗. >> >> I am trying to deploy 200 SQL unions and it seems all the tasks getting >> failing after some time. >> >> How do i allocate memory for task manager and job manager. What are the >> factors need to be considered . >> >> Cheers >> Dhanuka >> >> On Sun, 13 Jan 2019, 22:05 Fabian Hueske <fhue...@gmail.com wrote: >> >> > Hi Dhanuka, >> > >> > The important error message here is "AppendStreamTableSink requires that >> > Table has only insert changes". >> > This is because you use UNION instead of UNION ALL, which implies >> > duplicate elimination. >> > Unfortunately, UNION is currently internally implemented as a regular >> > aggregration which produces a retraction stream (although, this would >> not >> > be necessary). >> > >> > If you don't require duplicate elimination, you can replace UNION by >> UNION >> > ALL and the query should work. >> > If you require duplicate elimination, it is currently not possible to >> use >> > SQL for your use case. >> > >> > There is thea Jira issue FLINK-9422 to improve this case [1]. >> > >> > Best, Fabian >> > >> > [1] https://issues.apache.org/jira/browse/FLINK-9422 >> > >> > Am So., 13. Jan. 2019 um 14:43 Uhr schrieb dhanuka ranasinghe < >> > dhanuka.priyan...@gmail.com>: >> > >> >> Hi All, >> >> >> >> I am trying to select multiple results from Kafka and send results to >> >> Kafka >> >> different topic using Table API. But I am getting below error. Could >> you >> >> please help me on this. >> >> >> >> Query: >> >> >> >> SELECT TiggerID,'Rule3' as RuleName,mytime(ts) as ts1 , >> >> mytime(CURRENT_TIMESTAMP) as ts2 FROM dataTable WHERE InterceptID = >> >> 4508724 >> >> AND Provider_Info = 'Dhanuka' AND LIID LIKE '%193400835%' >> >> UNION >> >> SELECT TiggerID,'Rule2' as RuleName,mytime(ts) as ts1 , >> >> mytime(CURRENT_TIMESTAMP) as ts2 FROM dataTable WHERE InterceptID = >> >> 4508724 >> >> AND Provider_Info = 'Dhanuka' AND LIID LIKE '%193400835%' >> >> UNION >> >> SELECT TiggerID,'Rule1' as RuleName,mytime(ts) as ts1 , >> >> mytime(CURRENT_TIMESTAMP) as ts2 FROM dataTable WHERE InterceptID = >> >> 4508724 >> >> AND Provider_Info = 'Dhanuka' AND LIID LIKE '%193400835%' >> >> >> >> >> >> *Error:* >> >> >> >> 2019-01-13 21:36:36,228 ERROR >> >> org.apache.flink.runtime.webmonitor.handlers.JarRunHandler - >> Exception >> >> occurred in REST handler. >> >> org.apache.flink.runtime.rest.handler.RestHandlerException: >> >> org.apache.flink.client.program.ProgramInvocationException: The main >> >> method >> >> caused an error. >> >> at >> >> >> >> >> org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$7(JarRunHandler.java:151) >> >> at >> >> >> >> >> java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:870) >> >> at >> >> >> >> >> java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:852) >> >> at >> >> >> >> >> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) >> >> at >> >> >> >> >> java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1595) >> >> at >> >> >> >> >> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) >> >> at >> >> >> >> >> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) >> >> at java.lang.Thread.run(Thread.java:748) >> >> Caused by: java.util.concurrent.CompletionException: >> >> org.apache.flink.client.program.ProgramInvocationException: The main >> >> method >> >> caused an error. >> >> at >> >> >> >> >> org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$getJobGraphAsync$10(JarRunHandler.java:228) >> >> at >> >> >> >> >> java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) >> >> ... 3 more >> >> Caused by: org.apache.flink.client.program.ProgramInvocationException: >> The >> >> main method caused an error. >> >> at >> >> >> >> >> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:546) >> >> at >> >> >> >> >> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421) >> >> at >> >> >> >> >> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:83) >> >> at >> >> >> >> >> org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:78) >> >> at >> >> >> >> >> org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:120) >> >> at >> >> >> >> >> org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$getJobGraphAsync$10(JarRunHandler.java:226) >> >> ... 4 more >> >> Caused by: org.apache.flink.table.api.TableException: >> >> AppendStreamTableSink >> >> requires that Table has only insert changes. >> >> at >> >> >> >> >> org.apache.flink.table.api.StreamTableEnvironment.writeToSink(StreamTableEnvironment.scala:382) >> >> at >> >> >> >> >> org.apache.flink.table.api.TableEnvironment.insertInto(TableEnvironment.scala:784) >> >> at org.apache.flink.table.api.Table.insertInto(table.scala:877) >> >> at >> >> >> >> >> org.monitoring.stream.analytics.FlinkDynamicLocalSQL.main(FlinkDynamicLocalSQL.java:153) >> >> 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.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529) >> >> ... 9 more >> >> >> >> >> >> *Source Code:* >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> *StreamTableEnvironment tableEnv = >> >> TableEnvironment.getTableEnvironment(env); >> >> tableEnv.registerFunction("mytime", new MyTime(10)); >> tableEnv.connect(new >> >> Kafka().version("0.10").topic("testin") .properties(kConsumer) >> >> .startFromLatest()) .withFormat(new >> >> >> Json().jsonSchema(schemaContent).failOnMissingField(false).deriveSchema()) >> >> .withSchema(new Schema() .field("InterceptID", "DECIMAL") >> >> .field("Provider_Info", "VARCHAR") .field("LIID", "VARCHAR") >> >> .field("TiggerID", "DECIMAL") .field("ts", Types.SQL_TIMESTAMP()) >> >> .rowtime(new >> >> Rowtime().timestampsFromSource().watermarksPeriodicBounded(1000))) >> >> .inAppendMode() .registerTableSource(sourceTable); // WindowedTable >> >> windowedTable = // >> >> tableEnv.scan(sourceTable).window(Tumble.over("50.minutes")); >> >> //tableEnv.sqlQuery(query) StringBuilder multi = new StringBuilder(); >> >> for(String sql : rules) { if(multi.length() > 0) { multi.append(" >> >> UNION >> >> ").append("\n"); } multi.append( sql); } >> >> LOGGER.info("********************************* " + multi.toString()); >> >> Table >> >> result = tableEnv.sqlQuery(multi.toString()); tableEnv // declare the >> >> external system to connect to .connect(new Kafka().version("0.10") >> >> .topic("testout").startFromEarliest() .properties(kProducer) ) >> >> .withFormat(new Json().failOnMissingField(false).deriveSchema()) >> >> .withSchema(new Schema() .field("TiggerID", Types.DECIMAL()) >> >> .field("RuleName", Types.STRING()) .field("ts1", Types.STRING()) >> >> .field("ts2", Types.STRING()) ) // specify the update-mode for >> streaming >> >> tables .inAppendMode() // register as source, sink, or both and under a >> >> name .registerTableSourceAndSink("ruleTable"); //tableEnv.sqlUpdate( >> >> "INSERT INTO ruleTable " + result); result.insertInto("ruleTable"); >> >> Cheers,Dhanuka* >> >> >> >> >> >> >> >> -- >> >> Nothing Impossible,Creativity is more important than knowledge. >> >> >> > >> > -- Nothing Impossible,Creativity is more important than knowledge.
