(no subject)
Locked
(no subject)
 
|
Hi all, I'm currently testing PyFlink against PySpark for batch jobs. The query I'm using is:selectMy configuration are default, created from official docker image:jobmanager.heap.size 1024m Thanks! Violeta |
Re:
|
|
Re:
|
|
Re:
|
|
Re:
|
|
Re:
|
|
Hi Timo, Thanks for your answers, I tried again with new connectors, and blink planner and execute_sql(), plan is simpler indeed, but results are almost the same.  Tried both queries with 1.11 version, 7-join is old one with old connectors, sql queries and planner, and 7b is the new one with new connectors, execute_sql() and blink planner  As you can see almost the same, these are two consecutive runs of quieres. I stayed with version 1.10 and old connectors and planner for now, and version 1.10 does not support execute_sql()  Thanks, Violeta On Thu, Sep 10, 2020 at 11:45 AM Timo Walther <[hidden email]> wrote: Hi Violeta, |
Re:
|
|
Could you upload the log of the last operator, the one begins with "Sort"? Best, Kurt On Fri, Sep 11, 2020 at 4:54 AM Violeta Milanović <[hidden email]> wrote:
|
Re:
|
|
Hi Violeta, Sorry for the late reply. I'm curious why the last operator ("Sort -> SortAgg -> CsvSink") is so slow, it's input has only 90M data. Regarding to your first image, the last operator took nearly two minutes, which only writes 32M data. So I guess the CSV writer is the performance bottleneck. Could you use other sinks (e.g. print sink) instead of csv sink to check whether only the csv sink is the bottleneck. Best, Godfrey Kurt Young <[hidden email]> 于2020年9月16日周三 下午2:22写道：
|
Re:
|
|
Hi Violeta, I did a similar test in my local environment based on TPCH [1] test data with scale factor 1. I started a session cluster in my local, and submitted the test query in sql client with blink planner. here is my test details: DDLs: create table lineitem ( l_orderkey bigint not null, l_partkey bigint not null, l_suppkey bigint not null, l_linenumber int not null, l_quantity double not null, l_extendedprice double not null, l_discount double not null, l_tax double not null, l_returnflag varchar not null, l_linestatus varchar not null, l_shipdate date not null, l_commitdate date not null, l_receiptdate date not null, l_shipinstruct varchar not null, l_shipmode varchar not null, l_comment varchar not null ) with ( 'connector.type'='filesystem', 'format.type'='csv', 'format.field-delimiter'='|', 'connector.path'='file:///<your_tpch_data_path>/lineitem.tbl' ); create table orders ( o_orderkey bigint not null, o_custkey bigint not null, o_orderstatus varchar not null, o_totalprice double not null, o_orderdate date not null, o_orderpriority varchar not null, o_clerk varchar not null, o_shippriority int not null, o_comment varchar not null ) with ( 'connector.type'='filesystem', 'format.type'='csv', 'format.field-delimiter'='|', 'connector.path'='file:///<your_tpch_data_path>/orders.tbl' ); create table my_sink ( o_orderstatus varchar, o_comment varchar, avg_ta double not null, avg_income double not null, spending double not null ) with ( 'connector.type'='filesystem', 'format.type'='csv', 'format.field-delimiter'='|', 'connector.path'='file:///<your_result_path>/result' ); QUERY (similar with yours) insert into my_sink select max(c.o_orderstatus), max(c.o_comment), avg(l_quantity) as avg_ta, avg(l_extendedprice+l_tax) as avg_income, avg(l_extendedprice+l_tax) - avg(l_discount) as spending from lineitem t left join orders c on t.l_orderkey = c.o_orderkey group by c.o_orderkey; And here is my execution result:  the last operator only takes 9 seconds. So, just as Kurt's mentioned, we need your execution log to do more analysis. Best, Godfrey godfrey he <[hidden email]> 于2020年9月16日周三 下午10:45写道：
|
Re:
|
|
Hi Godfrey, I'm sending you my log files. Best, Violeta On Mon, Sep 21, 2020 at 3:49 AM godfrey he <[hidden email]> wrote:
 flink--client-jobmanager.log (19K) Download Attachment flink--standalonesession-0-jobmanager.log (133K) Download Attachment flink--taskexecutor-0-27ab260df8e6.log (685K) Download Attachment |
Re:
|
|
Hi Violeta, I do not find any useful info in you logs, Only brief information shows that the last operator started running at 05:19:37,427 and finished at 05:21:41,621. 2020-09-21 05:19:37,427 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Sort(orderBy=[customer_id0 ASC]) -> SortAggregate(isMerge=[true], groupBy=[customer_id0], select=[customer_id0, Final_MAX(max$0) AS EXPR$0, Final_MAX(max$1) AS EXPR$1, Final_AVG(sum$2, count$3) AS avg_ta, Final_AVG(sum$4, count$5) AS avg_income]) -> Calc(select=[EXPR$0, EXPR$1, avg_ta, avg_income, (avg_income - avg_ta) AS spending]) -> SinkConversionToRow -> Map -> Sink: CsvTableSink(first_name, last_name, avg_ta, avg_income, spending) (1/6) (f8fa1ac890d6482446298d03061c9316) switched from DEPLOYING to RUNNING. 2020-09-21 05:21:41,621 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Sort(orderBy=[customer_id0 ASC]) -> SortAggregate(isMerge=[true], groupBy=[customer_id0], select=[customer_id0, Final_MAX(max$0) AS EXPR$0, Final_MAX(max$1) AS EXPR$1, Final_AVG(sum$2, count$3) AS avg_ta, Final_AVG(sum$4, count$5) AS avg_income]) -> Calc(select=[EXPR$0, EXPR$1, avg_ta, avg_income, (avg_income - avg_ta) AS spending]) -> SinkConversionToRow -> Map -> Sink: CsvTableSink(first_name, last_name, avg_ta, avg_income, spending) (1/6) (f8fa1ac890d6482446298d03061c9316) switched from RUNNING to FINISHED. No more detailed info is in the log files. Can you change the log level to DEBUG, or can you try to replace the csv sink with other sink ? Best, Godfrey Violeta Milanović <[hidden email]> 于2020年9月21日周一 下午1:34写道：
|
| Free forum by Nabble | Edit this page |