dataesr/scientific-paragraphs-categorization huggingface dataset:
Introduction In the past decade, the volume of healthcare data and its accessibility rose quickly. For instance, in France, the SNDS claims database contained 86% of the French population in 2010  [45]  to reach 98. 8% in 2015  [46]  leading to one of the world's largest health Large Observational Database (LOD)  [46, 7]. The exhaustivity of LODs such as SNDS has proven useful for public health research, by improving the statistical power of algorithms using this data and by mitigating the sensitivity to selection biases  [46]. However, such an abundance of data comes at a cost: SNDS is a very complex database, with data spread across hundreds of tables and columns. Its scale makes data manipulation non-trivial. More importantly, using this data requires a tremendous amount of knowledge from SNDS experts.
Many coding or data recording subtleties, such as data duplication caused by administrative complexity, might bewilder inexperienced users. Deriving proper health events definitions and extracting them accurately is, therefore, a difficult task, having important consequences on the derived studies  [46, 13]. These issues are of course not unique to SNDS but shared by many LODs  [25]. This paper proposes an answer to this problem by introducing SCALPEL3 (SCALable Pipeline for hEaLth data), an open-source framework intending to reduce such entry barriers to LODs. This framework attempts to simplify medical concept extraction by providing a set of tools performing batch Extract-Transform-Load (ETL) tasks, while an interactive API eases the manipulation and the exploration of longitudinal cohorts.
Thus, this research focuses on the following objectives: 1. Design and implement a scalable tool allowing to extract and manipulate longitudinal patient data from large observational databases; 2. Simplify methodological research by reducing SNDS data complexity and by easing data loading into formats used by common machine learning libraries; The main concepts used by SCALPEL3 and some related works are presented in Section 2. The LOD for which SCALPEL3 was initially designed for is described in Section 3, together with SCALPEL3 methods and abstractions. The scalability of SCALPEL3 is evaluated in Section 4, while Section 5 discusses its strengths and limitations.
Background LODs are not designed to perform medical research. Electronic Health Records (EHR) data directly supports clinical care and are used to justify care billing and reimbursement, while claims data are primarily used for reimbursement purpose. The data models and terminologies used in such databases were optimized to suit these particular goals, resulting in normalized data models built around hospital stays, transactions, or cash flows  [46]. Extracting meaningful patients care pathways from such data can be decomposed into two tasks. First, all the data corresponding to a set of patients need to be identified and collected. When the data is not normalized around the patients, this task requires several join operations which can be very costly in terms of computations as the data volume increases.
Second, medical concepts have to be properly identified from administrative codes: this phenotyping task relies heavily on a combination of medical and database knowledge. The algorithms used to perform concept extraction from administrative data are either disclosed through scientific publications or shared as lengthy SQL queries  [24]. Their code or the description of the algorithms involved can vary in quality, hindering reuse, and reproducibility. As a result, building a study from scratch might be faster than reusing poorly documented code from previous works  [35, 24]. Besides, access to LODs such as SNDS might rely on proprietary software such as SAS  [40]  or SPSS  [44]. While these tools are suitable to produce public health studies, they hinder methodological research as they do not interact easily with R or Python packages that implement state-of-the-art machine learning algorithms.
They also rely on a normalized data model combined with SQL databases. A collection of open-source software has been developed on top of these models, implementing analytics or visualization tools  [18]. These softwares can take the form of R libraries  [18] , or compiled Java  [41]  programs with a graphical user interface. While making these softwares freely available is an important step to foster methodological research, they do not seem to be easily extensible or interoperable as they do not provide documented APIs to build new software upon it. Besides, the process of transforming an existing database in order to conform to such standards is costly, as it requires to build complex mappings between shared representations expressed through highly heterogeneous codes from one information system to the other.
In the case of the SNDS database considered in this work, such a mapping is still work in progress  [12]. In other fields, web-scale analytics have shifted from the use of normalized SQL databases towards NoSQL technologies relying on distributed computing, denormalization, and columnar storage. The use of distributed computing allowed gains in computational power using low cost, commodity servers instead of expensive dedicated hardware  [8]. A work from OHDSI  [37]  compared the ACHILLES software (R  [38] , PostgreSQL  [36] ) with Apache Spark  [49]  using common SQL requests. They observed performance gains for Spark even on a single server or small clusters, at the exception of requests leading to large network I/O, since such operations are known to be the slowest operations in a distributed computing framework because of network latencies and throughput.
It can create bottlenecks when many data chunks are sent across the servers in the cluster to perform a join or a groupby operation (leading to so-called shuffles). Denormalization can be a way to circumvent this issue by performing a set of join operations beforehand, once and for all  [48, 23, 11] , reducing join operations to simple look-ups over a very large table. The data duplication resulting from such joins operations might lead to storage issues, which can be mitigated with the help of columnar storage formats  [23, 26]  using compression strategies. To the best of our knowledge, such an approach has not been implemented to perform ETL on large health databases. Prior works are either relying on SQL and normalized schemas  [19, 31]  or applied to small datasets  [14]. This paper describes and implements such an approach for large health databases, as explained in the next section.
Material and Methods This work focuses on (i) denormalizing the data in combination with columnar storage and distributed computing to perform concept extraction, (ii) providing a structured and re-usable concept library, and (iii) introduce useful abstractions to handle cohort data. Scalability issues are handled by (i), while (ii) and (iii) foster the reuse of code and knowledge across studies. This is achieved by reducing both studyspecific code and database entry barriers by providing ready-to-use concepts. SCALPEL3 provides Scala  [30]  and Python APIs to ensure easy extension and interoperability with numerous libraries. All the code supporting this paper is open source and freely available. This paper is not about data integration from disparate sources, such as multiple EHR systems, but rather about an ETL based on batch distributed processing of a large, centralized claims database.
The SNDS database This work was performed using the Système National des Données de Santé (SNDS), a large claims database containing pseudonymized data on 98. 8% of the French population (66 million patients in 2015)  [46, 7]. It contains time-stamped information about medical events leading to reimbursement (see Table  1  in  [46]  for an exhaustive list of available data) in the last 3 years 1. It contains more than 20 billion health events per year, representing roughly 70TB of data. SNDS is composed of multiple "sub-databases", each one with a star schema. The central table records events leading to cash flows that need to be joined to many other tables to access medical information 2. In this form, retrieving patient information for statistical studies is very costly in terms of computation and expert knowledge: targeted data can be spread across multiple databases, tens of tables, and hundreds of columns, and its identification requires a deep administrative knowledge of the French health-care reimbursement mechanisms.
SCALPEL3: a SCAlable Pipeline for hEaLth data SCALPEL3 is based on Apache Spark  [49] , a robust and widely adopted distributed in-memory computation framework. Spark provides a powerful SQL-like high-level API and a more granular API to perform data operations. It can be coupled with the Hadoop File System (HDFS)  [43]  replication system to accelerate large files reading and distribution over a computing cluster. SCALPEL3 is an open-source framework organized in the following three components. [22]  denormalizes the data "once and for all" to avoid joining many tables each time the data of a patient is accessed. Its input is a set of CSV files extracted from the original SNDS database. [34]  defines concepts extractors that process the denormalized data and transformers, that compute more complex events based on extractors output.
SCALPEL-Analysis As SCALPEL-Flattening and SCALPEL-Extraction perform batch operations, they need to read (resp. write) input (resp. output) data from the file-system (local or HDFS). They are implemented in Scala in order to access Spark's low-level API and take advantage of functional programming and static typing, resulting in rigorous automated testing (94% of the Scala code is covered by unit tests). Both can be configured through textual configuration files or be used as libraries. SCALPEL-Analysis is a python module implemented in Python/PySpark and designed for interactive use. It can be used in a Jupyter notebook  [21]  for instance. This workflow is illustrated in Figure  1.
SCALPEL-Flattening: denormalization of the data As mentioned earlier, performing data analysis on SNDS patients' health requires many joins and can consequently be extremely slow. To circumvent this issue, the data are denormalized by joining the tables sequentially to obtain a big table in which each line corresponds to a patient identifier and a wide representation of an event. Denormalizing a star-schema database results in a really big table due to values replications. To circumvent storage and computation issues, the denormalized data is stored in Parquet  [3]   open-source columnar storage format implementing Google's Dremel  [26]  data model. Parquet is wellintegrated in the Spark ecosystem  [4] , allowing us to take advantage of the columnar storage in terms of data compression and query optimization.
SCALPEL-Flattening first converts the input CSV files containing exports of SNDS tables to Parquet files. Then, it recursively performs left joins with these tables, starting with the central table. Finally, it writes the results in a single Parquet file. To ensure the scalability of these big join operations, the input data can be automatically divided with respect to some time unit (such as years, months) before performing the join operations. In this case, the joins results are sequentially appended to the output parquet file. These operations are repeated for each SNDS sub-databases. The size of the temporal slicing used in the joins, the schema, and the joining keys can be tuned by the end-user through a configuration file, which defaults to the denormalization of tables containing only medical data (as opposed to econometric and administrative data).
SCALPEL-Extraction: extraction of concepts SCALPEL-Extraction provides fast extractions of medical concepts from the denormalized tables produced by SCALPEL-Flattening. By providing ready-to-use medical events, SCALPEL-extraction encapsulates SNDS technical knowledge but keeps medical data as raw as possible, so that end-users have access to fine-grained data which is critical when designing observational studies  [47, 16]. The extracted concepts are organized around two abstractions: Patient and Event. The Patient abstraction has a unique patientID, a gender, a birthDate and eventually a deathDate. The Event abstraction allows to represent any event associated to a patient. It can be punctual (e. g. , medical act) or continuous (e. g. , hospitalization). All concepts are automatically extracted into Patient or Event objects by a set of Extractors and Transformers, designed to fetch the data in the relevant tables and columns of the SNDS Sources.
Wide denormalized table Step 2: filter out null values A Transformer is based on specific algorithms requiring multidisciplinary knowledge from epidemiologists, statisticians, clinicians, physicians, and SNDS experts  [46]. Transformers usually combine events built by Extractors to build more complex events, such as computing drug exposures from timestamped drug dispenses. Extractors and Transformers can be used through a Scala API or controlled using a textual configuration file. Many Transformers used in several studies such as  [27, 29]  are implemented and ready to use. Besides Parquet files containing extracted events, SCALPEL-Extraction outputs metadata tracking the data used to build each type of extracted events. This file can be leveraged by SCALPEL-Analysis to build Cohorts and flowcharts, as explained below.
SCALPEL-Analysis: interactive manipulation and analysis of cohorts While SCALPEL-Flattening and SCALPEL-Extraction are implemented in Scala/Spark for performance and maintainability, SCALPEL-Analysis is implemented in Python/PySpark  [49]  since it is designed for interactive environments, such as Jupyter notebooks  [21]. SCALPEL-Analysis eases the manipulation and analysis of cohort data. It is based on the following abstractions: The Cohort abstraction is a set of Patients and their associated Events in a [startDate, endDate] time-window. Basic operations such as union, intersection, and difference can be performed between Cohorts, while a human-readable description is automatically updated in the results. More granular control is kept available through accesses to the underlying Spark DataFrames (using Spark DataFrame API).
This combination allows easy data engineering and fine-grained, yet reproducible, experiments. The CohortCollection abstraction is a collection of Cohorts on which operations can be jointly performed. The CohortCollection has metadata that keeps the information about each Cohort, such as the successive operations performed on it, the Parquet files they are stored in and a git commit hash of the code producing the extraction from the Source. International guidelines  [5]  regarding studies based on LODs insist on the explanation of cohort construction to highlight eventual population biases, motivating the following CohortFlow abstraction. The CohortFlow abstraction is an ordered iterator defined as the following left fold operation foldl(c : CohortCollection, ∩) := (((c 0 ∩ c 1 ) ∩ c 2 ) ∩.
Flowcharts can easily be produced to track how many subjects were removed at each stage. Flowcharts can be produced either from a CohortFlow, or the metadata tracking the data extraction process produced by SCALPEL-Extraction. Examples are provided in Supplementary Material. SCALPEL-Analysis also provides tools producing datasets in formats compatible with popular machine learning libraries. At the core of these tools is the FeatureDriver abstraction. The FeatureDriver abstraction is used to transform Cohorts into data formats suitable for machine learning algorithms, such as numpy. ndarray [20] , tensorflow. tensor [1]  and pytorch. tensor [33]. It is mainly a transformation of a Spark dataframe representation into a tensor-based format. FeatureDrivers perform several sanity checks, such as time-zone and event dates consistency, and can be easily extended by end-users, thanks to the PySpark API.
Results Scaling experiments presented in this section were performed on a SNDS subset containing 13. 7 million patients followed up to three years described in Table  1  containing common data types (timestamps, integers, floats, small strings), normalized according to the SNDS data model. The testing data consisted in outpatient data (DCIR) and inpatient data excepted home hospitalization, rehabilitation centers and psychiatric hospitals (PMSI-MCO). Raw data was extracted from the SNDS by CNAM, the French agency that manages this database. Extracts were dumped on the testing cluster as a set of CSV files. SCALPEL3 was tested on a Mesos  [15]  cluster of commodity servers with 14 worker nodes driven by 4 master nodes. Worker nodes resources amount to 224 2. 4Ghz logical cores, 1. 7Tb of RAM, and 448Tb of storage distributed over 88 spinning hard drives.
These resources are shared over the cluster by HDFS  [43]  for data storage and by Spark for memory storage and computations. This cluster and the configuration of the jobs were not fine-tuned for the usage of SCALPEL3, but follow standard guidelines for cluster configuration for distributed computing with Spark. Denormalizing this dataset using SCALPEL-Flattening took about 6 hours using the 14 worker nodes. During the conversion of CSV tables to parquet files, worker nodes CPU and memory usage are maxed out on most worker nodes. During the join operations, resource usage is first dominated by network I/O to shuffle the data across the workers, followed by an increase in CPU and memory usage reaching two-thirds of the cluster capacity. Note that the current framework used for SNDS data cannot handle such denormalization so that there is no element of comparison for SCALPEL-Flattening with it.
SCALPEL-Extraction was evaluated on the following extraction tasks, that correspond to typical events required for public health research studying relations between fractures and some drug exposures: (a) extraction of patient demographics (gender, age, eventual date of death), (b) extraction of drug dispenses, (c) filtering of patients w. r. t their first date of drug use (prevalent drug users, 65 drugs), (d) computation of drug exposures based on drug dispenses dates, (e) extraction of reimbursed medical acts, (f) extraction of diagnoses, (g) identification of fractures using the algorithm described in  [9]  based on medical acts and diagnoses. Indicative baseline performance was established by executing similar queries on the current SNDS infrastructure, based on SAS Enterprise Guide for analytics  [40] , connected to an Oracle SQL database hosted on Oracle Exadata servers  [32].
This baseline performance was computed with a single run, as the current SNDS framework is designed to allocate resources dynamically each time a new query is submitted. The monitoring of resource usage on this SAS-Oracle infrastructure is not straightforward, since computations are divided between SAS and Oracle jobs, and since the resources of the Oracle Exadata infrastructure are divided across servers focused on storage or computation. At peak use (for task (c)), the Oracle job was using 10 CPUs supported by 4. 9GB of PGA memory, while SAS was using 1 to 6GB of RAM. An assessment of the horizontal scaling of SCALPEL3 is performed by varying the number of executors (4 logical cores and 25 GB RAM) to perform these queries. All the results are displayed in Figure  3. SCALPEL-Analysis aims at providing useful abstractions to ease cohort data manipulation.
Discussion SCALPEL-Extraction reaches performances similar to SQL-SAS based SNDS framework when using 6 executors (Figure  3 (h)). It is consistently faster on tasks involving large data volumes or complex operations such as tasks (b), (c), (d), and (g). On the other hand, tasks involving the PMSI-MCO database (tasks (e) and (f)) exhibit poor performance. This is rooted in the flat table structure as PMSI-MCO is not sparse-by-block like DCIR (see the difference in the ratio of Rows in central table w. r. t. denormalized table in Table  1 ). It results in performing more tests on row values and data shuffle than necessary when performing queries on PMSI-MCO. Performance on these tasks could be further improved by slightly modifying the join strategy in the flattening step to ensure PMSI-MCO sparsity by block.
The cost of data denormalization should be considered to be fixed as this operation is done once and for all. The denormalized data can then be updated incrementally when new data are fed into the cluster (typically a few times a year). SCALPEL-Extraction scales almost linearly from 4 to 16 executors. The scaling gains then slow down, reaching peak performance at 28 executors (see Figure  3 ). These diminishing returns can be caused by the cluster resource sharing between storage services (HDFS) and computation (SCALPEL3). As a result, SCALPEL3 resource usage can be in conflict with HDFS resources as soon as the number of nodes used by SCALPEL3 excess one-third of the cluster 3. Splitting the cluster nodes between storage nodes and computation nodes could improve horizontal scalability. Note that for very small tasks (such as (c), (d), (g)), runtime is dominated by I/O operations and do not benefit particularly from additional CPUs.
Besides performance considerations, note that SCALPEL3 uses only open-source, free software and runs on commodity hardware, which is likely cheaper than Oracle Exadata servers and easier to scale if (h) All tasks the data volume increases: a Spark cluster easily scales "horizontally" by adding more nodes. The performance comparison between the two infrastructures is limited by (i) the impossibility to set the resources used by SAS-Oracle beforehand for these experiments does not allow for multiple runs and (ii) slight differences in query implementation caused by design differences such as columnar vs row orientation. Nonetheless, it shows that SCALPEL3 can be used as a viable open-source alternative running on commodity hardware while benefiting from horizontal scaling on very large jobs.
Besides, SCALPEL3 greatly improves the maintainability, audit, and reproducibility of studies using SNDS. First, continuous integration of code updates and large code coverage (94%) with unit testing is a big improvement in terms of maintainability over copy-pasted SQL snippets. Secondly, SNDS expertise encapsulation for events extraction is fully tested and maintained in SCALPEL3, so it eases extraction algorithms reuse for studies and lowers the entry-barrier to SNDS. Obviously, design and maintenance of SNDS concept extractors by a team of developers and SNDS specialists is a mandatory task, as the database contents are constantly evolving. Moreover, the relevance of extracted data (to answer a trade issue) requires some SNDS knowledge and is the responsibility of the user. The combination of expert knowledge encapsulation (SCALPEL-Extraction) and interactive cohort manipulation (SCALPEL-Analysis) results in smaller and more readable user-code, leading to easily shared and reproducible studies, supported by data tracking and automated audit reports.
Finally, SCALPEL3 allows producing datasets compatible with several Python machine learning libraries formats, fostering methodological research on SNDS data, which was not possible with the proprietary software that is currently used. The choice of the Python language might help SCALPEL3 adoption among the data science and machine learning community, while it might hinder its use among public health researchers who are traditionally using proprietary statistical softwares or the R language. SCALPEL3 can be used in standalone mode  4  or in distributed mode  5  when working on large datasets. The knowledge and skills required to manage a computing cluster are not yet widespread which could also impede a large adoption of the distributed mode among small organizations. Finally, while SCALPEL3 does not support international data standards yet, the development of vocabulary mapping tables in France was anticipated so as to ease future support of data standards such as OMOP-CDM  [39]  or FHIR  [6]  to SCALPEL3.