Modern Earth Observation (EO) often analyses hundreds of gigabytes of data from thousands of satellite images. This data usually is processed with hand-made scripts combining several tools implementing the various steps within such an analysis. A fair amount of geographers' work goes into optimization, tuning, and parallelization in such a setting. Development becomes even more complicated when compute clusters become necessary, introducing issues like scheduling, remote data access, and generally a greatly increased infrastructure complexity. Furthermore, tailor-made systems are often optimized to one specific system and cannot easily be adapted to other infrastructures. Data Analysis Workflow engines promise to relieve the workflow developer from finding custom solutions to these issues and thereby improve scalability, reproducibility, and reusability of workflows while reducing development cost at the infrastructure side. On the other hand, they require the workflow to be programmed in a particular language, to obey certain principles of distributed processing, and to properly configure and tune the execution stack, which puts additional burden to data scientists. Here, we study this trade-of using a concrete EO workflow for long-term vegetation dynamics in the Mediterranean. The original workflow was programmed with FORCE, a custom-made framework for assembling and executing EO workflows on stand-alone servers. We ported it to the scientific workflow system Nextflow, which is capable of seamlessly orchestrating workflows over a large variety of infrastructures. We discuss the pitfalls we faced while porting the workflow, advantages and disadvantages of such an approach, and compare in detail the eficiency of both implementations on various infrastructures. We quantify the overhead in execution time incurred by the workoflw engine and give hints on how to deal with heterogeneous tasks. Overall, our Nextflow implementation shows promising behavior in terms of reusability and scalability, though this does not apply to all workflow stages.
distributed system should be used. Although customDeveloping, modifying, and executing workflows is the made systems are still very popular, they face the addidaily business of many EO scientists. Those workflows tional challenge that they are often optimized for a spetypically deal with large amounts of input data passed cific environment and accordingly not or only with large through a sequence of diferent tools transforming and efort portable to other systems. This impedes reusability extracting valuable insights from the data. Due to the and reproducibility [1, 2]. sheer amount of data and the complexity of some of the Data Analysis Workflow (DAW) engines like Nextprocessing steps, EO workflows are rather resource hun- lfow [ 3], Airflow [ 4], or Pegasus [5] promise to reduce gry; at the same time, the tools involved exhibit very the complexity by providing automatic parallelization, heterogeneous requirements in key factors such as mem- distribution, and scalability over large clusters, improved ory, I/O performance, or compute power. reproducibility and reusability, and generally reduced
Building infrastructures for designing and executing development cost. However, they require a steep learning such workflows is demanding. It requires a deep un- curve, which might look rather unattractive for a domain derstanding of parallelization strategies, tool synchro- scientist interested in analyzing only a concrete data set. nization, scheduling, and data transport mechanisms. In this paper, we study the trade-of between the efort necessary to port an EO workflow from a custom solution to a DAW engine and the benefits one can harvest once the port is done. Thereby, we do not change the software used. We report typical problems one faces in such a transformation and analyze the behavior of the diferent approaches on diferent infrastructures in detail.
Although the choice of infrastructure for running DAWs over large scientific data sets is a decision with long-lasting impact, there are comparably few works that provide guidance and benchmarks [2, 6, 7, 8, 9]. Such comparisons also quickly become outdated as infrastructures and systems evolve. Here, we want to contribute to closing this gap for the particular case of EO workflows.
Our paper is structured as follows: Section 2 presents FORCE and the original workflow. Section 3 describes Nextflow, and Section 4 discusses important design decisions that had to be taken for porting the workflow.
In Section 5, we outline our experiments and Section 6 presents a quantitative evaluation of both approaches.
Our findings are discussed in Section 7. The original and the ported workflow can be found at GitHub 1. 2. FORCE and the EO workflow g n i rssceo p e r P generateAnalysis
Mask lve g Le in
s rh sc e e igH rPo g isn processMosaic trssceo p o P generateTile
AllowList
Preprocess processHigherLevel processPyramid
generateAnalysisMask (1) Stage: Prepare Stage: Preprocess Preprocess (2,794) mergeBOA (88)
mergeQAI (88) Stage: Merge processHigherLevel
(28) processPyramid processMosaic
(280) (10) Stage: Higher level processing
checkResults (1) (a) FORCE workflow DAG (b) NextFlow workflow DAG FORCE2 (Framework for Operational Radiometric Correc- Figure 1: DAGs: Boxes represent processes and arrows their tion for Environmental monitoring [10]) is an all-in-one execution order (a) or mutual dependencies (b). Solid arrows solution for the analysis of large volumes of medium- mean that all parent tasks/the parent process must finish resolution EO data. FORCE was initially developed as a caorrmowplsetienldyibcaetfoertehdaetpaenddeepnetndtaesnktctaanskstcaarnt, swtharetreaasssdoaosnheads means to reduce the entry barrier of large-scale EO analy- its parent task has been processed. Solid boxed mark CPU-, ses by providing a rich and integrated toolset that may be dashed boxes IO-bound tasks. Numbers in brackets represent linked with operational monitoring systems [11]. FORCE the number of executions. consists of various tools that can be assembled into com- The 1-arcsecond SRTM Digital Elevation Model (DEM: plex EO workflows, including methods for data download, 1GB) was used for preprocessing. Gaps were filled image preprocessing, data aggregation/reduction, and using the ASTER DEM. A pre-compiled water vapor analyses techniques based on machine learning or time database [14, 15] was used for correcting gaseous abseries analysis. FORCE workflows are configured via pa- sorption during atmospheric correction (3GB). rameter files to define processing options, input/output locations, parallelization parameters, etc.
Original workflow Long-term vegetation dynamics on Crete The original workflow consists of multiple sequential processing stages (Figure 1a). Processing within stages In this paper, we focus on a specific workflow to re-assess is parallelized. All tools involved are executed in Docker the widespread rangeland degradation in the Mediter- containers orchestrated through a Bash script. ranean as reported 20 years ago with limited input 1) Preprocessing. All input images are converted to data [12, 13]. With the unlimited data access of today, Level 2 Analysis Ready Data (ARD [10]) with corrections our analysis shows that total vegetation on the island for atmospheric, topographic, adjacency, and BRDF efof Crete, Greece, did rather increase. Yet, we still can- fects, as well as cloud/shadow detection [16, 17, 18, 19, not dispel that vegetation degradation occurred as most 20, 21]. Corrected images are reprojected to a shared proincrease in vegetation cover was found in the woody jection (EPSG:3035) and split into image chips according vegetation, which potentially represents a degradation to a regular 30km x 30km reference grid, thus forming process related to the increase of impalatable species. a tiled data cube structure. FORCE processes multiple Data: For the analysis in this paper, we used 304GB of input images in parallel. input data covering the island of Crete. We leveraged 2) Higher Level Processing. For each ARD image, multispectral data from the Landsat mission, which ob- fractional vegetation cover is derived [22, 23], followed serves the land surface at 30m spatial resolution each by noise-based [24] outlier detection and inlier restora8–16 days. In total, 2,794 images were retrieved3 (300GB). tion, as well as temporal interpolation [25]. PhenologWe downloaded all available L1TP/T1 data from Landsat ical metrics are acquired [26] to decompose the time 4, 5, and 7 for the years 1984 – 2006 with a cloud coverage series into woody and herbaceous vegetation cover, folof less than 70%. lowed by a change and trend analysis [27]. As higher level processing is often I/O-bound and memory limitations might occur, FORCE sequentially processes the tiles with a nested multithreading parallelization strategy (OpenMP), wherein image blocks are processed in 1https://github.com/CRC-FONDA/FORCE2NXF-Rangeland 2https://github.com/davidfrantz/force 3https : / / console.cloud.google.com / storage / browser / gcp public-data-landsat sequential order with three threading pools taking care In Kubernetes, Nextflow first starts a driver pod which of input, compute, and output, respectively. spawns pods for each workflow task. All pods mount the 3) Postprocessing. To facilitate and accelerate visual same, POSIX compatible, shared file system (Ceph) for interaction, the last stage of the workflow generates im- storing and exchanging data. Each pod executes its task age pyramids and virtual mosaics for each output image. in a local, temporary directory and copies the output to
The workflow produces in total 29GB of output, the shared file system only upon completion. which includes images of interpolated time series, polartransformed time series, time series of phenological metrics, change and trend parameter images - along with 4. Bringing FORCE into Nextflow corresponding image pyramids and virtual mosaics.
After running the workflow, we performed a classification of change and trend parameters into broad land change categories. Area statistics of this map were used for testing reproducibility during our porting.
lfow installation was mostly straight-forward. Tools from FORCE are wrapped as processes in Nextflow processing only one task at a time, while the dependency structure of the workflow is modeled with channels between tasks. The resulting Nextflow workflow comprises nine tasks (Figure 1b). However, the distributed execution environment brings a number of (sometimes subtle) incompatibilities that required adaptation of the FORCE logic.
When executing a workflow, FORCE directly manages task parallelism and scheduling. Execution is expected Synchronized file access : To reduce data redundancy, to run on a single server, meaning that tasks may share FORCE, during preprocessing, merges files representing memory and write to the same disk. In a workflow system parts of the same observation. Specifically, FORCE relike Nextflow, the situation is rather diferent. Resources places two images by their average in case of reflectances on remote machines, as well as remote data access, must or by the latest for quality images. Since images are be carefully managed, and a scheduler must decide which compressed, they first must be read and then written. tasks to put on which compute node. In our installation To avoid data loss or corruption, the access is coordiof Nextflow, these duties are shared among three difer- nated through lock files. This, however, is impossible in a ent independent software systems: we use Kubernetes shared-nothing distributed setting as access may happen for distributed resource management, Nextflow for or- on diferent machines. chestrating and scheduling task executions, and Ceph as In the Nextflow workflow, we changed this pattern a distributed file system. We briefly describe Nextflow in to make tasks independent - and thus parallelizable the next paragraphs and refer the reader to the Kuber- from each other (see tasks in the same horizontal level netes documentation4 and [28] for details on Ceph. in Figure 1b). To retain the original functionality, we
Nextflow 5 is a rather recent, domain-agnostic DAW subsequently group all images by their tile, date, and engine consisting of a workflow language, the workflow satellite, and then merge every group in a separate, newly engine, and a set of connectors to run workflows on difer- introduced process (see merge step in Figure 1b). This ent infrastructures, including Kubernetes [3]. It emerged procedure improves scalability and flexibility regarding from Bioinformatics applications but also finds increas- the execution environment but also generates more I/O ing uptake in other scientific domains and in industry. since files potentially have to be moved to diferent nodes.
Nextflow workflows consist of channels (for data ex- This change, however, created a new issue. When imchange) and processes (for task execution). Processes plementing each merge step as an individual task, we may wrap any command-line tool, ofering the option to observed very ineficient workflow execution. The reainclude tasks implemented in diferent languages into the son is that each of these tasks requires only a few seconds same workflow. Channels define dependencies between but must be started and configured anew by Nextflow for processes by means of provider/consumer relationships every execution, leading to start-up times being higher at the file level. Files are passed into the according output than execution times. As this task is executed thouchannel if a process’ output matches a specified pattern. sands of times, performance degraded considerably. As a Once all output channels that serve as input channels workaround, we decided to execute batches of size 100 for a subsequent process are filled, the respective task as a single process in Nextflow, reducing the cumulative is ready for execution and gets scheduled by the work- start-up times accordingly. This batch size is another lfow engine. As a result, Nextflow automatically executes trade-of in the new system: larger batches reduce partasks in parallel and in a distributed manner. allelism but also wastage through process start-ups. We grouped these tasks by tile since all images of the same tile are processed in the same succeeding task. 4https://kubernetes.io/docs/home/ 5https://www.nextflow .io/ cant diference between FORCE and Nextflow is the way they deal with directory hierarchies. In FORCE, the data are structured in directories such that observation time and the observing satellite are encoded in the filename, whereas location is encoded in the directory name (as tile ID). In contrast, Nextflow only works on the file level and cannot easily cope with directories. As a remedy, we rename the output images of the preprocessing task by prefixing them with the tile ID. We clip this prefix for the higher level process in a custom wrapper and create a directory as expected by the task. Clearly, such tricks do not make a workflow easier to understand. 5. Evaluation setup that this is a somewhat outdated hardware, as clusters today typically have much faster network (e.g. 10GBit/s) and more memory per node (e.g. 128GB). Slow networks impede I/O hungry tasks, whereas small main memory reduces the degree-of-parallelism for memory-hungry tasks. Ceph was configured to use two 1TB hard drives per node as block storage, resulting in 12TB raw storage capacity on six nodes. In order to improve I/O operations, two 1Gbit/s network interfaces of the nodes are utilized to segment Ceph management and storage trafic. We ran the workflow using 1, 3, 5, 10, 14, 15, 20, and 21 compute nodes. Note that the 14 node setup ofers the same number of cores and threads as the HPS.
We performed two types of experiments to investigate whether the ported workflow scales as expected and to Results are summarised in Figure 2. The X-axis in all detect potential bottlenecks. We first ran the original graphics determines the number of nodes used; the runworkflow in its original environment to obtain confirmed time of the HPS setup is drawn at x=14, in addition to results and ensured that all other configurations produce the runtimes of the 14 node cluster. We first describe the the same results. We subtracted the runtime of the check- overall results of Experiment 1 and 2 and then provide a result task in the Nextflow workflow from the overall detailed analysis of individual workflow stages. We pubexecution time, to achieve comparable results. lished all logs, additional plots, a tabular preparation of
Experiments were repeated three times; we report the results, and the analysis script in our GitHub repository. median of the measured runtimes. We measure wall- Experiment 1: HPS: We first run FORCE on HPS, where clock execution times. For the distributed setting, we it finishes after 358min. Running the Nextflow workflow also report on eficiency of task executions, defined as on the same machine requires 384min. The 7.5% increase the theoretical time obtained by dividing single node ex- in runtime is expected as a system like Nextflow for a ecution time through the number of nodes, divided by single server only adds overhead without ofering any the observed runtime. Thus, an eficiency of 1 means performance benefits. perfect scaling, while 0.5 means that the distributed runtime is only half as good as theoretically possible. We Experiment 2: Cluster: We can only measure the Nextutilized Nextflow version 21.04.0-edge with bugfixes 67 flow workflow on the cluster as FORCE does not support in the cluster with Kubernetes version 1.19.3 and Ceph execution on Kubernetes. Running the Nextflow workversion 15.2.8, and Nextflow version 20.10.0 locally. lfow on a single node of the cluster took 4,883min, which decreased to 315min when using all 21 nodes (Figure 2d). This implies a reduction of a factor of 15.5, which is considerably less than the ideal factor of 21, hinting at problems in the parallelization. Using 21 nodes is thus 13.4% faster than using the HPS despite the slow network.
On HPS, on the other hand, the Nextflow workflow required only 11.0% less time than on a cluster with 14 nodes (i.e., an equal number of cores), meaning that the overhead of managing tasks and files in a distributed setting is surprisingly low.
original workflow is optimized for a specific HighPerformance Server (HPS), we first ran FORCE and the Nextflow port on this machine, a Linux server with an Intel Xeon Platinum 8176M CPU (56/112 cores/hyperthreads, 2.10GHz), 750GB main memory, and 98TB disc space (RAID6, ext4). Running the Nextflow workflow on this machine allows measuring the efects of our changes to the workflow and the computational overhead imposed by the more complex parallelizing strategy.
Experiment 2 (Cluster performance): To analyze the 4,338min on one cluster node; using 21 nodes decreased scalability of the Nextflow workflow, we ran it on a lo- runtime to 225min, i.e., a 19.2-fold improvement with an cal cluster consisting of 27 homogeneous nodes, each eficiency of 91.6% (Figure 2a). On HPS, it took 337min equipped with an Intel Xeon E3-1230 V2 CPU (Quad-Core, and 339min in our cluster with 14 nodes. The preprocess3.30Ghz), 16GB main memory, and three 1TB hard disks. ing is CPU bound; accordingly, the cumulative runtime Nodes are connected with a 1Gbit/s network link. Note is constant over all experiments (Figure 2c). The HPS shows a 46.4% higher cumulative runtime, which is likely due to a diferent CPU architecture. As the preprocessing 200% e sag150% u u .cp100% g v a 50% 100% 80% e itnm60% u r lta 40% o t 20% 0% 1 3 5 scales quite favorably, we conclude that our uncoupling configuration. The original setup is not able to benefit as described in Section 4 works well. Only in the stage’s as much of these features, reaching only 75.6% eficiency beginning, all tasks read 80 – 204MB simultaneously, on HPS, because it is not optimized for the case where which might decrease the eficiency when more nodes the preprocessing directly runs before and data is still in are used than available here. the cache. Thus, FORCE optimizes the read patterns in
The merging stage scales slightly inferiorly. The ef- the higher level processing. Therefore, it already reads ifciency decreases by around 2.5% per additional node, the following image blocks while processing the current resulting in a drop in the cumulative runtime when 20 ones (see Section 2) to reduce I/O wait times, leading to or more nodes are used. We assume this is because the less idle CPUs. The tiles, in turn, are processed in sequenmerge task is I/O-bound and thus heavily limited by the tial order, using all available cores via shared memory slow network. With more nodes, the load is fairly dis- parallelization, which also adds some synchronization tributed, and the individual connections are relieved. overhead. In Nextflow, multiple tiles are processed in par
The workflow reaches an eficiency of 90.0% for the allel leading to parallel reads. In the distributed setting aggregated preprocessing and merging stage when using this overloads the network switch, resulting in high I/O 14 nodes (56 cores). This setup is slightly slower than wait times and hence longer runtime (Figure 2c). On HPS, the FORCE workflow on HPS (same number of cores); many of these parallel reads are avoided by exploiting it surpasses FORCE from 16 nodes on, finishing 24.5% the large caching capacities, leading to lower runtime for earlier than FORCE on HPS when using all 21 nodes. Nextflow than for FORCE.
Higher Level Processing: The higher level processing stage scales considerably worse because it is very I/O 7. Discussion and Conclusion intensive, which turns the slow network into a major bottleneck (see the decrease of average CPU usage in Figure Overall, our experiments show that a workflow engine 2b). This stage takes 319min on one node, which is only like Nextflow can help to make EO workflows more scal4.7 times longer than using 21 nodes. The cumulative able and to ease their reuse on diferent infrastructures. runtime increased significantly (Figure 2c), growing to The benefits should especially pay of when using larger 245.4% for 20 nodes as the I/O times rise for each task. clusters for larger input data sets which should be possi
The I/O problems in this stage also become apparent ble without any changes to the workflow. Besides scalabilwhen comparing HPS with the cluster. Using 14 nodes, ity concerns, the initial price one has to pay for learning the Nextflow workflow had an eficiency of 33.4% (Fig- the specificities of a workflow engine and the peculiariure 2a). Running the Nextflow workflow on the HPS ties of a distributed execution is notable; this investment, with the same number of cores, eficiency was 96.6%. The on the other hand, should pay of quickly when more throughput increased through (a) Linux’s caching capabil- and more workflows are to be implemented or truly large ities combined with the large memory, and (b) the RAID6 analysis tasks, exceeding the capabilities of stand-alone servers, are approached. In the following, we discuss for larger analysis tasks, reusing such preexisting disa number of further and more detailed conclusions we tributed software infrastructures seems inevitable. draw from our experiments. Second, programming a workflow in Nextflow requires more code than in a Bash script. Tasks must be wrapped, Scalability of stages and their inputs, outputs, and parameters defined. This creates boilerplate code which can make a workflow more Overall, our Nextflow workflow scaled quite well, reach- dificult to read and debug. These efects occur especially ing an eficiency of 73.7%. Nextflow’s distribution and when existing code is ported into a workflow system; parallelization techniques ofer a comparable perfor- they are probably less of an issue when workflows are mance to manual ones, outperforming them in excep- designed ab-initio for such a distributed system. tional cases. The scalability, however, is not equal be- Third, while Nextflow supports multiple resource mantween the diferent stages. While preprocessing and agers, it is partially impossible to reuse the same paramemerging scale very well, higher level processing suf- ters in diferent systems. For example, the memory usage fers a lot from the distributed data access and slow net- varies on other machines even for the same task/input work. This should be taken into account when groups combination since I/O speed difers and tool-internal focus their research on only one of these aspects. The caching gets more heavily used. This could also happreprocessing step is more and more transitioning into pen within a heterogeneous cluster. We plan to develop the hands of institutional data providers through the automatic, accurate, and in particular input-dependent widespread emergence of ARD (see e.g. [29]), thus data resource predictions for future work [31, 32]. providers would benefit largely from using a workflow Fourth, Nextflow (and Kubernetes) are under continmanagement system. End users, however, increasingly uous development. There are a number of features one start from existing ARD and thus focus on higher level could imagine to make their application easier and their analyses, which at least in our setup, did not benefit as executions even more scalable. For instance, task agmuch from the workflow porting. We are, however, con- gregation [33] to tune the start-up/runtime ratio (see ifdent that also this stage can be adapted better for a Section 4) is a known technique from high performance distributed execution, for instance, by using some com- computing that did not yet make it into the workflow pute nodes as storage nodes or by using a location-aware world. Handling directories could be improved to ofer a scheduler to reduce data movement and better explicit op- more powerful file management (see Section 4). Ofering eration system caching. Besides, we would already expect a shared-memory and/or streaming interface between a quite diferent behavior when running the workflow tasks could help to reduce intermediate I/O, although it on a cluster with more up-to-date network bandwidth. must be carefully designed to also work smoothly in a
The scalability of other higher-level EO workflows distributed setting since it creates additional constraints. might also difer. For example, more compute-heavy Finally, Nextflow also ofers a number of features we workflows with machine learning-based components (e.g. did not further discuss here. For instance, it is capable to [30]) require much more computational power and thus resume workflow execution after a stop or crash, which outweigh the I/O time as the limiting factor. However, avoids a lot of unnecessary computation and comes in these workflows usually include a feature generation task very handy during workflow development. However, we with a similar resource requirement as the higher-level faced problems with the resume functionality when readworkflow used in this paper. ing input files through the GDAL API in the higher level processing, thus disabling Nextflow’s resume capability.
Developing EO workflows with Nextflow Moreover, Nextflow ofers comfortable monitoring features while running a workflow and extensive logging to Scientific workflow systems like Nextflow promise addi- perform runtime profiling, analysis, and tuning. Finally, tional benefits besides scalability. We give a short account it ofers the Nextflow Tower 8 for managing and adminison our experiences during porting. tering entire repositories of workflows, which becomes
First, Nextflow achieves automatic parallelization, ad- important when organizations grow larger. equate distributed scheduling, and distributed file han- To compare Nextflow with other workflow engines, we dling. For scheduling and parallelization, each task in also work on porting the workflow to Apache Airflow 9. Nextflow is defined with its required memory and CPU, Nevertheless, while porting the workflow, Nextflow was wherein FORCE explicitly defines the level of parallelism. very intuitive and helped us in understanding the workNot having to implement these features oneself is a con- flow and tasks’ dependencies due to its simple DSL. siderable benefit. On the other hand, designing worklfows with Nextflow requires “distributed thinking” and prevents many tricks that are possible in local machines.
For smaller analysis tasks, this probably is an overhead;
Landsat data courtesy of the U.S. Geological Survey. We
thank Google for openly mirroring Landsat collection
1 on the Google Cloud Platform. We thank the two
anonymous reviewers for their valuable feedback on our
manuscript. This work was funded by the German
Research Foundation (DFG), CRC 1404: "FONDA:
Foundations of Workflows for Large-Scale Scientific Data
Analysis".
ows in Multitemporal Dryland Landsat Imagery: ries of Satellite Data, IEEE Journal of Selected
Extending Fmask, IEEE Geoscience and Remote Topics in Applied Earth Observations and
ReSensing Letters 12 (
Lewis, C. B. Schaaf, Q. Sun, J. Li, H. Huang, V. Koval- [31] C. Witt, M. Bux, W. Gusew, U. Leser, Predictive
skyy, A General Method to Normalize Landsat Re- Performance Modeling for Distributed Computing
lfectance Data to Nadir BRDF Adjusted Reflectance, using Black-Box Monitoring and Machine Learning,
Remote Sensing of Environment 176 (
doi:10.3390/f11060606. [27] J. N. Hird, G. Castilla, G. J. McDermid, I. T. Bueno,
A Simple Transformation for Visualizing Nonseasonal Landscape Change From Dense Time Se