1
\$\begingroup\$

This is a C# console application that creates Parquet file for SQL Server table.

Given a 5GB SQL table, it takes nearly 65 minutes, so for our bigger tables (up to 500GB), it may end up taking days to complete.

Can someone suggest how I can improve performance of this solution?

namespace Parquet_Generator
{
    public static class Program
    {
        private const int batchsize = 10000;
        static void Main(string[] args)
        {
            var SourceObjectName = new[] { @args[3] };
            string SourceObject_IsTable = args[2];
            string ServerName = @args[0];
            string DBName = @args[1];
            string DestinationPath = @args[4];

            long count = 0;
            try
            {
                foreach (var SourceObject in SourceObjectName)
                {
                    Task task = ParquetWriter(ServerName, DBName, SourceObject, DestinationPath, count, SourceObject_IsTable);
              //      Console.WriteLine($"End Time {DateTime.Now}");
                    task.Wait();
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                Console.Error.WriteLine(ex.Message);
                throw ;
            }
        }
        private static async Task<long> ParquetWriter(string ServerName, string DBName, string SourceObject, string DestinationPath, long rowcount, string SourceObject_IsTable)
        {
            Console.WriteLine($"Start Time {DateTime.Now}");
            int cnt = 0;
            var watch = Stopwatch.StartNew();
            var connectionstring = "Data Source = " + ServerName + ";Initial Catalog = " + DBName + ";Integrated Security = True;";
            using (var connection = new SqlConnection(connectionstring))
            {
                try
                {
                    await connection.OpenAsync();
                    using (var command = connection.CreateCommand())
                    {
                        string text = null;
                        CommandType texSourceObject_IsTable = 0;
                        if (SourceObject_IsTable == "1")
                        {
                            text = $"SELECT  * FROM {SourceObject}";
                            texSourceObject_IsTable = CommandType.Text;
                        }
                        if (SourceObject_IsTable == "0")
                        {
                            text = SourceObject;
                            texSourceObject_IsTable = CommandType.StoredProcedure;
                        }
                        command.CommandText = text;
                        command.CommandType = texSourceObject_IsTable;
                        command.CommandTimeout = 999999999;
                        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 Batchrowcount = 1;
                            var outputfile = new FileInfo(DestinationPath + SourceObject + ".Gzip.parquet");
                            outputfile.Directory.Create();
                            using (var filestream = outputfile.Create())
                            {
                                using (var writer = new ParquetWriter(schema, filestream)
                                { CompressionMethod = CompressionMethod.Gzip })
                                {
                                    while (await reader.ReadAsync())
                                    {
                                        var row = Enumerable.Range(0, reader.FieldCount)
                                            .Select(o => ToParqueSourceObject_IsTable(reader.GetValue(o)))
                                            .ToArray();
                                        dataset.Add(row);
                                    //    cnt++;
                                        if (Batchrowcount == batchsize)
                                        {
                                            writer.Write(dataset);
                                             dataset.Clear();
                                            Batchrowcount = 1;
                                        }
                                       Batchrowcount++;
                                        rowcount++;
                                    }
                                   writer.Write(dataset);
                                    dataset.Clear();
                                    filestream.Flush();
                                }
                            }
                        }
                    }
                    Console.WriteLine($"Total Rows: {cnt}");
                    return rowcount;
                }
                catch (Exception 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.TotalMinutes} Minutes!");
                        Console.ReadKey();
                }
            }
        }
        private static object ToParqueSourceObject_IsTable(object inputObject)
        {
            var outputObject = inputObject;
            if (inputObject == DBNull.Value) outputObject = null;
            if (inputObject is DateTime time) outputObject = new DateTimeOffset(time,new TimeSpan(0,0,0));
            return outputObject;
        }
    }
}
\$\endgroup\$

3 Answers 3

7
\$\begingroup\$

Quick remarks:

  • Your code doesn't follow the coding conventions (camel-case, pascal-case, etc.).

  • string SourceObject_IsTable = args[2]; has a bad name and isn't the correct type: it should be a boolean.

  • Why do you do this: var SourceObjectName = new[] { @args[3] };? Also, SourceObjectName is singular, not plural, and thus a bad name for an array.

  • Your console job has five parameters. In such a case I'd look for a better way to parse them, e.g. a NuGet package like CommandLineParser.

  • ParquetWriter has six parameters; at that point I'd be using a custom class and pass that class instead this long list of parameters which are almost all strings.

  • Why the "old" string concatenation in var connectionstring = "Data Source = " + ServerName + ";Initial Catalog = " + DBName + ";Integrated Security = True;";?

  • Both if (SourceObject_IsTable == "1") and if (SourceObject_IsTable == "0") cannot be true at the same time, so there should be an else.

  • The biggest int can be is 2,147,483,647. Do you really expect to write more than 2 billion lines?

  • rowcount is a compound word and thus should be spelled rowCount. Ditto Batchrowcount.

  • Don't do new FileInfo(DestinationPath + SourceObject + ".Gzip.parquet"). Path.Combine exists to construct paths.

  • cnt is never updated, so there is no point to Console.WriteLine($"Total Rows: {cnt}");.

  • ToParqueSourceObject_IsTable is a really bad method name, not just because it doesn't follow the coding guidelines, but more so because its name doesn't tell me what it does (and frankly seems to be incorrect).

\$\endgroup\$
1
  • \$\begingroup\$ I am new to c#, I will update with all your suggestions , thanks \$\endgroup\$
    – Srini
    Mar 10, 2023 at 17:49
1
\$\begingroup\$

Aside from the good points made by BCdotWEB, I think your code performance will already be quite good thanks to batch processing.

One way improve the performance would be to interleave the reading of data from SQL Server with the writing of the parquet output so that while the current batch is written out you already read in the next batch from SQL Server.

This will necessitate to break up your code into more well defined concepts such as a data source (that produces the data read from SQL Server) and a data sink (that will write the output). The coupling between the two can achieved with a BlockingCollection<T>().

Conceptually something like this (illustrative only, won't compile):

static const int BatchSize = 10_000;

Task SqltoParquet(string server, string db, ... etc. ...)
{
    var buffer = new BlockingCollection<object[]>(BatchSize);

    ... create new connection and start reader ...

    Task parquetTask = Task.Complete;
    using (var reader = await command.ExecuteReaderAsync())
    {
        var columns = Enumerable.Range(0,  reader.FieldCount)
         .... infer schema ...
        var schema = new Schema(columnSchema);

        // start the parquet writer
        parquetTask = WriteToParquet(buffer, schema, ... etc. ...)

        while (await reader.ReadAsync())
        {
            var row = Enumerable.Range(0, reader.FieldCount)
                                .Select(o => ToParquetValue(reader.GetValue(o)))
                                .ToArray();
            buffer.Add(row);
        }
        buffer.CompleteAdding();
        await parquetTask;
    }
    
}

Task WriteToParquet(BlockingCollection<object[]> source, Schema schema, string targetDirectory, string targetFile, ... etc. ...)
{
    ... open output file ...
    ... create dataset based on schema ...
    var currentCount = 0;
    foreach (var row in source.GetConsumingEnumerable())
    {
        dataset.Add(row);
        currentCount++;
        if (currentCount >= BatchSize)
        {
            ... write out dataset ...
            currentCount = 0;
            dataset.Clear();
        }
    }
    ... write out remaining dataset etc. ...
}

Overall it will help readability and maintainability if you break out complex sub tasks (like determining the schema) into their own functions rather than inlining everything into one big chunk of code.

\$\endgroup\$
-4
\$\begingroup\$

Srini, You're original question was requesting "Can someone suggest how I can improve performance of this solution"

I'm not seeing any helpful recommendations from comments by BCdotWeb. Just a lot of gas. If allowed on your server, you may want to construct a CLR and ship the output to your final destination.

To add to the common knowledge base, how did you finally resolve your issue?

\$\endgroup\$
2

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.