Thanks Tianlang. I saw the DAG on YARN, but what really solved my problem is adding intermediate steps and evaluating them eagerly to find out where the bottleneck was. My process now runs in 6 min. :D
Thanks for the help. []s On Thu, 15 Aug 2019 at 07:25, Tianlang <tianlangstu...@aliyun.com.invalid> wrote: > Hi, > > Maybe you can look at the spark ui. The physical plan has no time > consuming information. > 在 2019/8/13 下午10:45, Marcelo Valle 写道: > > Hi, > > I have a job running on AWS EMR. It's basically a join between 2 tables > (parquet files on s3), one somehow large (around 50 gb) and other small > (less than 1gb). > The small table is the result of other operations, but it was a dataframe > with `.persist(StorageLevel.MEMORY_AND_DISK_SER)` and the count on this > dataframe finishes quickly. > When I run my "LEFT_ANTI" join, I get the execution plan down bellow. > While most of my jobs on larges amount of data take max 1 h on this > cluster, this one takes almost 1 day to complete. > > What could I be doing wrong? I am trying to analyze the plan, but I can't > find anything that justify the slowness. It has 2 shuffles followed by a > zip, but other jobs have similar things and they are not that slow. > > Could anyone point me to possible actions I could take to investigate this? > > Thanks, > Marcelo. > > == Physical Plan == > *(2) Project [USAGE_AGGREGATED_METADATA_ID#1493, > SENDER_RECORDING_IDENTIFIER#1499, AIP127258 AS SENDER_IP_ID#1702, > USAGE_AGGREGATED_METADATA_HASH#1513] > +- *(2) BroadcastHashJoin [coalesce(USAGE_AGGREGATED_METADATA_ID#1493, ), > coalesce(SENDER_RECORDING_IDENTIFIER#1499, )], > [coalesce(USAGE_AGGREGATED_METADATA_ID#356, ), > coalesce(SENDER_RECORDING_IDENTIFIER#357, )], LeftAnti, BuildRight, > ((USAGE_AGGREGATED_METADATA_ID#356 <=> USAGE_AGGREGATED_METADATA_ID#1493) > && (SENDER_RECORDING_IDENTIFIER#357 <=> SENDER_RECORDING_IDENTIFIER#1499)) > :- InMemoryTableScan [USAGE_AGGREGATED_METADATA_ID#1493, > SENDER_RECORDING_IDENTIFIER#1499, USAGE_AGGREGATED_METADATA_HASH#1513] > : +- InMemoryRelation [USAGE_AGGREGATED_METADATA_ID#1493, > ISRC#1494, ISWC#1495, RECORDING_TITLE#1496, RECORDING_DISPLAY_ARTIST#1497, > WORK_WRITERS#1498, SENDER_RECORDING_IDENTIFIER#1499, > RECORDING_VERSION_TITLE#1500, WORK_TITLE#1501, CONTENT_TYPE#1502, > USAGE_AGGREGATED_METADATA_HASH#1513], StorageLevel(disk, memory, 1 replicas) > : +- *(2) Project [ID#328 AS > USAGE_AGGREGATED_METADATA_ID#1493, isrc#289 AS ISRC#1494, iswc#290 AS > ISWC#1495, track_name#291 AS RECORDING_TITLE#1496, artist_name#292 AS > RECORDING_DISPLAY_ARTIST#1497, work_writer_names#293 AS WORK_WRITERS#1498, > uri#286 AS SENDER_RECORDING_IDENTIFIER#1499, null AS > RECORDING_VERSION_TITLE#1500, null AS WORK_TITLE#1501, SOUND AS > CONTENT_TYPE#1502, UDF(array(isrc#289, track_name#291, null, > artist_name#292, iswc#290, null, work_writer_names#293, SOUND)) AS > USAGE_AGGREGATED_METADATA_HASH#1513] > : +- *(2) BroadcastHashJoin [coalesce(isrc_1#1419, ), > coalesce(iswc_1#1420, ), coalesce(track_name_1#1421, ), > coalesce(artist_name_1#1422, ), coalesce(work_writer_names_1#1423, )], > [coalesce(isrc#289, ), coalesce(iswc#290, ), coalesce(track_name#291, ), > coalesce(artist_name#292, ), coalesce(work_writer_names#293, )], Inner, > BuildLeft, (((((isrc#289 <=> isrc_1#1419) && (iswc#290 <=> iswc_1#1420)) && > (track_name#291 <=> track_name_1#1421)) && (artist_name#292 <=> > artist_name_1#1422)) && (work_writer_names#293 <=> > work_writer_names_1#1423)) > : :- BroadcastExchange > HashedRelationBroadcastMode(List(coalesce(input[1, string, true], ), > coalesce(input[2, string, true], ), coalesce(input[3, string, true], ), > coalesce(input[4, string, true], ), coalesce(input[5, string, true], ))) > : : +- *(1) Project [ID#328, isrc#289 AS isrc_1#1419, > iswc#290 AS iswc_1#1420, track_name#291 AS track_name_1#1421, > artist_name#292 AS artist_name_1#1422, work_writer_names#293 AS > work_writer_names_1#1423] > : : +- *(1) Filter isnotnull(ID#328) > : : +- InMemoryTableScan [ID#328, > artist_name#292, isrc#289, iswc#290, track_name#291, > work_writer_names#293], [isnotnull(ID#328)] > : : +- InMemoryRelation [ID#328, isrc#289, > iswc#290, track_name#291, artist_name#292, work_writer_names#293], > StorageLevel(disk, memory, 1 replicas) > : : +- *(2) Project [ID#328, > isrc#289, iswc#290, track_name#291, artist_name#292, work_writer_names#293] > : : +- *(2) BroadcastHashJoin > [coalesce(ISRC#329, ), coalesce(ISWC#330, ), coalesce(RECORDING_TITLE#331, > ), coalesce(RECORDING_DISPLAY_ARTIST#332, ), coalesce(WORK_WRITERS#333, )], > [coalesce(isrc#289, ), coalesce(iswc#290, ), coalesce(track_name#291, ), > coalesce(substring(artist_name#292, 0, 1000), ), > coalesce(work_writer_names#293, )], RightOuter, BuildLeft, (((((isrc#289 > <=> ISRC#329) && (iswc#290 <=> ISWC#330)) && (track_name#291 <=> > RECORDING_TITLE#331)) && (substring(artist_name#292, 0, 1000) <=> > RECORDING_DISPLAY_ARTIST#332)) && (work_writer_names#293 <=> > WORK_WRITERS#333)) > : : :- BroadcastExchange > HashedRelationBroadcastMode(List(coalesce(input[1, string, true], ), > coalesce(input[2, string, true], ), coalesce(input[3, string, true], ), > coalesce(input[4, string, true], ), coalesce(input[5, string, true], ))) > : : : +- *(1) Project > [ID#328, ISRC#329, ISWC#330, RECORDING_TITLE#331, > RECORDING_DISPLAY_ARTIST#332, WORK_WRITERS#333] > : : : +- *(1) Filter > ((isnull(WORK_TITLE#334) && isnull(RECORDING_VERSION_TITLE#335)) && > (CONTENT_TYPE#336 <=> SOUND)) > : : : +- *(1) FileScan > parquet > [ID#328,ISRC#329,ISWC#330,RECORDING_TITLE#331,RECORDING_DISPLAY_ARTIST#332,WORK_WRITERS#333,WORK_TITLE#334,RECORDING_VERSION_TITLE#335,CONTENT_TYPE#336] > Batched: true, Format: Parquet, Location: InMemoryFileIndex[ > file:/Users/marcelo.valle/git/amra-cloud-usage-ingestion/target/test-classes/ua..., > PartitionFilters: [], PushedFilters: [IsNull(WORK_TITLE), > IsNull(RECORDING_VERSION_TITLE), EqualNullSafe(CONTENT_TYPE,SOUND)], > ReadSchema: > struct<ID:string,ISRC:string,ISWC:string,RECORDING_TITLE:string,RECORDING_DISPLAY_ARTIST:string,W... > : : +- *(2) FileScan parquet > [isrc#289,iswc#290,track_name#291,artist_name#292,work_writer_names#293] > Batched: true, Format: Parquet, Location: InMemoryFileIndex[ > file:/Users/marcelo.valle/git/amra-cloud-usage-ingestion/target/test-classes/ua..., > PartitionFilters: [], PushedFilters: [], ReadSchema: > struct<isrc:string,iswc:string,track_name:string,artist_name:string,work_writer_names:string> > : +- *(2) FileScan parquet > [uri#286,isrc#289,iswc#290,track_name#291,artist_name#292,work_writer_names#293] > Batched: true, Format: Parquet, Location: InMemoryFileIndex[ > file:/Users/marcelo.valle/git/amra-cloud-usage-ingestion/target/test-classes/ua..., > PartitionFilters: [], PushedFilters: [], ReadSchema: > struct<uri:string,isrc:string,iswc:string,track_name:string,artist_name:string,work_writer_names:... > +- BroadcastExchange HashedRelationBroadcastMode(List(coalesce(input[0, > string, true], ), coalesce(input[1, string, true], ))) > +- *(1) FileScan parquet > [USAGE_AGGREGATED_METADATA_ID#356,SENDER_RECORDING_IDENTIFIER#357] Batched: > true, Format: Parquet, Location: InMemoryFileIndex[ > file:/Users/marcelo.valle/git/amra-cloud-usage-ingestion/target/test-classes/ua..., > PartitionFilters: [], PushedFilters: [], ReadSchema: > struct<USAGE_AGGREGATED_METADATA_ID:string,SENDER_RECORDING_IDENTIFIER:string> > > This email is confidential [and may be protected by legal privilege]. If > you are not the intended recipient, please do not copy or disclose its > content but contact the sender immediately upon receipt. > > KTech Services Ltd is registered in England as company number 10704940. > > Registered Office: The River Building, 1 Cousin Lane, London EC4R 3TE, > United Kingdom > > -- > > TianlangStudio <https://edu.51cto.com/sd/aca72> > > Some of the biggest lies: I will start tomorrow/Others are better than > me/I am not good enough/I don't have time/This is the way I am > <https://edu.51cto.com/sd/aca72> > <https://github.com/TianLangStudio> > > > This email is confidential [and may be protected by legal privilege]. If you are not the intended recipient, please do not copy or disclose its content but contact the sender immediately upon receipt. KTech Services Ltd is registered in England as company number 10704940. Registered Office: The River Building, 1 Cousin Lane, London EC4R 3TE, United Kingdom
