Docker container job abend issue "The main method caused an error: Pipeline execution failed" then Caused by shows heap space error

[heschmidt04]

Could use some help to debug around this and figure out the right balance of settings for the Docker container. Jared then can document it and share with CHoRUS/Bridge2AI.
Thanks!

Tagging @qqndrew proactively.

Looks like it might be a Java Heap error?
What are ideal Java settings for processing 4G worth of notes in approx 2.1k individual files?

Meantime, I'll try increasing Flink conf memory for job mgr and task mgr in the meantime and see if I get past the issue.

Main error: "The main method caused an error: Pipeline execution failed"
Caused by: org.apache.beam.sdk.util.UserCodeException: java.lang.OutOfMemoryError: Java heap space

Steps taken to run the run_pipeline_local.sh on non de-identified notes.csv to get the tagging information for the note_nlp table.

• Built local image from Dockerfile - Packages listed below

https://github.com/OHNLP/Backbone/releases/download/v3.0.28/Backbone.zip
https://dlcdn.apache.org/flink/flink-1.18.1/flink-1.18.1-bin-scala_2.12.tgz
https://github.com/OHNLP/MedTagger/releases/download/v1.0.81/MedTaggerForBackbone.zip

• Java heap and version settings in Docker container

root@e14bb1607b87:/BackboneRoot# java -XX:+PrintFlagsFinal -version | grep HeapSize
size_t ErgoHeapSizeLimit = 0 {product} {default}
size_t HeapSizePerGCThread = 43620760 {product} {default}
size_t InitialHeapSize = 130023424 {product} {ergonomic}
size_t LargePageHeapSizeThreshold = 134217728 {product} {default}
size_t MaxHeapSize = 2057306112 {product} {ergonomic}
size_t MinHeapSize = 8388608 {product} {ergonomic}
uintx NonNMethodCodeHeapSize = 5839564 {pd product} {ergonomic}
uintx NonProfiledCodeHeapSize = 122909338 {pd product} {ergonomic}
uintx ProfiledCodeHeapSize = 122909338 {pd product} {ergonomic}
size_t SoftMaxHeapSize = 2057306112 {manageable} {ergonomic}
openjdk version "21.0.5" 2024-10-15
OpenJDK Runtime Environment (build 21.0.5+11-Ubuntu-1ubuntu124.04)
OpenJDK 64-Bit Server VM (build 21.0.5+11-Ubuntu-1ubuntu124.04, mixed mode, sharing)

• Copied run_filesystem_csv.json into Docker container in config directory
• Parsed the Progress Notes 4G notes.csv into individual files in /BackboneRoot/source with each note id being the filename.csv with no headers just comma separated values with double quotes around the note_text column data.

cat 500000025.csv # contents of the file with all the note OMOP table columns
500000025,123,1111-01-11,1111-01-11 21:08:00.000,32817,3000735,Progress Note Type,"Lots of text in between these quotes",32678,4180186,160801,269966,,Progress Note Type,,

• Started the /Backbone/run_pipeline_local.sh shell file after chmod 0755 on the /Backbone/run_pipeline_local.sh file to parse the source directory files into the output - 1 empty part*.csv created before failure

Changes made to /BackboneRoot/run_pipeline_local.sh

FLINK_DIR=flink-1.18.1/bin

Changes made to /BackboneRoot/flink-1.18.1/conf/flink-conf.yaml

jobmanager.memory.process.size: 3600m
taskmanager.memory.process.size: 3600m

The Flink job fails with the following

Starting taskexecutor daemon on host e14bb1607b87.
Flink Cluster Started - Job Progress Can be Seen via Configured WebUI Port (Default: localhost:8081)
Submitting Job...
Job has been submitted with JobID f3890a3b9f5d8b420528de15fe2e36ed

---

The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Pipeline execution failed
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:851)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:245)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1095)
at org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1189)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
at org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1189)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1157)
Caused by: java.lang.RuntimeException: Pipeline execution failed
at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:112)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:321)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:307)
at org.ohnlp.backbone.core.BackboneRunner.main(BackboneRunner.java:24)
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
... 9 more
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: f3890a3b9f5d8b420528de15fe2e36ed)
at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
at org.apache.flink.client.program.ContextEnvironment.getJobExecutionResult(ContextEnvironment.java:126)
at org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:77)
at org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.executePipeline(FlinkPipelineExecutionEnvironment.java:134)
at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:109)
... 15 more
Caused by: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: f3890a3b9f5d8b420528de15fe2e36ed)
at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:130)
at java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:646)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)
at org.apache.flink.util.concurrent.FutureUtils.lambda$retryOperationWithDelay$6(FutureUtils.java:302)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)
at org.apache.flink.client.program.rest.RestClusterClient.lambda$pollResourceAsync$33(RestClusterClient.java:794)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)
at org.apache.flink.util.concurrent.FutureUtils.lambda$retryOperationWithDelay$6(FutureUtils.java:302)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFuture.postFire(CompletableFuture.java:614)
at java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1163)
at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:128)
... 23 more
Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:176)
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:107)
at org.apache.flink.runtime.scheduler.DefaultScheduler.recordTaskFailure(DefaultScheduler.java:285)
at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:276)
at org.apache.flink.runtime.scheduler.DefaultScheduler.onTaskFailed(DefaultScheduler.java:269)
at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:764)
at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:741)
at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83)
at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:488)
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:309)
at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:307)
at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:222)
at org.apache.flink.runtime.rpc.pekko.FencedPekkoRpcActor.handleRpcMessage(FencedPekkoRpcActor.java:85)
at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleMessage(PekkoRpcActor.java:168)
at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:33)
at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:29)
at scala.PartialFunction.applyOrElse(PartialFunction.scala:127)
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126)
at org.apache.pekko.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:29)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
at org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547)
at org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545)
at org.apache.pekko.actor.AbstractActor.aroundReceive(AbstractActor.scala:229)
at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590)
at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557)
at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:280)
at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:241)
at org.apache.pekko.dispatch.Mailbox.exec(Mailbox.scala:253)
at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:387)
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1312)
at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1843)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1808)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:188)
Caused by: org.apache.beam.sdk.util.UserCodeException: java.lang.OutOfMemoryError: Java heap space
at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
at org.apache.beam.sdk.io.FileIO$MatchAll$MatchFn$DoFnInvoker.invokeProcessElement(Unknown Source)
at org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:211)
at org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:188)
at org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement(DoFnRunnerWithMetricsUpdate.java:62)
at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction.flatMap(FlinkDoFnFunction.java:122)
at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction.flatMap(FlinkDoFnFunction.java:59)
at org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:79)
at org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:204)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: java.lang.OutOfMemoryError: Java heap space
at java.base/java.lang.AbstractStringBuilder.(AbstractStringBuilder.java:101)
at java.base/java.lang.StringBuilder.(StringBuilder.java:106)
at org.apache.beam.sdk.io.LocalResourceId.(LocalResourceId.java:49)
at org.apache.beam.sdk.io.LocalResourceId.fromPath(LocalResourceId.java:45)
at org.apache.beam.sdk.io.LocalFileSystem.toMetadata(LocalFileSystem.java:323)
at org.apache.beam.sdk.io.LocalFileSystem.matchOne(LocalFileSystem.java:300)
at org.apache.beam.sdk.io.LocalFileSystem.match(LocalFileSystem.java:105)
at org.apache.beam.sdk.io.LocalFileSystem.match(LocalFileSystem.java:98)
at org.apache.beam.sdk.io.FileSystems.match(FileSystems.java:121)
at org.apache.beam.sdk.io.FileSystems.match(FileSystems.java:142)
at org.apache.beam.sdk.io.FileSystems.match(FileSystems.java:154)
at org.apache.beam.sdk.io.FileIO$MatchAll$MatchFn.process(FileIO.java:719)
at org.apache.beam.sdk.io.FileIO$MatchAll$MatchFn$DoFnInvoker.invokeProcessElement(Unknown Source)
at org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:211)
at org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:188)
at org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement(DoFnRunnerWithMetricsUpdate.java:62)
at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction.flatMap(FlinkDoFnFunction.java:122)
at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction.flatMap(FlinkDoFnFunction.java:59)
at org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:79)
at org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:204)
at org.apache.flink.runtime.taskmanager.Task$$Lambda/0x000000600156b780.run(Unknown Source)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
at java.base/java.lang.Thread.runWith(Thread.java:1596)
Job Complete, Press [ENTER] to Shut Down Embedded Flink Cluster
Stopping taskexecutor daemon (pid: 17221) on host e14bb1607b87.
Stopping standalonesession daemon (pid: 16914) on host e14bb1607b87.

[heschmidt04]

I might have gotten around it by setting job and task manager in Flink to 9200m - It's what Jared had originally set in his Docker config. I'll report back today if this is what fixes it.

[heschmidt04]

It is running now. It is taking ~5 minutes per note. In one hour and 10 minutes the pipeline has parsed 18 of the total 2,597,521 files for one note type. This will take time.

[qqndrew]

@heschmidt04 which pipeline are you running? Information Extraction or De-ID? This seems very slow, the throughput should be 2-3 documents per CPU core per second.

Also you can scale up the number of cpu cores used by scaling up task slots and parallelism in the flink config.

The recommended defaults that we supply in the setup script is (you probably bypass this with your docker image)

taskmanager.numberOfTaskSlots (generally number of cores available for use, good starting point is CPU * .8 rounded down), parallelism.default (set equal to number of task slots), and taskmanager.memory.process.size (good start is 2GB * number of task slots)

[heschmidt04]

Hi Andrew,

I am running this on my personal M1 laptop with 32GB memory and 8-core CPU
Can you sanity check my assumptions that the idea of 6 is valid with the OS mem and core I have?
I'll use 4 just to be cautious with Docker Desktop in the mix.
I have upped the root volume size to 150 GB, 2 GB swap, and 8GB memory to Docker desktop.

taskmanager.numberOfTaskSlots = 6 # I'll dial that back to 4 to have mem left over since it is 1 by default
parallelism.default = 6 # ditto as above ; default is 1
taskmanager.memory.process.size = 8000m # based on 4

Do I need to set jobmanager.memory.process.size to be the same value as taskmanager.memory.process.size?

Also to lessen the processing load - I am splitting out the Progress notes.csv file into 4 chunks of 650,000 note ids to process in 4 batches - then recombine the part*.csv files to load into note_nlp in chunks.

I am running the run_pipeline_local.sh script inside the Docker container.
We aren't doing De-id on note_text column at this point for CHoRUS Dec data delivery for OMOP note table data.
From what I understand the December data deliverable is to provide the

• OMOP Note table without note_text or note_title, and to provide
• OMOP NOTE_NLP with out snippet or lexical variant.

I'm using the run_pipeline_local.sh shell script to apply tagging to the Note table data to use for loading into the Note_nlp table.
The config file is attached earlier to the issue - run_filesystem_csv.json

The input are CSV files that follow the pattern note_id.csv ; where the note_id is a number and the contents is the text from the note_text column.
Ex: 500791606.csv is the filename
And 500791606.csv filename contents are like
500791606,2898,2000-01-24,2000-01-24 11:55:00.000,32817,3000735,Progress Note,"<readacted>",32678,4180186,272255,95931,,Progress Note,,
The JSON config file high at level takes data from /BackboneRoot/source/note_id.csv and writes the output to /BackboneRoot/output/part*.csv

grep inputTag configs/run_filesystem_csv.json
"inputTag" : "Filesystem Records"
"inputTag" : "MedTaggerBackboneTransform"
"inputTag" : "MedTaggerOutputToOHDSIFormatTransform"

[qqndrew]

Ah in this case it seems you are running IE. In that case something weird is definitely going on, it should be processing in excess of 2-3 documents per second per CPU core

Do I need to set jobmanager.memory.process.size to be the same value as taskmanager.memory.process.size?

No

Also to lessen the processing load - I am splitting out the Progress notes.csv file into 4 chunks of 650,000 note ids to process in 4 batches - then recombine the part*.csv files to load into note_nlp in chunks.

Ah yes, ideally, if you could split it into smaller chunks things can be parallelized (if there is only 1 file, then it will only use 1 thread for processing.

Alternatively, enable the "fanout" flag in CSVExtract and it will do that for you, assuming you are doing CSV-based reads (if you do this you will need to increase jobmanager memory though_, but it seems based on your description like you are doing a file-based read instead (which is already parallelized?)

Can you share run_filesystem_csv.json (obviously after opening it and scrubbing/checking for any personal info)?

[heschmidt04]

The config is attached and does not have any personal info. Sent you email to coordinate a video call. It would be easier to target the main suggestions via video. Whatever ends up working well I'll document so folks can package it up and run it in a container if they don't have larger infrastructure to run it on.

…

On Thu, Dec 12, 2024, 13:23 Andrew ***@***.***> wrote: Ah in this case it seems you are running IE. In that case something weird is definitely going on, it should be processing in excess of 2-3 documents per second per CPU core Do I need to set jobmanager.memory.process.size to be the same value as taskmanager.memory.process.size? No Also to lessen the processing load - I am splitting out the Progress notes.csv file into 4 chunks of 650,000 note ids to process in 4 batches - then recombine the part*.csv files to load into note_nlp in chunks. Ah yes, ideally, if you could split it into smaller chunks things can be parallelized (if there is only 1 file, then it will only use 1 thread for processing. Alternatively, enable the "fanout" flag in CSVExtract and it will do that for you, assuming you are doing CSV-based reads (if you do this you *will* need to increase jobmanager memory though_, but it seems based on your description like you are doing a file-based read instead (which is already parallelized?) Can you share run_filesystem_csv.json (obviously *after* opening it and scrubbing/checking for any personal info)? — Reply to this email directly, view it on GitHub <#7 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAYR4GOQY2NOIAUPHL6LQEL2FHIB5AVCNFSM6AAAAABTLTEYHSVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDKMZZG4ZDCMZZGE> . You are receiving this because you were mentioned.Message ID: ***@***.***>