dataesr/scientific-paragraphs-categorization huggingface dataset:
This paper is organized mainly following the data flow through the EventIndex components. "Requirements and Global Architecture" describes the use cases and the overall system architecture; "Data Production" and "Data Collection" explain the selection of datasets to index, the indexing method and the index data collection components; "Data Storage in Hadoop and HBase" and "Data Storage in Oracle" describe the current data storage methods in Hadoop [11], HBase  [12]  and Oracle. "System Monitoring" and "Operations and Performance" cover system operations and monitoring. Finally, "System Evolution" outlines the developments towards a higher performance system for LHC Run 3 and beyond.
Several other use cases can be served by a complete event catalogue. A second group of use cases are related to data quality assurance. Production consistency and completeness checks can be run, for example counting the number of events in the input and output datasets for any processing task and making sure that all events have been processed, and there is no data duplication in the output datasets. Running indexing jobs on all produced data provides in addition a check that all output files are stored correctly on disk and are available for further analysis. Further use cases are related to the calculations of overlaps among trigger chains within a given dataset, and and among derived datasets. With the complete trigger information for each event stored in the catalogue, it is possible to count the number of events that satisfied each trigger chain and also measure the overlaps among trigger chains.
Similarly, it is possible to select particular events on the basis of combinations of their properties, such as the trigger or the instantaneous luminosity (interaction rate), and then count their occurrences or retrieve them directly for full analysis. Not all event processing tasks output the same number of events they had in input. The derivation procedures take all fully reconstructed events as input and output only the selected events that are useful for one or a few particular analyses. As there are almost 100 derivations that run on ATLAS events, it is useful to have the possibility to check the amount of overlaps between derivation streams, in order to reduce if/when possible their number and hence the processing time and the disk space for the output datasets. Differently from the trigger stream overlap checks that are done within a specific dataset, the derivation overlap check involves a number of different but related datasets; related in the sense that all these derived datasets must have been produced from the same parent one.
Location information The GUID of the file that contains this event and the internal pointers within that file, for the file that is currently indexed and also for the upstream files in the processing chain (provenance). The GUID can be passed to Rucio to identify, locate and retrieve the file containing a given event in order to extract it or analyse it directly. The provenance record is useful to reduce the number of datasets that have to be indexed; for example, the pointers to the RAW datasets can be obtained by indexing the corresponding AOD datasets.
Performance Requirements The catalogue must sustain a record ingestion rate that is at least as large as the real data production rate (1 kHz during LHC Run 2), plus the simulated data processing rate (about the same when averaged over a year). Given the foreseen increase of trigger rates by the end of LHC Run 3 to over 3 kHz and the corresponding increase in simulation production computing power, the catalogue for LHC Run 3 needs to withstand an ingestion rate of 10 kHz at least, allowing for some contingency in case of operation backlogs. The query rate is very small compared to the ingestion rate, but all data are equally important and all queries are different, so it is necessary to have a flat internal structure and caching does not help much. User queries for small data samples are done through client code with a command-line interface, or through web services, requiring that the response times for simple queries be compatible with human response expectations (below 1 s); queries for large amount of data or computations of global counts, trigger overlaps or derived dataset overlaps can be executed as batch processes but need to return their results within (roughly) an hour, and never fail, in order to be useful.
System Architecture The information flow through the EventIndex system is linear, so it was natural to match the system architecture to the data flow  [10]. One needs first to extract the relevant metadata from the event data files and store them in a central store, which client programs can query to perform their tasks. The EventIndex system is therefore partitioned into a number of components: -Data production. This component takes care of extracting the metadata from each data file as soon as it is produced at CERN or on any of the ATLAS Grid sites, format them for transfer to the central store and send this information to CERN. -Data collection: This component deals with the data transfer infrastructure, the metadata completeness checks for each dataset, assembling the information produced by all files in a given dataset and formatting it for storage, including decoding the trigger information and presenting it in an optimized format for fast searches.
-Data storage. This is the core system. It includes the setup of the EventIndex data storage cluster in Hadoop, the code to import the data and internally index them, and the web service providing the command line and graphical interfaces for the clients. As a subset of the EventIndex data are also replicated to an Oracle database for access performance reasons, this component includes also the support for the Oracle store, the data import code and the graphical interface for the users. -Monitoring. All servers and all processes have to be constantly and automatically monitored. This component collects, stores and displays the relevant information, and sends automatic alerts in case of service interruptions or malfunctioning. Regular functional tests are also submitted in order to monitor the performance for the most common use cases.
Dataset Selection As soon as new ATLAS data are processed on the CERN Tier-0 cluster  [15]  and the corresponding AOD datasets are available, jobs are launched to extract the EventIndex information for all "physics" datasets; calibration, test and monitoring streams are excluded. From this indexing step, information on the location of each event in RAW format is also extracted. Many more ATLAS datasets are produced using the ATLAS resources of the WLCG Grid, and other additional resources that can be available from time to time. They include the whole simulation chain, from event generation to detector simulation, then event reconstruction and selection for analysis; real events are also re-reconstructed from time to time on the Grid, and all analysis selections also take place in a distributed fashion.
All AOD datasets are indexed, and for real data all types of derived AODs (DAODs) are indexed too. For the simulated data only some types of DAODs are indexed, if requested by the analysis groups that use them. In addition, all EVNT datasets are also indexed. Datasets produced on the Grid are selected for indexing according to the following criteria, based on information obtained from the ATLAS metadata database AMI  [16] : -The dataset is marked in AMI as complete and validated for use in analysis or further processing. -The dataset is marked as long-lived, to avoid indexing transient datasets that are only used in internal steps of the production procedure. -The dataset is part of a regular production processes and not just used for checks or validations of the software or the trigger configurations.
Indexing Job Submission Datasets that were selected for indexing have to be processed by the Production and Distributed Analysis system (PanDA)  [17]. PanDA takes the list of new datasets and generates jobs that run a predefined "transformation" (a script containing a sequence of algorithms to be executed on a data file  [18] ) on the WLCG Grid. The transformation used and its configuration in general depend on the data format of the dataset and the type of data (simulated or real); for example, the trigger information, which constitutes a large fraction of the EventIndex data, is collected for each event only from datasets in AOD format as it will not change with subsequent processings of the same event. The progress of these jobs can be monitored through a dedicated dashboard; if necessary jobs can be aborted or rerun.
Producer Transformation The Producer is in charge of extracting the EventIndex information from the actual input files, store it into temporary files and send them to a central location at CERN. It has to be able to run using the ATLAS production infrastructure on all available production facilities (the Tier-0 cluster at CERN and the WLCG Grid), so it is implemented in a way very similar to standard ATLAS data processing programs using the ATLAS transformation framework running within the Athena software framework  [19]. Python was chosen as the implementation language for the Producer code, as it only accesses the header records of each event. The python interfaces to the Athena classes methods written in C++ do not change between releases, so the Producer code can be rather stable. The producer input can be one or several files in the ATLAS specific ROOT format  [20, 21] , such as those in AOD, DAOD and EVNT datasets.
The EventIndex transformation class implements all required methods by the Athena framework to initialize a job, execute the event loop (process the events) and finalize the job. The current implementation runs using a serial processing model, so the input data structure (file and event ordering) is preserved without needing further post-processing. The EventIndex producer has two separate steps, both running within the transformation. In the first step, it reads events (execute method), extracts information and saves the relevant information into a temporary file. When all events are read, the second step starts (finalize method), in which the output file is transferred to a central store at CERN. Besides the EventIndex information itself, some additional environment and processing information is stored: the PanDA task and job identification, input dataset name, total number of files and events, starting and ending processing times as well as identification (GUID) and number of events for each file read.
Index Record Format Two different file formats were used to store the output from the producer, adapted to the specific needs of the processing. Initially a SQLite3 [29] format was reused from other Athena tools, with data stored in key:value pairs using only one table with two columns, "key" (TEXT) and "value" (BLOB). "Value" is the serialized representation of a python object using cpickle, so arbitrary objects can be saved and retrieved into the database, allowing a flexible "blackboard" style storage of key:value pairs. The producer transformation used several key: value pairs to store general information like the number of files and events processed, date and time of processing, job and task identification, file GUIDs, input collection name (dataset name), etc. Events were saved consecutively as an ordered tuple with key "Entry N" where N is the entry number.
The file can contain six different message types: Header, Trailer, BeginGUID, EndGUID, TriggerMenu and EIEvent: 1. Header: contains global information about the processing step, like task and job identifications, input dataset name and start processing time. 2. Trailer: contains global information collected during the processing, like number of files read, total number of events and end processing time. 3. BeginGUID: marks the start of a new input file being processed. These messages contains the input file unique global identifier (GUID), and the start processing time besides some other ATLAS metadata information like the processing version, the stream and project names. 4. EndGUID: marks the end of the input file processing. It contains the number of events read for this file and the end processing time.
Supervisor The Data Collection Supervisor is in charge of tracking all data collection steps. It validates the produced data and informs the consumers about the presence of data to be transferred into the Hadoop cluster at CERN. It also allows following the indexing progress of datasets and containers thanks to its web interface. The Supervisor receives messages sent by each producer job with information about what has been processed. This information includes among other things: the dataset name, the task and job identifiers, the location of the produced object store, the GUIDs of the processed files, and the number of events and the number of different event identifiers processed per file. The supervisor collects this information, and it is thus able to know which datasets are being indexed by which tasks in which system.
As part of the file and dataset metadata information, Rucio stores the number of events that they contain. This information is used to track the progress of dataset indexation. Furthermore, since each job sends the number of events that it has processed per file, this information can be compared against the one provided by Rucio to identify possible inconsistencies. The supervisor also retrieves the information of the indexing tasks before they achieve a final state. To do this, the supervisor has to contact and decode the information provided by two different monitoring systems: conTZole  [32]  if the tasks is running in Tier-0, and PanDA Monitoring  [33]  if the task is running through PanDA on the WLCG Grid. Both monitoring systems provide information on the progress of tasks, like the number of jobs, number of events processed, status of the task and jobs, etc.
The consumers are informed through the messaging system about the validation objects that they should consume; they first retrieve the validation objects, process them retrieving from the Object Store the information that should be consumed and put into Hadoop. When all data have been consumed, they notify the supervisor about it, signalling once again how many events have been consumed. With this last message, the supervisor can mark the dataset as indexed. Inconsistencies in the number of events, unprocessed files, lost or delayed messages, can be identified thanks to this validation procedure. A dataset that is not validated is kept in a validation queue, where the validation will be retried in after receiving possible delayed messages. With all these pieces of information from different systems the supervisor is able to: -Monitor the progress of the tasks that index each dataset; -Identify failed production tasks; -Declare obsolete indexing tasks that have problems and are going to be replaced by other tasks; -Detect if any messages were lost; -Identify inconsistencies between the processed files and the information retrieved from Rucio; -Complete missing pieces of information in Rucio in the rare cases when they occur; -Notify, through email if needed, about datasets that have duplicated event numbers detected at job level, i.
Consumers The Consumers are in charge of storing the EventIndex data in the final data store. They run centrally at CERN and in the current system there are as many consumers as messaging brokers, as this is sufficient for the current production rates. They are stateless independent entities that can be scaled up in case of necessity. Consumers wait for validation messages from the Supervisor, containing references to the actual EventIndex data objects to be ingested. These objects are read from the Object Store with data encoded with Protocol Buffers  [30]  format. The data are then formatted for the current production schema, and stored in Hadoop files. These files are organized in directories named after each dataset container, and the current granularity is to write a file per dataset, but this is also configurable in the validation object.
Each file contains data organized by key containing RunNumber-EventNumber, and its related value encoded in a CSV schema with all the event information. Information about the status of the processing is communicated back to the Supervisor, starting with the acknowledgment of the request. When all the objects are consumed and the file is written, the result is sent back with a control message again. In case of any failure, details are included. It must be noted that the granularity of the validation data can vary from a single object reference, to thousand of them belonging to a particular dataset. This procedure is repeated for all validation messages produced for a dataset container. At this point the validation of the complete dataset is possible, and a different control message will trigger the final validation of a dataset container.
This procedure writes a control text file in the Hadoop file system, containing all the URLs of the individual dataset files that were validated. This file is used by Oracle ("Data Storage in Oracle") to know which data have to be imported. An individual Consumer typically processes a mean of 15 kHz (events processed per second), and we have observed a maximum of 28 kHz. The current implementation of the Consumer is a multi-threaded Java program, with thread pools using the Future pattern  [34]  in order to save resources, and exploit parallelism among internal data access and transformation tasks. The setup of the Hadoop writing channels and the input/output largely dominates the processing time, with the CPU used on data mangling and schema transformation taking a small percentage of the time.
Data Import to MapFiles and Copy to HBase The Consumers ("Consumers") write in Hadoop one sequential file for each dataset. These files are converted into Map-File format and copied into their dedicated space. All those MapFiles are then registered in the Catalog, which is implemented as an HBase table. After each successful dataset import, the event information (without trigger record, for space and performance reasons) is uploaded to an auxiliary HBase table for fast event lookup operations. All import and search operations are also registered in the system journal, together with all relevant information. The journal is also implemented as an HBase table. The full data flow within the Hadoop system is shown in Fig. 3.
Trigger Decoding The trigger record for each event is transferred as a bitmap, where each bit corresponds to a trigger chain. In order to store the trigger data in Hadoop in an easily searchable and retrievable way, it has to be decoded with the help of the trigger mask for the given dataset, which in turn can be retrieved from the trigger database using the trigger key (SMK) of the dataset. The trigger tables are available in different databases: the COMA (COnditions MetadatA) database  [39]  contains all trigger information for real data and the MonteCarlo Trigger DB (TriggerDBMC) in Oracle contains the data for MC simulation. The EventIndex replicates the trigger tables from COMA and TriggerDBMC to HBase tables and then uses them for trigger decoding  [40]. If the SMK is absent in the event record, it is possible to obtain it from the run number for the real data and from the reconstruction tag for MonteCarlo simulation.
Derived Statistics and Correlation Tables After the dataset import and successful verification, several derived tables are created automatically: -The dataset overlap table contains the numbers of common events between different datasets in the dataset derivation chain. -The trigger overlap table contains for each dataset the number of trigger chain pairs which were fired simultaneously. -The trigger statistics table contains for each dataset the number of fired trigger chains of each type. While this table is created separately, it can be seen as a subtable of the trigger overlaps table. All derived tables can be interrogated with the standard tools because they are implemented as normal MapFiles. Overlap tables can be also visualized as Graphs (see Fig. 5 ).
Command Line Interface Several commands were implemented to give access to the stored data: -Catalog (catalog) to search and modify Catalog entries. -EventIndex (ei) to search all datasets using either direct searches or complex Map/Reduce jobs. The EventIndex command allows the use of any legal Java code as a search or result clause. -EventLookup (el) for fast search of the physical datasets corresponding to an event (specified as a pair of run number and event number). -TriggerInfo (ti) to perform search and analyses of the trigger information. -Inspector (inspect) to see the actual content of a Map-File. Commands to access the EventIndex store are available directly in the CERN Hadoop cluster for the data The EventIndex information in the Hadoop file system (HDFS) contains the trigger key for each event, which is used to retrieve the copy of the relevant trigger table stored in HBase.
Event Lookup or "Event Picking" Event lookup is the most important and heavily used functionality. It returns the GUIDs and optionally the stream type, dataset name and other parameters for user specified sets of real or simulated events, identified by run and event numbers. The search can be narrowed by specifying the trigger stream, data format and version. By default, the lookup is performed in the HBase table, the best performing back-end. It is also possible to run identical queries against the MapFiles from which the data are ingested into the HBase table, as described in "Data import to MapFiles and copy to HBase", and which were historically the first event lookup implementation.
Comment on Free and Open-Source Software When it became necessary to convert all MapFiles to the block compressed SequenceFile format (see "Compression of MapFiles"), it turned out that random access queries were not working on the block compressed files. Thanks to the availability of the source code, we were able to fully investigate the issue and track it down to a bug in a Map-File method. Having resolved the problem and incorporated the patched version into the EventIndex, we contributed the patch to the Hadoop project; the patch was accepted and we also had a chance to learn the Hadoop project practices. The Hadoop project software is released under the Apache License 2. 0 [44], which is a free software license according to the Free Software Definition  [45] , and this turned out to be a crucial point.
Data Storage in Oracle The initial implementation of the EventIndex store in Hadoop showed several important shortcomings by the end of 2015, the first year of LHC Run 2. With the versions of Hadoop and HBase and the hardware setup provided by CERN that we could use at that time, all queries became substantially slower as the amount of stored data increased. Simple event lookup queries for 10 events started taking over 1 min instead of the expected sub-second response, and counting events across large datasets (100 million events) took tens of minutes. In addition to optimizing the Hadoop cluster setup, it was then decided to explore the possibility of storing a subset of the real data information in an Oracle database, exploiting this well-known technology to support the most important and time-consuming use cases, primarily event picking for real data events.
Data Structures A relational model was found to be well suited to the task  [46]. The simple relationship of datasets to events lends itself to a very simple relational table structure as shown in Fig. 6. The two leftmost tables (in blue) store each indexed dataset and its events, respectively, driving the core functionality of the system as well as serving some secondary use cases. In ATLAS, datasets are uniquely identified by a string concatenating 6 fields, each of which is stored as a separate column in the Datasets table :.