Very excited to finally post my first question, but please nudge me if I am unclear or violate the standard etiquette. I sincerely appreciate any help that I can get.
I am attempting to use PySpark (in Databricks) to train embeddings for many corpora (each one corresponding with a different author) in parallel. The size of each corpus is under 1GB for each author
The corpi/df are of the form :
+----------------+------------------------------------------------------------+
| author| corpus|
+----------------+------------------------------------------------------------+
| john| [["hello"], ["these", "are", "john's", "thoughts"]] |
| steve| [["hello"], ["these", "are", "steve's", "thoughts"]] |
| markus| [["hello"], ["these", "are", "markus's", "thoughts"]] |
+----------------+------------------------------------------------------------+
Past attempts :
- Defined a UDF which uses
gensim's Word2Vec function applied to each row of above. This worked for small corpi, but I get OutOfMemory error no matter how large I make thespark.executor.memorywhich causes me to lose executors, hang indefinitely, etc... - Now, I'm trying to
explodethe above dataframe to where each row is a sentence. Then I was thinking that I would do adf.groupBy('author')and I would define a UDF which implements Spark'sWord2Vecmodel to train on the groups and save the model. Unfortunately, even when only using a few authors, repartitioning, etc..., it won't let me explode the dataframe without hanging indefinitely, so I'm stuck with the dataframe that I showed above.
Cluster :
- 1 driver = 56GB Memory, 16 cores
- 1 - 8 workers (autoscaled) 56GB Memory, 16 cores
Configuration (inspired by here) :
yarn.nodemanager.pmem-check-enabled false
spark.databricks.delta.preview.enabled true
spark.executor.cores 5
spark.executor.memory 16gb
spark.executor.instances 26
spark.driver.memory 16gb
spark.yarn.executor.memoryOverhead 2gb
yarn.nodemanager.vmem-check-enabled false
spark.default.parallelism 260
spark.driver.cores 5
Performance :
Error on Executor :
WARN HangingTaskDetector: Task 1527 is probably not making progress because its metrics (Map(internal.metrics.shuffle.read.localBlocksFetched -> 0, internal.metrics.shuffle.read.remoteBytesReadToDisk -> 0, internal.metrics.shuffle.write.bytesWritten -> 0, internal.metrics.output.recordsWritten -> 0, internal.metrics.shuffle.write.recordsWritten -> 0, internal.metrics.memoryBytesSpilled -> 0, internal.metrics.shuffle.read.remoteBytesRead -> 0, internal.metrics.diskBytesSpilled -> 0, internal.metrics.shuffle.read.localBytesRead -> 0, internal.metrics.shuffle.read.recordsRead -> 0, internal.metrics.output.bytesWritten -> 0, internal.metrics.input.bytesRead -> 119650145, internal.metrics.input.recordsRead -> 4096, internal.metrics.shuffle.read.remoteBlocksFetched -> 0))
Driver stdout Log :
2021-07-29T02:27:25.955+0000: [GC (Allocation Failure) [PSYoungGen: 5309440K->135150K(5444608K)] 5520626K->401147K(16629760K), 0.0788413 secs] [Times: user=0.22 sys=0.14, real=0.08 secs]
Log :
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_29_piece0 on 10.161.179.16:42153 in memory (size: 7.4 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_2_piece0 on 10.161.179.6:40551 in memory (size: 28.6 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_7_piece0 on 10.161.179.19:38301 in memory (size: 14.2 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_37_piece0 on 10.161.179.19:36657 in memory (size: 13.4 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_7_piece0 on 10.161.179.19:36657 in memory (size: 14.2 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_31_piece0 on 10.161.179.10:45589 in memory (size: 11.5 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_35_piece0 on 10.161.179.10:40843 in memory (size: 8.3 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_7_piece0 on 10.161.179.16:42707 in memory (size: 14.2 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_16_piece0 on 10.161.179.19:38301 in memory (size: 8.0 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_16_piece0 on 10.161.179.19:36657 in memory (size: 8.0 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_16_piece0 on 10.161.179.6:37313 in memory (size: 8.0 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_7_piece0 on 10.161.179.6:37313 in memory (size: 14.2 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_8_piece0 on 10.161.179.6:37313 in memory (size: 62.5 KiB, free: 8.4 GiB)
21/07/29 02:27:27 INFO BlockManagerInfo: Removed broadcast_37_piece0 on 10.161.179.8:41599 in memory (size: 13.4 KiB, free: 8.4 GiB)
21/07/29 02:29:16 INFO DriverCorral: DBFS health check ok
21/07/29 02:29:17 INFO HikariDataSource: metastore-monitor - Starting...
21/07/29 02:29:17 INFO HikariDataSource: metastore-monitor - Start completed.
21/07/29 02:29:17 INFO HikariDataSource: metastore-monitor - Shutdown initiated...
21/07/29 02:29:17 INFO HikariDataSource: metastore-monitor - Shutdown completed.
21/07/29 02:29:17 INFO MetastoreMonitor: Metastore healthcheck successful (connection duration = 84 milliseconds)
21/07/29 02:29:25 INFO HiveMetaStore: 1: get_database: default
21/07/29 02:29:25 INFO audit: ugi=root ip=unknown-ip-addr cmd=get_database: default
21/07/29 02:29:25 INFO DriverCorral: Metastore health check ok
So my question is twofold,
- Am I doing something wrong with simply trying to explode the rows and then train the Word2Vec embeddings with a UDF applied to
groupBy.('author')? I understand thatexplodeis very expensive since it is creating an enormous amount of rows, but it seemed like the best option. - Is there a more efficient/different way to approach this that avoids the
explodehangup altogether?
Please let me know, if I can provide any other information that may help, and I welcome any advice which will help me ask questions more clearly in the future.
Update:
After changing the spark.sql.shuffle.partitions to a large number, things certainly improved, but while the executors are completing tasks, they are continuously getting created and destroyed because of "worker lost" or "Remote RPC client disassociated. Likely..." (which has plagued me in the past despite changing to many different driver/executor memory/core configurations). The event timeline looks like this:
Therefore, my above question still holds, but finding the proper configuration seems one step closer.