flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhanuka ranasinghe <dhanuka.priyan...@gmail.com>
Subject Re: Multiple select single result
Date Mon, 14 Jan 2019 07:35:21 GMT
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 <chenghequn@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.priyanath@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 <fhueske@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.priyanath@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.

Mime
View raw message