3
\$\begingroup\$

I'm building a basic data pipeline in .NET to extract data from tables in our RDBMS systems, that exports into columnar storage using Parquet. I'm using Parquet's row group to batch up the writes into manageable segments as switching from rows to columns means holding a chunk of data in memory.

Having to write each column in the batch synchronously (to preserve order) means that it's a fairly slow endeavor to begin with. I'm dealing with datasets with hundreds of millions of rows deep and 100s of columns wide, I need to squeeze every last drop of performance I can muster.

Working example so far:

using System;
using System.Collections.Generic;
using System.Data;
using System.Data.OleDb;
using System.Diagnostics;
using System.IO;
using System.Linq;
using Parquet;
using Parquet.Data;
using Parquet.Data.Rows;

namespace ParquetDBTest
{
    public static class Program
    {
        private const int BatchSize = 5000;

        public static void Main(string[] args)
        {
            var tables = new[] {"TABLE"};

            foreach (var table in tables) Writer(table);

            Console.ReadKey();
        }

        public static async void Writer(string tableName)
        {
            Console.WriteLine($"Start Time {DateTime.Now}");
            var watch = Stopwatch.StartNew();

            var rowCount = 0;
            var connectionString =
                "Data Source=;User ID=;Password=;Provider=OraOLEDB.Oracle.1;Persist Security Info=True;PLSQLRSet=1;";
            using (var connection = new OleDbConnection(connectionString))
            {
                try
                {
                    await connection.OpenAsync();
                    using (var command = connection.CreateCommand())
                    {
                        command.CommandText = $"SELECT * FROM {tableName}";
                        command.CommandType = CommandType.Text;
                        command.CommandTimeout = 1000;

                        using (var reader = await command.ExecuteReaderAsync())
                        {
                            var columns = Enumerable.Range(0, reader.FieldCount)
                                .Select(o => new {FieldName = reader.GetName(o), DataType = reader.GetFieldType(o)})
                                .ToList();

                            var columnSchema = new List<DataField>(columns.Count);
                            columnSchema.AddRange(from column in columns
                                let columnType = column.DataType.IsClass
                                    ? column.DataType
                                    : typeof(Nullable<>).MakeGenericType(column.DataType)
                                let fieldType = typeof(DataField<>).MakeGenericType(columnType)
                                select (DataField) Activator.CreateInstance(fieldType, column.FieldName));

                            var schema = new Schema(columnSchema);

                            var dataset = new Table(schema);
                            var i = 1;

                            var outputFile =
                                new FileInfo($"C:\\Temp\\{tableName}\\{tableName}.snappy.parquet");
                            outputFile.Directory.Create();

                            using (var fileStream = outputFile.Create())
                            {
                                using (var writer = new ParquetWriter(schema, fileStream)
                                    {CompressionMethod = CompressionMethod.Snappy})
                                {
                                    while (await reader.ReadAsync())
                                    {
                                        var row = Enumerable.Range(0, reader.FieldCount)
                                            .Select(o => ToParquetType(reader.GetValue(o)))
                                            .ToArray();

                                        dataset.Add(row);

                                        if (i == BatchSize)
                                        {
                                            writer.Write(dataset);
                                            dataset.Clear();

                                            i = 1;
                                        }

                                        i++;
                                        rowCount++;
                                    }
                                }
                            }
                        }
                    }
                }
                catch (OleDbException ex)
                {
                    Console.WriteLine(ex.Message);
                    throw;
                }
                finally
                {
                    connection.Close();
                    watch.Stop();

                    Console.WriteLine($"Finish Time {DateTime.Now}");
                    Console.WriteLine("--------------");
                    Console.WriteLine($"Wrote {rowCount} row(s) in {watch.Elapsed.TotalSeconds} seconds!");

                }
            }
        }

        private static object ToParquetType(object inputObject)
        {
            var outputObject = inputObject;

            if (inputObject == DBNull.Value) outputObject = null;

            if (inputObject is DateTime time) outputObject = new DateTimeOffset(time);

            return outputObject;
        }
    }
}

Are there any obvious optimizations I can make?

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.