{
  "$type": "site.standard.document",
  "description": "In this post, we will talk about how to implement a Rust workflow that can process a large number of BGP data files as fast as we can. We will use BGPKIT Parser...",
  "path": "/parallel-mrt-files-parsing-with-bgpkit/",
  "publishedAt": "2022-03-16T00:00:00.000Z",
  "site": "at://did:plc:y2ynk4qyitlzipkvhcpu4e7v/site.standard.publication/3mp4zlyvjak24",
  "tags": [
    "tutorial"
  ],
  "textContent": "In this post, we will talk about how to implement a Rust workflow that can process a large number of BGP data files as fast as we can. We will use BGPKIT Parser and Broker for data collection and parsing, and Rayon crate for parallelization of the code.\n\n Task Overview\n\nBefore we begin to talk about the code design, we first need to introduce the data we are dealing with. We want to process the BGP data collected by various collectors, saved in compressed binary MRT format, and archived to files with a fixed interval. In this post, we are using RouteViews archive data as an example. The average data file size ranges from 2MB to 10MB by different collectors (AMSIX collector for example has pretty large files).\n\nFor processing, we can use the simplest task possible to do: sum the number of MRT records in all the files. We want to download and process all the updates files for one hour from all the collectors in RouteViews project. Here is the estimated amount of data we are dealing with:\n\n 35 collectors\n    \n 5-minute interval — 12 files per collector\n    \n 420 total number of files to download and process\n    \n 840MB to 4.2GB total download size (it’s somewhere in between)\n    \n\nOk! Now that we know what we want to do and have a sense of the estimated workload of the overall task, let’s coding!\n\nPhoto by Glenn Carstens-Peters on Unsplash\n\n---\n\n 1\\. Sequential Parsing\n\nOur first attempt to achieve the goal is to design and implement a naive sequential workflow as described below\n\n1. find all BGP updates files within the hour of interest\n    \n2. iterate through each file, parse the MRT data and count the number of records\n    \n3. sum all record counts and print out the result\n    \n\nFor this sequential workflow, we will need to pull in two dependencies into Cargo.toml:\n\nThe bgpkit-broker handles looking for updates files within the hour, while bgpkit-parser handles parsing each individual file.\n\n Finding files\n\nBGPKIT Broker indexes all available BGP MRT data archive files from both RouteViews and RIPE RIS in close-to-real-time. For each data file, it saves the following information:\n\n project: route-views or riperis\n    \n collector: the collector ID, e.g. rrc00 or route-views2\n    \n url: the URL to the corresponding MRT file\n    \n timestamp: the UNIX time of the start time of the MRT data file.\n    \n\nWith all this information indexed, we can then query the backend and retrieve files information as we want. BGPKIT Broker provides both RESTful API, Rust API, as well as a Python API. Here we use the Rust API to pull in the information we need:\n\nThe above block queries the broker and prints out all information of the retrieved files' metadata. The BgpkitBroker::newwithparams call accepts two parameters, one for the endpoint of the broker instance, and the other specifies the filtering criteria. In this example, we search for all BGP updates files from RouteViews with timestamps between 2022-01-01T00:00:00 and 2022-01-01T00:59:59 UTC. It prints out the output as the following:\n\n Parse each MRT file\n\nPreviously, in the for loop, we only print out the retrieved meta information of the MRT files. Now let's add the actual parsing of the files into the loop. The code is very simple, as designed:\n\nWe first define a mutable variable sum outside the loop. Then for each file, we create a new Parser instance by BgpkitParser::new(&item.url). Here, as our goal is to count the number of records, we call the parser's .intorecorditer() function to create a iterator over the records of the file, and then .count() to get the count of the records. Lastly, we add the count to the overall sum variable.\n\n Run and timing\n\nFor testing, I use a fairly powerful VM on a host with AMD 3950x CPU (32 threads), then build the release build and time the release run. The runtime includes downloading the MRT files to my machine with 1Gpbs down link in Southern California.\n\nIt ended up taking about 1 minute and 23 seconds to sequentially parse 144 MRT files from RouteViews for all available ones within the first hour of 2022 (UTC).\n\n 2\\. Parallel Parsing\n\nSince the parsing of each file is completely independent of each other, we can parse the files in parallel and then sum up the count for each thread at the end. In Rust with Rayon, this conversion is very simple.\n\nLet's first add the dependency of Rayon first:\n\nThen we change the broker code one tiny bit to collect all meta information for files into a vector first.\n\nThis would enable us to fully utilize rayon's great syntax sugar to turn our sequential code into a parallel one.\n\nThe key difference here is the calling of .pariter(). It turns a sequential iterator into a parallel iterator, and by default going to utilize all available cores on the host machine for scheduling. Then we call .map() to define the parsing steps for each file, and then call .sum() at the end to add all results up.\n\nThe final result is approximately 10x faster than the sequential version, and it took only 8 seconds to parse all MRT files and get the record counts.\n\nThe full code for this example is as follows:\n\n---\n\nThe source code of the two examples is available on GitHub. Feel free to poke around and tweak it as you wish.\n\n%[https://gist.github.com/digizeph/c23ba39968c6cb4e1ad323520540010ffile-parallel-parsing-rs] \n\nhttps://github.com/bgpkit/bgpkit-tutorials/tree/main/parallel-parsing",
  "title": "Parallel MRT Files Parsing with BGPKIT"
}