The ATLAS EventIndex provides a global event catalogue and event-level metadata for ATLAS analysis groups and users. The LHC Run 3, starting in 2022, will see increased data-taking and simulation production rates, with which the current infrastructure would still cope but may be stretched to its limits by the end of Run 3. This talk describes the implementation of a new core storage service that will provide at least the same functionality as the current one for increased data ingestion and search rates, and with increasing volumes of stored data. It is based on a set of HBase tables, coupled to Apache Phoenix for data access; in this way we will add to the advantages of a BigData based storage system the possibility of SQL as well as NoSQL data access, which allows the re-use of most of the existing code for metadata integration.
The ATLAS Collaboration [
Investigations on several structured storage formats for the main EventIndex data to replace
the Hadoop MapFiles [
HBase [
While updating the core storage system, all other components have to be revised and if
necessary updated or replaced. In particular, the Data Collection system and the Supervisor [
The main motivation for the further evolution of the Data Collection system is the usage of
modern storage like HBase/Phoenix, and data processing technologies like Spark [
The Producer implementation is currently done in python with a single thread. It will be
upgraded to work with the latest data analysis software and external libraries like stomp.py [
The CERN Object Store [
The Supervisor will be improved in several areas:
Ability to control the creation of indexing tasks: The current supervisor discovers indexation
tasks sent by separate data production processes directly to the Grid workload management
system PanDA [
Increased granularity in the indexation process: Sometimes the indexation tasks fail because a few files of a dataset were unavailable. In this case the task is now discarded and the whole dataset is re-indexed. The new supervisor will be able to create sub-datasets that contain just the files whose indexation has failed, avoiding the re-indexation of the whole task.
Increased granularity of the data consumption: The current supervisor validates the dataset once it is completely indexed and then signals the consumers to start its consumption. This approach is inefficient for large datasets in case the ingestion process is interrupted due to service problems or input file corruption and a process that may have been running for several hours has to be restarted. The new supervisor will be able to split the information consumption into smaller chunks.
The Consumers can currently write to Phoenix as standalone processes, but in the new
approach they will be converted to Spark [
The calculation of duplicates, overlaps or other analytic jobs over larger amounts of data will be done offline with Spark jobs over the Hbase/Phoenix stored data.
HBase works best for random access, which is perfect for the event picking use case where we want low-latency access to a particular event to get its location information. Use cases where we need information retrieval (trigger info, provenance) for particular events are served by fast HBase gets. In addition, analytic use cases where we need to access a range of event information for one or several datasets (derivation or trigger overlaps calculation), can be solved with range scans on these data. They can be optimized with a careful table and key design that maintain related data close within the storage, reducing access time.
HBase is a column-family grouped key-value store, so we can benefit from dividing the event information into different families according to the data accessed in separated use cases; for example, we can maintain event location, provenance, and trigger information in different families.
Apache Phoenix is a layer over HBase that enables SQL access and provides an easy entry point for users and other applications. Although HBase is a schema-less storage, Apache Phoenix requires a schema and data typing to provide its SQL functionalities.
Best performance is gained from using row keys. We are going to have several billions of entries and we want single row access and scans to be as efficient as possible, so we need to include the most needed information in the key and leave other information in column families. Pure valuebased access is always a full scan, so all the index information should be in the row key for better performance, while maintaining the row key size to its minimum. A representation of the schema of the events table can be seen in Figure 1.
An event record is uniquely identified by its dataset name: Project.RunNumber.StreamName. prodStep.dataType.AMItag_[tidN] and its EventNumber. To satisfy all use cases we need to access different information: ‒ ‒
Event picking: needs to know the EventNumber and which dataset to get the event from. Event selection based on trigger: needs to know the dataset name and selection criteria. ‒
Derivation overlap: needs to run over all the derivations for datasets having the same dataset name except for the dataType.
Therefore we include this information in the row key chosen as a composite value dspid.dstypeid.eventno.seq (16 bytes), where: ‒ dspid (Integer: 4 bytes) is an identifier for the dataset name, excluding the dataType. ‒ ‒ ‒ ‒ ‒ ‒ dstype (Integer: 2 bytes) is an identifier for the dataType. eventno (Long: 8 bytes) is the event number.
‒ seq (Short: 2 bytes) is used to deduplicate event entries when the EventNumber collides.
The dspid is generated by the Supervisor during data ingestion. Generating monotonically increasing values is not good for HBase, as it can create hot spots and not distribute the load among regions. This can be solved by reversing the bit order (which has the property to distribute and cover the whole key space from the beginning).
The dstype allows scanning for the datasets (having the same dspid) over all the data types, which is the use case of dataset overlaps computation. It is internally computed into dataTypeFormat (5 bits = 32 values) and dataTypeGroup (11 bits = 2 048 values) for optimal usage.
The seq is computed as the crc16 [
Families represent related data that is stored together on the file system. Therefore, all column family members should have the same general access pattern. The current defined families are: A: Event location (and MC info), B: Event provenance (from the processing history), C: Level 1 trigger (L1), D: High Level Trigger (L2 and EF for Run 1 data and HLT from Run 2 onwards).
In the Event Location family, we store information of the location of the event. It includes the tid (production task identifier) of the original dataset, allowing easy dataset deletion and ingestion crosschecks. It also includes the self-reference of the physical location of the event, as a 24-byte binary value, with 16 bytes representing the GUID identifier of the file that contains the event, and the OID1 and OID2 fields with 4 bytes each.
In the Event Provenance family, we store the chain of processing steps for this event. It is very similar to the previous self-reference field, but adding stream information using dataTypeFormat and dataTypeGroup. It is an array of records, where each record is represented as a 26-byte binary value (2 bytes for the dataType, and 24 bytes for the reference).
The other families maintain the trigger information, which comprises the bulk of the payload, and are divided into Level 1 and High-Level Trigger (EF or HLT) plus Level 2 for Run 1 data, including lb (luminosity block) and bcid (bunch crossing identifier).
We need other tables to keep the dataset generated identifiers and bookkeeping data, as well as other related information. Currently we use these tables:
Datasets table: stores information for quick dataset location, and related generated identifiers like the dspid. It also contains bookkeeping information relative to the status of the dataset during importing or ingestion phases, including the start and ending times. In addition it contains metadata information like the number of events, including computed information like the number of unique or duplicated events.
Data types table: stores information and numerical identifiers about the data type formats and data type groups for derivations, as they will be used in the dstype field in the row key, and other fields in the data location and provenance families.
The current production data has to be imported to Hbase/Phoenix with the new defined schema, so tools have been developed to perform this task when the new system enters production. The current tools can use Map/Reduce or Spark jobs to do the data loading, conversion, and writing the data in the Events table. They also write bookkeeping information of the import process in the Datasets table. Loading and check-pointing can be done at the individual file, dataset or container level, or even complete project or campaign.
As tid values for the datasets are needed, and they are not available in the currently stored files, we defined a new auxiliary table (Dsguids) to retrieve a tid by looking for a GUID. This table contains data exported from the Supervisor database for the production data and will be only needed during the data import procedure.
The Trigger Counter is a web service able to provide information about the trigger composition of a given dataset. It is able to perform three different operations:
It can count the occurrences of each trigger in the dataset. The counts can be grouped by Luminosity Block (LB) or Bunch Crossing IDentifier (BCID) if desired.
It can provide a list of events, given trigger-based search criteria.
It can calculate the overlap of the triggers in the dataset.
When asking for any of these operations, it is possible to apply a filter: a logical expression that will be applied to each event. The logical expression can include parentheses, boolean operators, trigger names or numbers and comparison operators to restrict integer ranges for LB and BCID. ‒ ‒ ‒ ‒ ‒
Since Trigger Counter is a user interactive web service, it must be reasonably fast when responding to the required operation. For this reason the data processed by Trigger Counter have been thoughtfully and thoroughly optimized to provide high performance when processing a dataset while keeping it as small as possible packing Level 1 triggers (512 possible values) into 10 bits using 32 -bit words and HLT triggers (4 096 possible values) into 12 bits using 64-bit words and using LZ4 [20] to further compress the output which offers a satisfactory compression ratio while its decompression is fast.
Once the user has filled the form a Map/Reduce job is run and within a couple of minutes, the results are displayed in interactive plots. Additionally, JSON and CSV versions of the results are also available for download. Figure 2 shows some screenshots of the Trigger Counter data displays.
Trigger Counter is going to be migrated to make use of the Hbase/Phoenix infrastructure. The new data structures have fields and families to store the six trigger masks of the event. Some of the optimizations that were gained with the current implementation will be lost in favour of standard data structures offered by Phoenix.
We developed a new back-end plugin for HBase/Phoenix ingestion that can be used with the Consumer part of the Data Collection architecture. Input data is stored in the Object Store and read by Consumers that make the necessary schema adaptations to the data, using then the JDBC1 Phoenix driver to store the data in HBase with the current schema. The first simple test inserting 48 datasets coming from the 2018 Tier-0 production showed a baseline performance of 3 kHz (3 000 events/s) per single-thread consumer. This implementation is still far from the 15 kHz of the previous Hadoop consumer plugin, but it must be taken into account that the first test did not use multiple threads, and that the event table design was slightly different without optimizations.
The current approach to the import of existing production data is to use Map/Reduce or Spark jobs to import all data from the Hadoop MapFiles. We can submit importer jobs with the desired input paths: from a single dataset to all the project data. The job analyzes the input data and spawns a single task per dataset (each dataset is stored in a single MapFile).
The first tests running Map/Reduce jobs used the CERN Analytix production cluster with 32 HBase region servers, the proposed event schema table with the described rowkey and four families of 1 Java DataBase Connectivity, the Java API to the database. data, and Phoenix features like automatically salted keys (10 buckets), FastDiff encoding and Snappy compression (low performance overhead and provides space savings). We first tested individual dataset ingestion performance with a simple job using one Yarn container, one VCPU, and one mapper (no file splits). We ingested datasets in the order of 1 MB, 100 MB, 1 GB, and 10 GB, and yielded ingestion rates varying from 500 Hz (1 MB dataset) to 4.2 kHz (1 GB datasets). The performance of massive ingestion was measured by ingesting 8 000 datasets, containing 70 billion events, with a mean rate of 115 kHz. The procedure lasted one week in the shared cluster, and at some point 1 000 concurrent containers were running with 4 TB of allocated memory (which corresponds to 20% of the Analytix cluster). The big majority of datasets were imported correctly. Some tasks were killed by Yarn pre-empting the container to submit higher priority tasks. In this case the procedure automatically restarted and produced the correct output and book-keeping records. There was an issue also with large datasets that were taking more than 24 hours. Tasks can run more than that, but when finishing the task, the JDBC driver closes the connection prematurely, losing the last batch of data (order of 100 events). This is a problem of the Phoenix driver that we still need to solve.
Queries to check the use cases were performed on different instances, first on a standalone test cluster, and later on the Analytix production cluster:
Datasets table: We can use the datasets auxiliary tables to discover datasets using partial dataset name components. We can also operate on the metadata stored per dataset, like for example the number of events. Even a full table scan will perform fast, so for example obtaining the total number of stored events summing up all the dataset entries in the entire ATLAS database will take less than one second.
Events table: Then we can perform queries on the events table, and for these we run the set of queries in two scenarios: o o 1st batch of queries while inserting data (100k ops/s writing load); 2nd batch of queries without writing load (on the day after the 1st batch).
Table 1 shows a few examples of typical queries to the Datasets and Events tables and their current performance. Count operations can be executed directly on the Datasets table and yield very fast results, as in examples (1) and (2).
A scan on the Events table is much slower, as query (3) on this table lasts over two minutes (151 s in the first batch and 140 s in the second batch). Note that this SQL query needs to use the dspid retrieved from the Datasets table, but this is a fast operation, and the bulk of the time is spent scanning the table with the dspid prefix.
For operations using trigger information, first we get the dspid from a particular entry of the Datasets table with query (4), then we use the previously retrieved dspid to select all the events from this dataset that also have (for example) the trigger 100 on the TAV (trigger after veto) mask, as in query (5). On the pre-production system this query lasted 107 seconds on the first batch, and 96 seconds on the second batch, but on the test machine with less data and no other users this test lasted only 12 seconds. Finding a particular event requires knowledge of the dataset (dspid), data type and event number. On the test instance, this was done retrieving the reference and provenance, and decoding it with user-defined functions (EI_REF0, EI_PRV0) installed in the cluster; altogether query (6) lasted less than one second. When this test was performed, this query was not possible on the production infrastructure as it was not possible to install dedicated functions for users.
Although the queries done while writing data take more time to complete, it is not a substantial amount of time. In addition, this is the expected scenario on the future production system, where we will be ingesting data constantly.
The ATLAS EventIndex was designed to hold the catalogue of all ATLAS events in advance of LHC Run 2 in 2012-2013, and all system components were developed and deployed in their first implementation by the start of Run 2 in 2015. Like any software project, it went through several stages of development and optimisation through the years. Thanks to the partitioned project architecture, each new component version could be tested in parallel with the production version and phased in when its performance was considered stable, and better than the previous version. The EventIndex operation and performance during and after the Run 2 period have been satisfactory.
The significant increases in the data rates expected in LHC Run 3 and the subsequent HLLHC runs require a transition now to a new technology for the main EventIndex data store. A new prototype based on HBase event tables and queries through Apache Phoenix has been tested and shows encouraging results. A good table schema was designed, and the basic functionality is ready. We are now working towards improved performance and better interfaces, with the aim to have the refactored system in operation well in advance of the start of Run 3 in 2022. According to our expectations, this system will be able to withstand the data production rates foreseen for LHC Run 4 and beyond.
Development Kit (SDK) for Python: