dataesr/scientific-paragraphs-categorization huggingface dataset:
Availability of data and materials The source code of AutoClassWeb is open-source, released under the BSD-3-Clause license, and available on the GitHub development platform https:// github. com/ pierr epo/ autoc lassw eb. AutoClassWeb source code in its current version 2. 2. 1 is archived in Software Heritage with the following reference: swh:1: dir: 173e8 46a51 37a4b 498ca 7da9e ca980 79063 1bc1a.
Anatomical images were processed using ImageJ software, which is intended for scientific use and is freely accessible on the website at www. imagej. net (Schindelin et al. , 2015)   (Rueden et al. , 2017). All measurements done in ImageJ software were done manually without using any automatic processing tools. After the hearts were scanned, they were cut into 3-5-mm-thick slices, which were placed side by side, photographed by the camera (Nikon DX AF-S NIKKOR), and analyzed using the same ImageJ software. We opted for this method of analysis because manual measurement of the pathology samples is associated with a risk of inhaling toxic formaldehyde fumes. Each lesion was measured three times and evaluated by two independent observations. As the last step, the heart was removed out from formaldehyde, rinsed under running water and several lesions with characteristic parameters (without "steam pop" effect) were cut out from the heart, with an approximate volume of 1 × 1 × 0.
Fast low-angle shot (FLASH) scout images were used to localize the left ventricle with ablation lesions. Protondensity (PD) weighted anatomical images were taken using FLASH sequence with TR = 1268. 9 ms, TE = 4. 43 ms, FA = 45. 7 °, FOV 100 mm x 100 mm, 8 averages, and a 256 × 256 image matrix. Eighty axial slices with a thickness of 1. 0 mm were acquired; the slices covered the whole heart. The raw MRI data format was exported to the DICOM format for the further analysis. These anatomical images provided sufficient background for optimal visualization of lesions and volumetric data analysis. Lesions were measured using ImageJ software. After image import, the ImageJ window was calibrated using the known pixel-to-mm ratio (1. 024 pixel/mm). The measurement of each lesion took place in a slice, where the lesion reached its maximum parameters of depth (A), width (B), estuary (D), and depth at the maximum diameter (C) (see Figure  1 ).
Using a long sharp knife, the heart is gradually sliced. Every heart was removed from the fixation solution, placed in a flow box, and manually sliced using the abovementioned macrotome to 3-5-mm-thick slices. Slices were laid side by side, and pictures were taken using a camera (see Figure  5 ). All hearts were photographed in this way. The whole preparation of the heart for measuring pathological sections took approximately 20 min (fix the heart on the macrotome and slice it, take a photo, and upload it to ImageJ). Photos were imported to the ImageJ program. The calibration of the software was set manually according to the settings of the caliper, which is located on each photo and shows exactly the size of 1 cm. The depth, width, estuary, and depth at the maximum diameter were measured in the same way, as in MRI scans (from photo where is the lesion in maximal dimensions, black surrounding zone not included) (see Figure  2 ).
Funding This research was supported by the  European Regional Development Fund-Project ENOCH  (No. CZ. 02. 1. 01/0. 0/0. 0/16_019/0000868 ) and  Medical University of Silesia  statutory funds ( KNW-1-048/N/9/K, PCN-1-005/N/0/K. The MR measurements were supported by the grant  LM2015062  " National Infrastructure for Biological and Medical Imaging (Czech-BioImaging )". This publication was supported also by  Masaryk University  (project " New imaging, computational and analytical methods in diagnostics and monitoring of cardiovascular disorders "  MUNI/A/ 1462/2021 ) with the support of the  Specific University Research Grant , as provided by the  MEYS  CR in 2022.
In the era of data deluge, Big Data gradually offers numerous opportunities, but also poses significant challenges to conventional data processing and analysis methods. MapReduce has become a prominent parallel and distributed programming model for efficiently handling such massive datasets. One of the most elementary and extensive operations in MapReduce is the join operation. These joins have become ever more complex and expensive in the context of skewed data, in which some common join keys appear with a greater frequency than others. Some of the reduction tasks processing these join keys will finish later than others; thus, the benefits of parallel computation become meaningless. Some studies on the problem of skew joins have been conducted, but an adequate and systematic comparison in the Spark environment has not been presented.
They have only provided experimental tests, so there is still a shortage of representations of mathematical models on which skew-join algorithms can be compared. This study is, therefore, designed to provide the theoretical and practical basics for evaluating skew-join strategies for large-scale datasets with MapReduce and Spark-both analytically with cost models and practically with experiments. The objectives of the study are, first, to present the implementation of prominent skew-join algorithms in Spark, second, to evaluate the algorithms by using cost models and experiments, and third, to show the advantages and disadvantages of each one and to recommend strategies for the better use of skew joins in Spark.
Introduction Big Data is a term that has been mentioned in many recent studies. People generate terabytes of data every hour, leading to challenges in storing and handling data in traditional ways. Therefore, Google designed the MapReduce processing model  [1]  for parallel and distributed processing of large-scale datasets. One of the major limitations of the MapReduce model is data-skew processing  [2]. A typical case is data skew in join operations, which are the operation of joining many relationships or datasets that have some common attributes into a new relationship  [3]. The join operation is used in many applications, such as the construction of search engines and some big data-intensive applications  [4]. Taking the example of joining two datasets R and L, R has the join key column R.
id = {1, 1, 1, 1, 1, 1, 2} and L the has join key column L. id = {1, 2, 3}. In MapReduce, Mappers read input data from the two datasets and distribute intermediate data to reducers based on join keys. Data with the same join keys will be sent to the same reducers to create join results. Assuming that dataset R is skewed at id = 1, the join key with a value of 1 occurs many times. In this case, the reducers r 2 and r 3 will finish faster than reducer r 1 , since reducer r 1 receives more data than the others. If there is a large amount of skewed data with id = 1, then overload and congestion will appear at the computing node r 1 and lead to inefficiency in the computational system. This is a real challenge for join operations in MapReduce. Some algorithms have been proposed for data-skew-join operations in MapReduce, such as hash-based partition  [5] , range-based partition  [5] , multi-dimensional range partition (MDRP)  [6] , MRFA-Join  [7] , and randomized partition  [8].
Chen et al. [9]  introduced LIBRA, a sampling and partitioning algorithm for handling high-frequency join keys in reduction tasks. Bruno et al. [10]  discussed challenges in large-scale distributed join operation and introduced novel execution strategies that robustly handle data skew. They used different partitioning schemes and join graph topologies for high-frequency key values. Zhou et al. [11]  proposed an efficient key-select algorithm to find skew key tuples and defined a lightweight tuple migration strategy to solve data-skew problems. Their FastJoin system improved the performance in terms of latency and throughput. Zhang and Ross  [12]  presented an index structure to reorder data so that popular items were concentrated in the cache hierarchy. They analyzed the cache behavior and efficiently processed database queries in the presence of skew.
e, hash-based partition, range-based partition, and multi-dimensional range partition. This helps users to choose suitable solutions for processing join operations in large-scale datasets. We built cost models for the algorithms for evaluation. This provided a scientific basis for the comparison. Experiments were conducted on a Spark cluster with different skew ratios. The structure of this paper is organized as follows. Section 2 presents the background related to the large-scale data processing model and platform. Section 3 provides the solutions for data skew in join operations on large datasets in detail. Section 4 presents an evaluation with the cost models and experiments conducted in the Spark cluster. The conclusion of the paper is presented in Section 5.
Join Operations in the MapReduce Model MapReduce  [1]  is a parallel and distributed large-scale data processing model. A program can be run on clusters with a number of computing nodes that can be up to thousands. Introduced in 2004, MapReduce has been widely used in the field of Big Data, since it allows users to simply focus on the design of data processing operations regardless of the parallel or distributed nature of the model  [3]. MapReduce is implemented through two basic functions, Map and Reduce, which are also two consecutive stages in data processing. The Map function receives input data to convert them into intermediate data (key-value pairs), and the Reduce function accepts the intermediate data created to perform calculations. Hadoop (http://hadoop. apache. org (accessed on 8 May 2022)) has become one of the popular Big Data processing platforms of the last decade  [16].
Considering the two datasets R (user data) and L (log data), to list a username and the corresponding event that the user accesses, we have the query R(uname, uid) uid=uid L(uid, event). In large-scale data applications, such as social networks, this query can enforce the joining of trillions of records. Therefore, the parallel and distributed processing model in MapReduce is a good solution to this problem. As shown in Figure  1 , the JobTracker of MapReduce creates three Mappers to process three partitions of the input data simultaneously. The first Mapper computes the first partition, consisting of three records of dataset R. The second Mapper processes one record of R, and the third Mapper processes two records of L. The Mappers transform the datasets R and L based on the join key uid.
• {(A, B), (C, B), (A, F)} → {(B, A), (B, C), (F, A)} • {(C, D)} → {(D, C)} • {(B, C), (D, F)} → {(B, C), (D, F)} The data after transformation are called intermediate data, and those are sent to the Reducers. Intermediate datasets with the same join key are sent to the same Reducer. Here, the reduction function is called for every single key with a list of values. Finally, each record of R finds the records of L that have the same join key to produce the join results. Figure  1  shows that there are three join results from the Reducers.
Apache Spark Apache Spark (http://spark. apache. org (accessed on 8 May 2022) ) is an open-source cluster computing framework that was originally developed in 2009 by AMPLab at the University of California, Berkeley. Spark has continued to be developed by the Apache Software Foundation since 2013. Given a task that is too large to be handled on a server, Spark allows us to divide this task into more manageable tasks. Then, Spark will run these small tasks in memory on a cluster of computing nodes. Apache Spark has three salient features. • Speed: Spark is 100 times faster when running in memory and 10 times faster when running on a disk than Hadoop  [17]. • Support for multiple programming languages: Spark provides built-in APIs in the Java, Scala, and Python languages. • Advanced analysis: Spark not only supports MapReduce models, but also supports SQL queries, streaming data, machine learning, and graph algorithms.
Resilient Distributed Datasets Resilient distributed datasets (RDDs) are the underlying data structure and outstanding feature of Spark. They are a type of distributed collection that can be temporarily stored in RAM with a high fault tolerance and the capability of parallel computation. Each RDD is divided into multiple logical partitions and can be computed on different nodes of a cluster. In this research, we use the Scala language, since Apache Spark is built mainly on Scala, so it has the best speed and support for this language. RDDs basically support two main types of operations.
Skew Join Data skew is a problem in which the data distribution is uneven or asymmetric. In a database, cases where some attributes appear with a greater frequency than other attributes usually occur  [18] , and this is called data skew. In distributed parallel computation systems, the join operations using these attributes will take a longer time than with data with a normal distribution. A join operation consists of several steps, including data uploading, projection and selection based on the query, partitioning of the data into partitions, and joining of datasets. Therefore, data skew limits the efficiency of join operations in parallel computation. There are several types of data skew based on the stage in which skew problems occur, such as tuple placement skew, selectivity skew, redistribution skew, and join skew  [19].
The initial placement of the datasets in the partitions may cause a tuple placement skew. Selectivity skew occurs when the selectivity of selection predicates is different between nodes. Redistribution skew is caused by a large amount of data on the partitions after the redistribution scheme is applied. Joining product skew is the result of join selectivity on each node. In this paper, we will not consider tuple placement skew, since MapReduce creates split files regardless of the size of the original file. Selectivity skew is also ignored because it does not have any significant impact on the performance, and we do not use projections and selections in our program. Joining product skew cannot be avoided, since it is created after joining two datasets together. Redistribution skew is the main and most important type of data skew, and it affects the distribution of the workload between nodes.
Thus, skewed data occur when one or more computing nodes have to process a much larger number of join keys than other computing nodes in a system  [20]. As a result, some nodes encounter bottlenecks or delays, and the remaining nodes are a waste of resources. In parallel computing, the join operation execution time is determined by the longest-running task. Therefore, if the data are skewed, the benefits of parallel computation become meaningless  [2]. A solution is to clearly point out skew data and to allow it to be distributed in different ways to avoid or reduce the skew effect before calculations begin. In a data-skew problem, heavily weighted partitions will appear. Spark assigns one task per partition, and each worker processes one task at a time; thus, heavy partitions will affect the performance.
The main idea now is to avoid heavily weighted partitions. As an example, we add more information on the skew-join key so that it can be distributed into different partitions (Figure  4b ). In Spark, data are divided into partitions on many different nodes in a cluster. Thus, it is difficult to avoid data shuffling between nodes with join operations in Spark. This shuffling process will slow down the processing speed and program performance. Therefore, reasonable partitioning of data before the join computation can improve the performance and reduce the effect of shuffling of data. Figure  5  presents the data flow of the three algorithms.
Hash-Based Partition In hash-based partition (HBP)  [5] , after mapping the input data, the partition processing will redistribute the workload to the nodes based on the join keys. Hash-based partition is not ideal for handling skewed data, since the data skew will go to the same computing nodes. Therefore, hash-based partition is not a good solution for skewed data  [21]. Suppose that we have two datasets R(k, v1) and L(k, v2), where the attribute k is a join key. The data flow of hash-based partition algorithm is presented in Figure  6. The join operation between two datasets R and L using a hash-based partition algorithm goes through two phases.
• Join phase: The partitions will receive a list of data with the same join key k. Here, we can use any join algorithms, such as join, rightOuterJoin, or leftOuterJoin. Given an example of two datasets R(k, v1) and L(k, v2) with R. k = {1, 1, 2, 2, 2, 2, 4, 4}, L. k = {1, 1, 2, 2, 2, 3, 4, 4}, and there are p = 4 partitions, in hash-based partition, one partition can receive more records than other partitions. The join computation with hash-based partition is shown in Figure  7. In this example, partition P 2 has four records {r3, r4, r5, r6} from dataset R and three records {l3, l4, l5} from dataset L. Thus, partition P 2 produces 12 join results {(r3, l3), (r3, l4), (r3, l5), (r4, l3), (r4, l4), (r4, l5), (r5, l3), (r5, l4), (r5, l5), (r6, l3), (r6, l4), (r6, l5)}, which is a larger number than in other partitions.
In this method, the first thing to do is to create a split vector from the two input datasets by calculating the appearance of the join keys in each dataset. The dataset that has the most skewed join keys will be selected as the split vector. We use the "fragmentreplicate" technique  [22]  before the join computation. When a mapper reads input records and maps into RDD key-value pairs, if the join key belongs to more than one partition, this key-value pair is a fragment or replicated. We will use a fragment of the key-value pairs on the dataset with the most skewed join key and use a replicate of the key-value pairs on the other dataset. The following steps need to be performed to partition key-value pairs in the dataset with more skewed join keys (a build relation): • A split vector is created from the input dataset and this vector is stored in the HDFS; • Each node in the cluster will read the split vector from the HDFS and create a range map, and each sub-range will be assigned to the corresponding partition; • Based on this range map, when a mapper reads input records and maps into RDD keyvalue pairs, if the join key belongs to only one sub-range, the data will be partitioned into the corresponding partition; if the join key belongs to more than one sub-range, the data will be randomly assigned to one of the partitions corresponding to the sub-ranges (fragment).
For a dataset with fewer skewed join keys, we use the split vector created above to partition key-value pairs (a probing relation). If the join key belongs to only one sub-range, the data will be assigned to the corresponding partition. Conversely, if the join key belongs to more than one sub-range, the data will be distributed to all partitions corresponding to the sub-range (replicate). The algorithm used to partition the datasets is presented in Algorithms 1 and 2. The partitions will have a list of data with a join key k belonging to the same sub-range. Here, we use a join algorithm to create the join results (presented in Algorithm 3). for each r i in tupleR do 7: Output r i l j 8: end for 9: end for End. Given an example of two datasets R(k, v1) and L(k, v2), as presented above, we create a split vector from dataset R.
The elements in the split vector will be chosen from R. k with the formula |R| p-1 , in which |R| is the number of records of dataset R. In the example (Figure  9 ), we have |R| = 8 and p = 4; thus, the 2nd, 4th, and 6th elements in R. k will be chosen to have the split vector {1, 2, 2}. From the split vector, we create four sub-ranges, i. e. , [-∞  According to the "fragment-replicate" technique, {r3, r4, r5, r6} are classified into two partitions and {l3, l4, l5} are replicated in two partitions (P 2 and P 3 ). As a result, we have two partitions, each of which produces a join result of six records instead of 12, as in the hash-based partition algorithm. The range-based partition algorithm allows data to be partitioned into more than two partitions. With the "fragment-replicate" technique used, we can divide the workload of the partitions when generating the join results.
Multi-Dimensional Range Partition Multi-dimensional range partition (MDRP) is an algorithm that combines a range partition and a random partition  [6]. Mappers read input data from the datasets and create split vectors to divide the data into sub-ranges. The data will be put into a partitioning matrix corresponding to the sub-range values of the two datasets. Input data are distributed to the reducers based on this partitioning matrix, in which cells containing more data in the matrix will be subdivided and assigned to two or more reducers. The data flow of multi-dimensional range partition algorithm is presented in Figure  10.
Partitioning Matrix The algorithm considers two datasets R and L with a join key of k. Suppose that there are p partitions, and the input datasets will be divided into p sub-ranges, i. e. , {R 1 , R 2 ,. , R p } and {L 1 , L 2 ,. , L p }. Creating sub-ranges of the two datasets is similar to creating sub-ranges with the range-based partition algorithm. For dataset R, the sub-range will include the entire domain of the join key k from α to β, α < R. k < β. Two special cases R 1 and R p will include sub-ranges [-∞, α] and (β, ∞], respectively. As in the previous example, we have R. k = {1, 1, 2, 2, 2, 2, 4, 4} and L. jk = {1, 1, 2, 2, 2, 3, 4, 4} with p = 4 partitions. We get two split vectors {1, 2, 2} and {1, 2, 3} from the two datasets R and L. In the partitioning matrix M, as shown in Figure  11 , the ith row represents the sub-range R i and the jth column represents the sub-range L j.
The cell (R i , L j ) is classified into one of the following two groups: candidate and non-candidate. Candidates are cells that produce join results and non-candidates are cells that do not generate join results. M(i, j) is the value representing the workload of a candidate cell (R i , L j ) to be processed. In the MDRP algorithm, M(i, j) = |R i L j |, where R i and L j are records that belong to the same sub-range. In the partitioning matrix M, M(2, 2) = 12 because we have four records whose join key is 2 in dataset R and three records with a join key of 2 dataset L, which belong to the same sub-range (R 2 , L 2 ). Similarly, M(1, 1) = 4 because we have two records {r1, r2} in dataset R and two records {l1, l2} in dataset L that have join keys belonging to the same sub-range (R 1 , L 1 ).
Multi-Dimensional Range Partition The MDRP algorithm is shown in Algorithm 5. The "fragment-replicate" technique is used for the join operation. Assume that a mapper receives a record r3 with the join key r3. k = 2 of dataset R; then, this record will belong to a cell (R 2 , L 2 ). However, this cell is divided into three partitions P 2 , P 3 , and P 4 , since (R 2 , L 2 , 4) ∈ C 2 , (R 2 , L 2 , 4) ∈ C 3 , and (R 2 , L 2 , 4) ∈ C 4. So, we should use a fragment or a replicate for record r3. By counting the number of records in R 2 , L 2 with join key k = 2, if the number of records of for dataset R is greater than that for dataset L, then r3 will be fragmented (randomly selected partition), and the records in dataset L will be replicated (replicated to both partitions). end if 12: end for End.