# Running Hail on Terra -- how should I optimize? **URL:** \ **Category:** Hail Query & hailctl\ **Created:** [January 13, 2021, 10:31pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889 "2021-01-13T22:31:12Z")\ **Posts on this page:** 8\ **Page:** 1
**Author:** ![aofarrel](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/aofarrel/32/530_2.png) [@aofarrel](https://discuss.hail.is/u/aofarrel)\ **Post date:** [January 13, 2021, 10:31pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/1 "2021-01-13T22:31:12Z")
I’m maintaining a Hail-related template notebook in the Broad’s Terra environment in a Jupyter notebook context on a Spark cluster as part of one the Biodata Catalyst grant. I am wondering what the best way to optimize for this kind of environment. Previously, we told users to use the following configuration when running on Freeze 5b data, n=1000: Master node: 8 CPUs, 30 GB mem, 500 GB disk space Workers: 170 workers of which 50 are preemptible, each having 8 CPUs, 30 GB mem, and 100 GB disk space. Now we’re running on Freeze 8 data, which has a truckload more variants. I am testing on a n=1111 cohort of genetically similar individuals, but likely due to the change to Freeze 8 the previous configuration appears inadequate. If I run only on chr 1 then I can complete the notebook with that compute in just under 50 minutes, but on all chromosomes, it appears to be taking too long or a preemptible is getting stuck. Avoiding preemptibles is an option, but it doesn’t fix the underlying issue – if at all possible we would like the notebook to be completable by users in a few hours tops for n=1000ish on Freeze 8 data, because compute will continue costing money after completion of all tasks in the notebook. Terra has a feature where it will “pause” the compute after about half an hour of no activity, but once you’re requesting this amount of resources, you are still spending about $20/hour when the compute is paused. In other words it can’t take so long that the user goes to bed and wakes up to a bunch of unnecessary extra charges. The specific tasks being performed are related to a GWAS. The GWAS itself is performed outside of the notebook, this is the preparation. Essentially, the tasks are: - Loading in VCF data and creating a hail matrix (mt) - mt.count() - Merging the hail table with a pandas dataframe containing phenotypic data - Filtering for common variants (MAF \> 0.01) via variant\_qc() - **hl.ld\_prune()**, which tends to take the longest - Perform a normalized PCA - Calculate a GRM I am already in touch with several Terra engineers but I was recommended to take this question to the Hail community, as I have heard the setup of Terra is not entirely unlike common setups for Hail. ---
**Author:** ![danking](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/danking/32/43_2.png) [@danking](https://discuss.hail.is/u/danking)\ **Post date:** [January 14, 2021, 10:10pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/2 "2021-01-14T22:10:10Z")
Hey @aofarrel ! Sorry you’re having performance headaches and high cost! I’m sure we can address this. Can you point us at the notebook? It’s easier for me to make suggestions on code. * * * Generally speaking, you don’t want to avoid preemptibles. You actually want as many preemptibles as possible for as long as possible! Hail power users often start a cluster, submit a Hail pipeline that does all the preemptible-safe computation and writes out the result, then re-configure the cluster for a subsequent pipeline that is not preemptible-safe. I wrote up some general notes [here](https://github.com/danking/hail-cloud-docs/blob/master/how-to-cloud-with-hail.md), but I’m happy to make specific comments on the notebook in question. * * * Looking quickly at the tasks, I’d structure this as: 1. Import VCF write to MT. We’ll never touch the VCF again. 2. Load MT, annotate with phenotypic data, calculate statistics, and write out just the rows of this table (much less data!). For example, ```auto mt = mt.filter_rows(passes_qc = (mt.variant_qc.AF[0] > 0.2) & (mt.call_rate > 0.9) & ...) rows = mt.rows() rows = rows.select() # keep just the locus and alleles (the key fields) rows.write('gs://.../qc_passing_variants') # the write triggers all the QC calculations ``` 1. Now, for each of ld\_prune, PCA, and GRM, I’d tune the cluster size, number of partitions, and preemptible to non-preemptible ratio to be successful for that operation. For each operation, I would freshly load and filter the matrix table. Once I have a set of ld\_pruned variants, I’d write that out as another row table, to be used for filtering. I’m pretty sure GRM is preemptible-safe if you set the hail temporary directory to a GCS directory (e.g. `hl.init(tmp_dir='gs://my-temporary-bucket/')`). `ld_prune` is a notoriously difficult operation. I’ll ask someone with more background to comment here about that operation. * * * Yeah, babysitting pipelines is a pain. I’d ask gnomAD team how they deal with this. I’m sure they have good tactics. I personally run expensive jobs in a script where the cluster shutdown runs immediately after the submit finished—successful or not. Is there a way to replicate this behavior in Terra? ```auto gcloud dataproc create ... gcloud dataproc submit ... gcloud dataproc stop ... ``` ---
**Author:** ![aofarrel](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/aofarrel/32/530_2.png) [@aofarrel](https://discuss.hail.is/u/aofarrel)\ **Post date:** [January 19, 2021, 8:35pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/3 "2021-01-19T20:35:29Z")
Thanks for your quick response. The notebook in question can be found [here in herzog format](https://github.com/DataBiosphere/bdcat_notebooks/blob/master/notebooks/GWAS_blood_pressure_p3/main.py). Herzog just allows us to represent the notebook in .py format instead of .ipynb – everything, including the markdown parts of the notebook, are in there. Anything outside of a `with herzog.Cell` block is not in the final rendered notebook. If you have access to Terra, you can see in in iPython format [here in this template workspace](https://terra.biodatacatalyst.nhlbi.nih.gov/#workspaces/biodata-catalyst/BioData%20Catalyst%20GWAS%20blood%20pressure%20trait/notebooks/launch/3-GWAS-genomic-data-preparation.ipynb). (The notebook says it is restricted access, but what I’m linking you is an empty template without data.) Due to the nature of the notebook compute, the contents of a single notebook shares the same resources. Multiple notebooks can share resources, or these can create new computes, but there is no way to change a compute on the fly – if you decide you want more memory, you have to spin up an entirely new compute, clearing all variables in the process. So, if the GRM is in the same notebook as ld\_prune (which is currently the case) then they share the same number of preemptibles, number of non-preempt workers, etc. I can split the notebooks if fine-tuning for each operation is really necessary, but since the process requires saving everything that was done so far, deleting a compute, creating a new one, spinning up the new one, re-installing hail and its dependencies in the new compute, and reloading everything necessary from the previous notebook, that also has a time loss – albiet one that tends to be in the range of about ten minutes, so if the time gain for carefully adjust the compute between these operations is large, I’m all for splitting the notebook. As for tuning the cluster size and the ratio, that’s what I’m looking for specific advice on, especially on ld\_prune. I’ve been testing several possibilities but so far have not seen any real gains on efficiency on this data, other than going from 120 non-preempt workers to 120 non-prempts+120 prempts, which reduced time on chr1 n=1111 by about just under five minutes but still taking more than 40 minutes total. I’m hesitant to keep running expensive tests with ever-increasing numbers of resources if there is knowledge out there as to what I should be aiming for. As far as I know the cluster cannot be shut down and deleted from the notebook directly. The reason why the deletion is necessary (once the output has been saved to the separate google bucket that the rest of the workspace operates in) is that cost remains about 1/3 of what it does when active. ie, if an active compute costs about $60/hour, when paused it will cost roughly $20/hour. The user can delete the compute from the UI relatively easily, but it can’t be set to be deleted after the notebook runs, which is why I’m worried about users who say “this is taking forever, let me run it overnight” being on the hook for unnecessary costs. ---
**Author:** ![danking](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/danking/32/43_2.png) [@danking](https://discuss.hail.is/u/danking)\ **Post date:** [January 20, 2021, 2:04am UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/4 "2021-01-20T02:04:01Z")
I’ll just comment on the notebook in particular in this response. I’ve pinged the Hail-compiler-team to come comment on ld\_prune. 1. I’d avoid `.show` until you’ve written the matrix table. Hail doesn’t have smart VCF indexing strategies, so that `show(5)` is likely triggering more computation than anyone wants. Generally, we try to write immediately after importing vcf. Including the variant\_qc and sample metadata in the MT itself is reasonable though, so I’d maybe just save the `show` for after the write. 2. How many partitions does the written MT have? It looks like you switch to 25 partitions for the export and for ld prune. 3. Oof, `BlockMatrix.to_numpy` is a really painful operation. It looks like you’re using that to then convert the GRM into a pandas table and output a comma separated table of text values. Most people who use the GRM in Hail do everything related to the GRM in Hail using BlockMatrix. At the very least, I’d recommend writing and reading the GRM BlockMatrix before calling `to_numpy`. I think the GRM is almost exclusively used as an input to Hail’s LMM module. Hmm. Unfortunately, this notebook uses Hail in a way that doesn’t take advantage of its efficient data representations. Converting from one text format to another text format is kind of the worst-case scenario for Hail, we spend a lot of time converting to our internal binary formats only to leave them again. I think a more compelling example would use `hl.linear_regression_rows` as we do in our [tutorial](https://hail.is/docs/0.2/tutorials/01-genome-wide-association-study.html). As I said, I’ll have someone comment on ld\_prune. I’m curious if you see a performance improvement on the GRM by writing/reading before `to_numpy`! ---
**Author:** ![aofarrel](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/aofarrel/32/530_2.png) [@aofarrel](https://discuss.hail.is/u/aofarrel)\ **Post date:** [January 22, 2021, 11:43pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/5 "2021-01-22T23:43:34Z")
The rationale for .show() was more for so tutorial purposes. On the current test data set (aformentioned freeze 8, chr1 only, n=1111), each .show() only takes 7 seconds, so it’s not a concern to us compared to other operations. Nevertheless we can make it optional. We set the minimum number of partitions to 200 by default. print(mt.n\_partitions()) returns 200 after the import, so it seems to be going with the minimum. Is that expected behavior? Should we increase the minimum? Thanks for your advice on the GRM – removing the numpy operation has saved several minutes. It went from nine minutes to just twenty seconds! ---
**Author:** ![danking](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/danking/32/43_2.png) [@danking](https://discuss.hail.is/u/danking)\ **Post date:** [January 22, 2021, 11:49pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/6 "2021-01-22T23:49:47Z")
Ah great! I think hail aims for 128MB partitions by default as a good balance of memory per core and speed. If it’s going with the minimum, then that’s probably fine. I’m pleasantly surprised to learn hail is so snappy on the VCF data! Regarding ld\_prune, the folks who know about that stuff are a bit busy at the moment so they can’t comment. Am I reading this code right that you repartition down to 25 partitions before calling ld\_prune? I’d try running ld\_prune on the data without any repartitioning. It’s possible there’s some bad interaction there. ---
**Author:** ![aofarrel](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/aofarrel/32/530_2.png) [@aofarrel](https://discuss.hail.is/u/aofarrel)\ **Post date:** [January 28, 2021, 9:05pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/7 "2021-01-28T21:05:33Z")
We do indeed repartition to 25 as it seems to affect the writing of VCFs. Oddly enough, when I remove that repartitioning, ld\_prune() seems to run much faster. However, it’s not something I can implement at the moment due to persistent errors. In the context of ld\_prune() there is a correlation between the number of partitions and the number of Spark workers (and block\_size?). It’s putting together that relationship that I’m struggling with, which is making adjustments to partitioning to improve performance quite difficult. This notebook configuration will fail on ld\_prune(): - 2 workers + 200 preemptibles - each worker gets 4 CPUs, 15 GB mem, and 50 GB disk This notebook configuration will complete successfully: - 2 workers + 240 preemptibles - each worker gets 8 CPUs, 30 GB mem, and 50 GB disk Unless I use that configration on more data. This whole time I’ve been testing on chr1, but if I expand to all chromosomes, LD\_prune() fails. I have a rough understanding of how ld\_prune() operates within and outside given partitions in a three step process. But it’s not clear to me what the actual correlation between workers/cores, number of partitions, and block\_size works. I saw this on ld\_prune()’s documentation… > The parallelism of writing the locally-pruned matrix table as a block matrix is `n_locally_pruned_variants / block_size` . …but I’m not sure how to make adjustments based on that. Should I be increasing block size? Adding more workers? Or am I barking up the wrong tree? * * * Here’s a stacktrace of the error: > [@](#): > > FatalError Traceback (most recent call last) > /usr/local/lib/python3.7/dist-packages/IPython/core/formatters.py in **call** (self, obj) > 700 type\_pprinters=self.type\_printers, > 701 deferred\_pprinters=self.deferred\_printers) > → 702 printer.pretty(obj) > 703 printer.flush() > 704 return stream.getvalue() > > /usr/local/lib/python3.7/dist-packages/IPython/lib/pretty.py in pretty(self, obj) > 392 if cls is not object > 393 and callable(cls. **dict**.get(‘ **repr** ’)): > → 394 return \_repr\_pprint(obj, self, cycle) > 395 > 396 return \_default\_pprint(obj, self, cycle) > > /usr/local/lib/python3.7/dist-packages/IPython/lib/pretty.py in _repr\_pprint(obj, p, cycle) > 698 “”“A pprint that just redirects to the normal repr function.”“” > 699 # Find newlines and replace them with p.break_() > → 700 output = repr(obj) > 701 lines = output.splitlines() > 702 with p.group(): > > /usr/local/lib/python3.7/dist-packages/hail/table.py in **repr** (self) > 1295 > 1296 def **repr** (self): > → 1297 return self. **str** () > 1298 > 1299 def data(self): > > /usr/local/lib/python3.7/dist-packages/hail/table.py in **str** (self) > 1292 > 1293 def **str** (self): > → 1294 return self.\_ascii\_str() > 1295 > 1296 def **repr** (self): > > /usr/local/lib/python3.7/dist-packages/hail/table.py in \_ascii\_str(self) > 1318 return s > 1319 > → 1320 rows, has\_more, dtype = self.data() > 1321 fields = list(dtype) > 1322 trunc\_fields = [trunc(f) for f in fields] > > /usr/local/lib/python3.7/dist-packages/hail/table.py in data(self) > 1302 row\_dtype = t.row.dtype > 1303 t = t.select(\*\*{k: hl.\_showstr(v) for (k, v) in t.row.items()}) > → 1304 rows, has\_more = t.\_take\_n(self.n) > 1305 self.\_data = (rows, has\_more, row\_dtype) > 1306 return self.\_data > > /usr/local/lib/python3.7/dist-packages/hail/table.py in \_take\_n(self, n) > 1449 has\_more = False > 1450 else: > → 1451 rows = self.take(n + 1) > 1452 has\_more = len(rows) \> n > 1453 rows = rows[:n] > > in take(self, n, \_localize) > > /usr/local/lib/python3.7/dist-packages/hail/typecheck/check.py in wrapper(\_\_original\_func, \*args, \*\*kwargs) > 612 def wrapper(_original\_func, \*args, \*\*kwargs): > 613 args, kwargs_ = check\_all(\_\_original\_func, args, kwargs, checkers, is\_method=is\_method) > → 614 return _original\_func(\*args, \*\*kwargs_) > 615 > 616 return wrapper > > /usr/local/lib/python3.7/dist-packages/hail/table.py in take(self, n, \_localize) > 2119 “”" > 2120 > → 2121 return self.head(n).collect(\_localize) > 2122 > 2123 @typecheck\_method(n=int) > > in collect(self, \_localize) > > /usr/local/lib/python3.7/dist-packages/hail/typecheck/check.py in wrapper(\_\_original\_func, \*args, \*\*kwargs) > 612 def wrapper(_original\_func, \*args, \*\*kwargs): > 613 args, kwargs_ = check\_all(\_\_original\_func, args, kwargs, checkers, is\_method=is\_method) > → 614 return _original\_func(\*args, \*\*kwargs_) > 615 > 616 return wrapper > > /usr/local/lib/python3.7/dist-packages/hail/table.py in collect(self, \_localize) > 1918 e = construct\_expr(rows\_ir, hl.tarray(t.row.dtype)) > 1919 if \_localize: > → 1920 return Env.backend().execute(e.\_ir) > 1921 else: > 1922 return e > > /usr/local/lib/python3.7/dist-packages/hail/backend/py4j\_backend.py in execute(self, ir, timed) > 96 raise HailUserError(message\_and\_trace) from None > 97 > —\> 98 raise e > > /usr/local/lib/python3.7/dist-packages/hail/backend/py4j\_backend.py in execute(self, ir, timed) > 72 # print(self.\_hail\_package.expr.ir.Pretty.apply(jir, True, -1)) > 73 try: > —\> 74 result = json.loads(self.\_jhc.backend().executeJSON(jir)) > 75 value = ir.typ.\_from\_json(result[‘value’]) > 76 timings = result[‘timings’] > > /usr/local/lib/python3.7/dist-packages/py4j/java\_gateway.py in **call** (self, \*args) > 1255 answer = self.gateway\_client.send\_command(command) > 1256 return\_value = get\_return\_value( > → 1257 answer, self.gateway\_client, self.target\_id, self.name) > 1258 > 1259 for temp\_arg in temp\_args: > > /usr/local/lib/python3.7/dist-packages/hail/backend/py4j\_backend.py in deco(\*args, \*\*kwargs) > 30 raise FatalError(‘%s\n\nJava stack trace:\n%s\n’ > 31 ‘Hail version: %s\n’ > —\> 32 ‘Error summary: %s’ % (deepest, full, hail. **version** , deepest), error\_id) from None > 33 except pyspark.sql.utils.CapturedException as e: > 34 raise FatalError(‘%s\n\nJava stack trace:\n%s\n’ > > FatalError: SparkException: Job aborted due to stage failure: Task 488 in stage 5.0 failed 4 times, most recent failure: Lost task 488.3 in stage 5.0 (TID 3904, saturn-521657d0-b18f-4cba-8b1b-badecf79009d-sw-w09z.c.biodata-catalyst.internal, executor 58): ExecutorLostFailure (executor 58 exited caused by one of the running tasks) Reason: Slave lost > Driver stacktrace: > > Java stack trace: > org.apache.spark.SparkException: Job aborted due to stage failure: Task 488 in stage 5.0 failed 4 times, most recent failure: Lost task 488.3 in stage 5.0 (TID 3904, saturn-521657d0-b18f-4cba-8b1b-badecf79009d-sw-w09z.c.biodata-catalyst.internal, executor 58): ExecutorLostFailure (executor 58 exited caused by one of the running tasks) Reason: Slave lost > Driver stacktrace: > at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1926) > at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1914) > at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1913) > at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) > at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) > at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1913) > at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:948) > at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:948) > at scala.Option.foreach(Option.scala:257) > at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:948) > at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2147) > at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2096) > at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2085) > at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) > at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:759) > at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061) > at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082) > at org.apache.spark.SparkContext.runJob(SparkContext.scala:2114) > at is.hail.sparkextras.ContextRDD.crunJobWithIndex(ContextRDD.scala:228) > at is.hail.rvd.RVD$.getKeyInfo(RVD.scala:1218) > at is.hail.rvd.RVD$.makeCoercer(RVD.scala:1293) > at is.hail.expr.ir.GenericTableValue.getRVDCoercer(GenericTableValue.scala:162) > at is.hail.expr.ir.GenericTableValue.toTableValue(GenericTableValue.scala:188) > at is.hail.io.vcf.MatrixVCFReader.apply(LoadVCF.scala:1773) > at is.hail.expr.ir.TableRead.execute(TableIR.scala:1100) > at is.hail.expr.ir.TableOrderBy.execute(TableIR.scala:2629) > at is.hail.expr.ir.TableSubset$class.execute(TableIR.scala:1324) > at is.hail.expr.ir.TableHead.execute(TableIR.scala:1332) > at is.hail.expr.ir.TableMapRows.execute(TableIR.scala:1845) > at is.hail.expr.ir.Interpret$.run(Interpret.scala:819) > at is.hail.expr.ir.Interpret$.alreadyLowered(Interpret.scala:53) > at is.hail.expr.ir.InterpretNonCompilable$.interpretAndCoerce$1(InterpretNonCompilable.scala:16) > at is.hail.expr.ir.InterpretNonCompilable$.is$hail$expr$ir$InterpretNonCompilable$$rewrite$1(InterpretNonCompilable.scala:53) > at is.hail.expr.ir.InterpretNonCompilable$$anonfun$1.apply(InterpretNonCompilable.scala:25) > at is.hail.expr.ir.InterpretNonCompilable$$anonfun$1.apply(InterpretNonCompilable.scala:25) > at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) > at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) > at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) > at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35) > at scala.collection.TraversableLike$class.map(TraversableLike.scala:234) > at scala.collection.AbstractTraversable.map(Traversable.scala:104) > at is.hail.expr.ir.InterpretNonCompilable$.rewriteChildren$1(InterpretNonCompilable.scala:25) > at is.hail.expr.ir.InterpretNonCompilable$.is$hail$expr$ir$InterpretNonCompilable$$rewrite$1(InterpretNonCompilable.scala:54) > at is.hail.expr.ir.InterpretNonCompilable$.apply(InterpretNonCompilable.scala:58) > at is.hail.expr.ir.lowering.InterpretNonCompilablePass$.transform(LoweringPass.scala:67) > at is.hail.expr.ir.lowering.LoweringPass$$anonfun$apply$3$$anonfun$1.apply(LoweringPass.scala:15) > at is.hail.expr.ir.lowering.LoweringPass$$anonfun$apply$3$$anonfun$1.apply(LoweringPass.scala:15) > at is.hail.utils.ExecutionTimer.time(ExecutionTimer.scala:81) > at is.hail.expr.ir.lowering.LoweringPass$$anonfun$apply$3.apply(LoweringPass.scala:15) > at is.hail.expr.ir.lowering.LoweringPass$$anonfun$apply$3.apply(LoweringPass.scala:13) > at is.hail.utils.ExecutionTimer.time(ExecutionTimer.scala:81) > at is.hail.expr.ir.lowering.LoweringPass$class.apply(LoweringPass.scala:13) > at is.hail.expr.ir.lowering.InterpretNonCompilablePass$.apply(LoweringPass.scala:62) > at is.hail.expr.ir.lowering.LoweringPipeline$$anonfun$apply$1.apply(LoweringPipeline.scala:14) > at is.hail.expr.ir.lowering.LoweringPipeline$$anonfun$apply$1.apply(LoweringPipeline.scala:12) > at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) > at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35) > at is.hail.expr.ir.lowering.LoweringPipeline.apply(LoweringPipeline.scala:12) > at is.hail.expr.ir.CompileAndEvaluate$.\_apply(CompileAndEvaluate.scala:28) > at is.hail.backend.spark.SparkBackend.is$hail$backend$spark$SparkBackend$$\_execute(SparkBackend.scala:354) > at is.hail.backend.spark.SparkBackend$$anonfun$execute$1.apply(SparkBackend.scala:338) > at is.hail.backend.spark.SparkBackend$$anonfun$execute$1.apply(SparkBackend.scala:335) > at is.hail.expr.ir.ExecuteContext$$anonfun$scoped$1.apply(ExecuteContext.scala:25) > at is.hail.expr.ir.ExecuteContext$$anonfun$scoped$1.apply(ExecuteContext.scala:23) > at is.hail.utils.package$.using(package.scala:618) > at is.hail.annotations.Region$.scoped(Region.scala:18) > at is.hail.expr.ir.ExecuteContext$.scoped(ExecuteContext.scala:23) > at is.hail.backend.spark.SparkBackend.withExecuteContext(SparkBackend.scala:247) > at is.hail.backend.spark.SparkBackend.execute(SparkBackend.scala:335) > at is.hail.backend.spark.SparkBackend$$anonfun$7.apply(SparkBackend.scala:379) > at is.hail.backend.spark.SparkBackend$$anonfun$7.apply(SparkBackend.scala:377) > at is.hail.utils.ExecutionTimer$.time(ExecutionTimer.scala:52) > at is.hail.backend.spark.SparkBackend.executeJSON(SparkBackend.scala:377) > 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 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) > at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) > at py4j.Gateway.invoke(Gateway.java:282) > at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) > at py4j.commands.CallCommand.execute(CallCommand.java:79) > at py4j.GatewayConnection.run(GatewayConnection.java:238) > at java.lang.Thread.run(Thread.java:748) > > Hail version: 0.2.61-3c86d3ba497a > Error summary: SparkException: Job aborted due to stage failure: Task 488 in stage 5.0 failed 4 times, most recent failure: Lost task 488.3 in stage 5.0 (TID 3904, saturn-521657d0-b18f-4cba-8b1b-badecf79009d-sw-w09z.c.biodata-catalyst.internal, executor 58): ExecutorLostFailure (executor 58 exited caused by one of the running tasks) Reason: Slave lost > Driver stacktrace: > 2021-01-28 18:33:37 Hail: INFO: Coerced sorted dataset > 2021-01-28 18:33:37 Hail: INFO: Coerced dataset with out-of-order partitions. ---
**Author:** ![tpoterba](https://yyz2.discourse-cdn.com/flex036/user_avatar/discuss.hail.is/tpoterba/32/61_2.png) [@tpoterba](https://discuss.hail.is/u/tpoterba)\ **Post date:** [February 3, 2021, 11:06pm UTC](https://discuss.hail.is/t/running-hail-on-terra-how-should-i-optimize/1889/8 "2021-02-03T23:06:49Z")
Sorry for the high-latency responses. This is something we need to replicate and debug internally. To clarify: You have WGS data, currently 1111 samples. This is roughly the pipeline failing: 1. load vcf into MT 2. annotate cols with some phenotype data 3. run variant\_qc 4. filter rows to MAF \> 0.01 5. LD prune I do expect LD prune to be somewhat sensitive to partitioning – we do a local in-partition prune, then write and do a matrix multiply and run a graph independence algorithm for the rest. > [@aofarrel](#): > > In the context of ld\_prune() there is a correlation between the number of partitions and the number of Spark workers (and block\_size?). It’s putting together that relationship that I’m struggling with, which is making adjustments to partitioning to improve performance quite difficult. In many Hail pipelines, we spawn a Spark job with one task per partition of your [matrix]table. A Spark task runs on a worker (multiple tasks may run on a worker, the default configurations allow one task per core of a 4-core worker JVM). However, the matrix operations do change partitioning here since we need to create a distributed block matrix with evenly-sized blocks. That’s the bit I’m most worried about, and most interested in replicating. I’m not sure what’s causing the crashes, though, and need to investigate further. Also, one aside on your comment in the original post: > [@aofarrel](#): > > Terra has a feature where it will “pause” the compute after about half an hour of no activity, but once you’re requesting this amount of resources, you are still spending about $20/hour when the compute is paused. In other words it can’t take so long that the user goes to bed and wakes up to a bunch of unnecessary extra charges. Using an autoscaling configuration in Google Dataproc should let your cluster scale up when there are jobs to run and scale down when there is idle time. This is a pretty effective way to use large clusters with high utilization (low $$$ waste).