Converting InfluxDB TSM and line protocol formats to Parquet

Andreas.Motl · May 18, 2021, 4:48pm

Hi there,

with the advent of GitHub - influxdata/influxdb_iox: Pronounced (influxdb eye-ox), short for iron oxide. This is the new core of InfluxDB written in Rust on top of Apache Arrow., there is some sweet tooling which supports different migration and data shoveling scenarios.

InfluxDB IOx offers a command line tool to convert both Time-Structured Merge Tree (TSM) files as well as InfluxDB line protocol files into the Parquet file format.

We collected a few references which offer a look under the hood.

With kind regards,
Andreas.

github.com

influxdata/influxdb_iox/blob/2784f89e6e582189c8e3d32fe09a69741978e8b3/src/commands/convert.rs#L237-L290


fn convert_line_protocol_to_parquet(
    input_filename: &str,
    mut input_reader: InputReader,
    compression_level: CompressionLevel,
    output_name: &str,
) -> Result<()> {
    // TODO: make a streaming parser that you can stream data through in blocks.
    // for now, just read the whole input at once into a string
    let mut buf = String::with_capacity(
        input_reader
            .len()
            .try_into()
            .expect("Cannot allocate buffer"),
    );
    input_reader
        .read_to_string(&mut buf)
        .context(UnableToReadInput {
            name: input_filename,
        })?;

This file has been truncated. show original

github.com

influxdata/influxdb_iox/blob/2784f89e6e582189c8e3d32fe09a69741978e8b3/src/commands/convert.rs#L292-L319


fn convert_tsm_to_parquet(
    index_stream: InputReader,
    index_stream_size: usize,
    compression_level: CompressionLevel,
    block_stream: InputReader,
    output_name: &str,
) -> Result<()> {
    // setup writing
    let writer_source: Box<dyn IOxTableWriterSource> = if is_directory(&output_name) {
        info!("Writing to output directory {:?}", output_name);
        Box::new(ParquetDirectoryWriterSource {
            compression_level,
            output_dir_path: PathBuf::from(output_name),
        })
    } else {
        info!("Writing to output file {}", output_name);
        Box::new(ParquetFileWriterSource {
            compression_level,
            output_filename: String::from(output_name),
            made_file: false,

This file has been truncated. show original

github.com

influxdata/influxdb_iox/blob/411cf134e9087198f74fe78ccb3c62e16344bf2f/benches/line_protocol_to_parquet.rs

use criterion::{criterion_group, criterion_main, Criterion, Throughput};
use influxdb_line_protocol::parse_lines;
use ingest::{
    parquet::writer::{CompressionLevel, IOxParquetTableWriter},
    ConversionSettings, LineProtocolConverter,
};
use internal_types::schema::Schema;
use packers::{Error as TableError, IOxTableWriter, IOxTableWriterSource};
use std::time::Duration;

use parquet::file::writer::TryClone;
use std::io::{Seek, SeekFrom, Write};

static LINES: &str = include_str!("../tests/fixtures/lineproto/metrics.lp");

/// Stream that throws away all output
struct IgnoringWriteStream {}

impl Write for IgnoringWriteStream {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {

This file has been truncated. show original