Re: Job migrated from EMR to Dataproc takes 20 hours instead of 90 minutes
Gzip files are not splittable. Hence using very large (i.e. non partitioned) gzip files lead to contention at reading the files as readers cannot scale beyond the number of gzip files to read. Better to use a splittable compression format instead to allow frameworks to scale up. Or manually manage scaling by using partitions, as you are doing now. On Mon, 30 May 2022, 08:54 Ori Popowski, wrote: > Thanks. > > Eventually the problem was solved. I am still not 100% sure what caused it > but when I said the input was identical I simplified a bit because it was > not (sorry for misleading, I thought this information would just be noise). > Explanation: the input to the EMR job was gzips created by Firehose and > partitioned hourly. The input to the Dataproc job is gzips created by Kafka > Connect and is not partitioned hourly. Otherwise the *content* itself is > identical. > > When we started partitioning the files hourly the problem went away > completely. > > I am still not sure what's going on exactly. If someone has some insight > it's welcome. > > Thanks! > > On Fri, May 27, 2022 at 9:45 PM Aniket Mokashi > wrote: > >> +cloud-dataproc-discuss >> >> On Wed, May 25, 2022 at 12:33 AM Ranadip Chatterjee >> wrote: >> >>> To me, it seems like the data being processed on the 2 systems is not >>> identical. Can't think of any other reason why the single task stage will >>> get a different number of input records in the 2 cases. 700gb of input to a >>> single task is not good, and seems to be the bottleneck. >>> >>> On Wed, 25 May 2022, 06:32 Ori Popowski, wrote: >>> Hi, Both jobs use spark.dynamicAllocation.enabled so there's no need to change the number of executors. There are 702 executors in the Dataproc cluster so this is not the problem. About number of partitions - this I didn't change and it's still 400. While writing this now, I am realising that I have more partitions than executors, but the same situation applies to EMR. I am observing 1 task in the final stage also on EMR. The difference is that on EMR that task receives 50K volume of data and on Dataproc it receives 700gb. I don't understand why it's happening. It can mean that the graph is different. But the job is exactly the same. Could it be because the minor version of Spark is different? On Wed, May 25, 2022 at 12:27 AM Ranadip Chatterjee < ranadi...@gmail.com> wrote: > Hi Ori, > > A single task for the final step can result from various scenarios > like an aggregate operation that results in only 1 value (e.g count) or a > key based aggregate with only 1 key for example. There could be other > scenarios as well. However, that would be the case in both EMR and > Dataproc > if the same code is run on the same data in both cases. > > On a separate note, since you have now changed the size and number of > nodes, you may need to re-optimize the number and size of executors for > the > job and perhaps the number of partitions as well to optimally use the > cluster resources. > > Regards, > Ranadip > > On Tue, 24 May 2022, 10:45 Ori Popowski, wrote: > >> Hello >> >> I migrated a job from EMR with Spark 2.4.4 to Dataproc with Spark >> 2.4.8. I am creating a cluster with the exact same configuration, where >> the >> only difference is that the original cluster uses 78 workers with 96 CPUs >> and 768GiB memory each, and in the new cluster I am using 117 machines >> with >> 64 CPUs and 512GiB each, to achieve the same amount of resources in the >> cluster. >> >> The job is run with the same configuration (num of partitions, >> parallelism, etc.) and reads the same data. However, something strange >> happens and the job takes 20 hours. What I observed is that there is a >> stage where the driver instantiates a single task, and this task never >> starts because the shuffle of moving all the data to it takes forever. >> >> I also compared the runtime configuration and found some minor >> differences (due to Dataproc being different from EMR) but I haven't >> found >> any substantial difference. >> >> In other stages the cluster utilizes all the partitions (400), and >> it's not clear to me why it decides to invoke a single task. >> >> Can anyone provide an insight as to why such a thing would happen? >> >> Thanks >> >> >> - >> To unsubscribe e-mail: user-unsubscr...@spark.apache.org > > >> >> -- >> "...:::Aniket:::... Quetzalco@tl" >> >
Re: protobuf data as input to spark streaming
Hello Stelios, friendly reminder if you could share any sample code/repo Are you using a schema registry? Thanks Kiran On Fri, Apr 8, 2022 at 4:37 PM Kiran Biswal wrote: > Hello Stelios > > Just a gentle follow up if you can share any sample code/repo > > Regards > Kiran > > On Wed, Apr 6, 2022 at 3:19 PM Kiran Biswal wrote: > >> Hello Stelios >> >> Preferred language would have been Scala or pyspark but if Java is proven >> I am open to using it >> >> Any sample reference or example code link? >> >> How are you handling the peotobuf to spark dataframe conversion >> (serialization federalization)? >> >> Thanks >> Kiran >> >> On Wed, Apr 6, 2022, 2:38 PM Stelios Philippou >> wrote: >> >>> Yes we are currently using it as such. >>> Code is in java. Will that work? >>> >>> On Wed, 6 Apr 2022 at 00:51, Kiran Biswal wrote: >>> Hello Experts Has anyone used protobuf (proto3) encoded data (from kafka) as input source and been able to do spark structured streaming? I would appreciate if you can share any sample code/example Regards Kiran >
Re: Unable to format timestamp values in pyspark
Yeah, Stelios. It worked. Could you please post it as an answer so that I can accept it on the post and can be of help to people? Thanks, Sid On Mon, May 30, 2022 at 4:42 PM Stelios Philippou wrote: > Sid, > > According to the error that i am seeing there, this is the Date Format > issue. > > Text '5/1/2019 1:02:16' could not be parsed > > > But your time format is specific as such > > 'M/dd/ H:mm:ss') > > You can see that the day specific is /1/ but your format is dd which > expects two digits. > > Please try the following format and let us know > > 'M/d/ H:mm:ss' > > > > > > On Mon, 30 May 2022 at 11:05, Sid wrote: > >> Hi Team, >> >> I am able to convert to timestamp. However, when I try to filter out the >> records based on a specific value it gives an error as mentioned in the >> post. Could you please help me with this? >> >> >> https://stackoverflow.com/questions/72422897/unable-to-format-timestamp-in-pyspark/72423394#72423394 >> >> >> Best Regards, >> Sid >> >
Re: Unable to format timestamp values in pyspark
Sid, According to the error that i am seeing there, this is the Date Format issue. Text '5/1/2019 1:02:16' could not be parsed But your time format is specific as such 'M/dd/ H:mm:ss') You can see that the day specific is /1/ but your format is dd which expects two digits. Please try the following format and let us know 'M/d/ H:mm:ss' On Mon, 30 May 2022 at 11:05, Sid wrote: > Hi Team, > > I am able to convert to timestamp. However, when I try to filter out the > records based on a specific value it gives an error as mentioned in the > post. Could you please help me with this? > > > https://stackoverflow.com/questions/72422897/unable-to-format-timestamp-in-pyspark/72423394#72423394 > > > Best Regards, > Sid >
Unable to format timestamp values in pyspark
Hi Team, I am able to convert to timestamp. However, when I try to filter out the records based on a specific value it gives an error as mentioned in the post. Could you please help me with this? https://stackoverflow.com/questions/72422897/unable-to-format-timestamp-in-pyspark/72423394#72423394 Best Regards, Sid
Re: Job migrated from EMR to Dataproc takes 20 hours instead of 90 minutes
Thanks. Eventually the problem was solved. I am still not 100% sure what caused it but when I said the input was identical I simplified a bit because it was not (sorry for misleading, I thought this information would just be noise). Explanation: the input to the EMR job was gzips created by Firehose and partitioned hourly. The input to the Dataproc job is gzips created by Kafka Connect and is not partitioned hourly. Otherwise the *content* itself is identical. When we started partitioning the files hourly the problem went away completely. I am still not sure what's going on exactly. If someone has some insight it's welcome. Thanks! On Fri, May 27, 2022 at 9:45 PM Aniket Mokashi wrote: > +cloud-dataproc-discuss > > On Wed, May 25, 2022 at 12:33 AM Ranadip Chatterjee > wrote: > >> To me, it seems like the data being processed on the 2 systems is not >> identical. Can't think of any other reason why the single task stage will >> get a different number of input records in the 2 cases. 700gb of input to a >> single task is not good, and seems to be the bottleneck. >> >> On Wed, 25 May 2022, 06:32 Ori Popowski, wrote: >> >>> Hi, >>> >>> Both jobs use spark.dynamicAllocation.enabled so there's no need to >>> change the number of executors. There are 702 executors in the Dataproc >>> cluster so this is not the problem. >>> About number of partitions - this I didn't change and it's still 400. >>> While writing this now, I am realising that I have more partitions than >>> executors, but the same situation applies to EMR. >>> >>> I am observing 1 task in the final stage also on EMR. The difference is >>> that on EMR that task receives 50K volume of data and on Dataproc it >>> receives 700gb. I don't understand why it's happening. It can mean that the >>> graph is different. But the job is exactly the same. Could it be because >>> the minor version of Spark is different? >>> >>> On Wed, May 25, 2022 at 12:27 AM Ranadip Chatterjee >>> wrote: >>> Hi Ori, A single task for the final step can result from various scenarios like an aggregate operation that results in only 1 value (e.g count) or a key based aggregate with only 1 key for example. There could be other scenarios as well. However, that would be the case in both EMR and Dataproc if the same code is run on the same data in both cases. On a separate note, since you have now changed the size and number of nodes, you may need to re-optimize the number and size of executors for the job and perhaps the number of partitions as well to optimally use the cluster resources. Regards, Ranadip On Tue, 24 May 2022, 10:45 Ori Popowski, wrote: > Hello > > I migrated a job from EMR with Spark 2.4.4 to Dataproc with Spark > 2.4.8. I am creating a cluster with the exact same configuration, where > the > only difference is that the original cluster uses 78 workers with 96 CPUs > and 768GiB memory each, and in the new cluster I am using 117 machines > with > 64 CPUs and 512GiB each, to achieve the same amount of resources in the > cluster. > > The job is run with the same configuration (num of partitions, > parallelism, etc.) and reads the same data. However, something strange > happens and the job takes 20 hours. What I observed is that there is a > stage where the driver instantiates a single task, and this task never > starts because the shuffle of moving all the data to it takes forever. > > I also compared the runtime configuration and found some minor > differences (due to Dataproc being different from EMR) but I haven't found > any substantial difference. > > In other stages the cluster utilizes all the partitions (400), and > it's not clear to me why it decides to invoke a single task. > > Can anyone provide an insight as to why such a thing would happen? > > Thanks > > > - > To unsubscribe e-mail: user-unsubscr...@spark.apache.org > > -- > "...:::Aniket:::... Quetzalco@tl" >
