Problem
I'm trying to create the most performant code to:
- Extract a text file from a zip
- parse its content
- serialize class
- post to Azure Queue Storage
Scenario
4 times per year I receive 20 zip files. All zip files have just one text file inside. Each text file is a fixed width file. Each line in the text file is a different type of data. There are 3 of them:
- Company (First character of line is "1")
- Partners (First character of line is "2")
- Activity (First character of line is "6")
They are always in sequence:
1First Company Data ..................................
2First Company First Partner .........................
2First Company Second Partner ........................
2First Company Third Partner .........................
6First Company Activity ..............................
1Second Company Data .................................
2Second Company First Partner ........................
2Second Company Second Partner .......................
2Second Company Third Partner ........................
2Second Company Fourth Partner .......................
6Second Company Activity .............................
There are always one line for company, one line for activity, but zero to many partners.
Activity is compound of 7 numbers sequence repeated multiple times. For example:
- 1111111 is an activity
if the company have more than one activity, they are putted in sequence like:
- 111111122222223333333
if there's no other activity, zeros are used until the end of line:
- 11111112222222333333300000000000000000000000000000000000
the line identifier for activity is "6", so, the full line stills like:
- 611111112222222333333300000000000000000000000000000000000
Two companys appear like this way:
1First Company Data .....................................
2First Company First Partner ............................
2First Company Second Partner ...........................
2First Company Third Partner ............................
611111112222222333333300000000000000000000000000000000000
1Second Company Data ....................................
2Second Company First Partner ...........................
2Second Company Second Partner ..........................
2Second Company Third Partner ...........................
2Second Company Fourth Partner ..........................
644444445555555000000000000000000000000000000000000000000
the zip files need to be read in sequence, because, the last line of a file may be in the middle of company data:
1First Company Data .....................................
2First Company First Partner ............................
2First Company Second Partner ...........................
2First Company Third Partner ............................
611111112222222333333300000000000000000000000000000000000
1Second Company Data ....................................
2Second Company First Partner ...........................
[------ END OF FILE 1 ------]
[----- BEGIN OF FILE 2 -----]
2Second Company Second Partner ..........................
2Second Company Third Partner ...........................
2Second Company Fourth Partner ..........................
644444445555555000000000000000000000000000000000000000000
About the code
I never worked with multithread before, I tried to put a thread to open the zip files, read its content, parse the data and post it to a Blocking Collection. The second thread I used to post Base 64 Encoded serialized class to Azure Queue Storage. I tried to use more than one thread to this but with no success. The third thread I use just to notify about the processing. I used a trick to post data to Azure Queue in batch, it sped up the performance, but I don't know if there is a more secure and better way to do this.
I'm posting to Azure Queue Storage, and, in the other side, I have another console to take data from Azure Queue and post to SQL Server. If there is a way to do this that is secure and reliable, but without using Azure Queue, I'll appreciate too. I tried this method in the first time, but network problems in the middle of process make me lost 120,000 companies data. The total of companies is about 43,000,000.
The data is public, and can be downloaded at: http://receita.economia.gov.br/orientacao/tributaria/cadastros/cadastro-nacional-de-pessoas-juridicas-cnpj/dados-publicos-cnpj
Code
this is the main console code
using Azure.Storage.Queues;
using BaseReceita.Producer.Model;
using ICSharpCode.SharpZipLib.Zip;
using Microsoft.Extensions.Configuration;
using Newtonsoft.Json;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
namespace BaseReceita.Producer
{
class Program
{
private static string FilesDirectory { get; set; }
static void Main(string[] args)
{
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Starting...");
Console.WriteLine("Specify the folder with zip files: ");
FilesDirectory = @"" + Console.ReadLine();
Start().Wait();
}
private static async Task Start()
{
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] starting to read...");
var watch = new Stopwatch();
watch.Start();
ServicePointManager.UseNagleAlgorithm = false;
ServicePointManager.DefaultConnectionLimit = 1000;
object lockobj = new object();
long RegistrosProcessados = 0;
var ts = new CancellationTokenSource();
CancellationToken ct = ts.Token;
IConfigurationRoot Configuration = new ConfigurationBuilder()
.SetBasePath(Directory.GetCurrentDirectory())
.AddJsonFile("appsettings.json")
.Build();
string storageConnectionString = Configuration["Storage:ConnectionString"];
QueueClient queueClient = new QueueClient(storageConnectionString, "rfb-update-queue");
//-------------------- Collection
BufferBlock<string> buffer = new BufferBlock<string>(new DataflowBlockOptions() { BoundedCapacity = 50000 });
//-------------------- Consumers
var Consumers = new List<Task>();
for (var i = 0; i < 1; i++)
{
Consumers.Add(Task.Run(async () => {
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Consumer starting.");
while (await buffer.OutputAvailableAsync(ct))
{
if (buffer.TryReceiveAll(out var items))
{
try
{
await SendMessagesAsync(queueClient, items.AsEnumerable());
lock (lockobj)
RegistrosProcessados = RegistrosProcessados + items.Count;
}
catch (Exception e)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Erro: {e.Message}");
Console.ResetColor();
//throw;
}
}
}
Console.ForegroundColor = ConsoleColor.DarkCyan;
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Consumer finalized");
Console.ResetColor();
}));
}
//-------------------- Notifier
Task Notifier = Task.Factory.StartNew(() =>
{
while (true)
{
if (!ct.IsCancellationRequested)
{
//F = Buffer Size
//P = Processed companies
//in the sequence, average processed per second, per minute and per hour
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] F-{buffer.Count} P-{RegistrosProcessados:n0} ({RegistrosProcessados / watch.Elapsed.TotalSeconds:n0}/s | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60):n0}/m | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60 / 60):n0}/h)");
Thread.Sleep(5000); //notify every 5 seconds
}
else
{
break;
}
}
});
//-------------------- Producer
Task Producer = Task.Run(async () =>
{
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Producer started.");
string conjunto = string.Empty;
string linha = string.Empty;
try
{
//open all zip files
foreach (string file in Directory.EnumerateFiles(FilesDirectory, "*.zip"))
{
//open zip
using (ZipFile zf = new ZipFile(file))
{
//take all files (aways will be one file
foreach (ZipEntry entry in zf)
{
//open as stream
using (var stream = zf.GetInputStream(entry))
using (var reader = new StreamReader(stream))
{
//read line from file
while ((linha = reader.ReadLine()) != null)
{
string tipoCampo = linha.Substring(0, 1);
if (tipoCampo == "1")
{
//every "1" is a new company, than, I parse the last company based on all the text extracted
EmpresaModel empresa = Parse(conjunto);
if (empresa != null)
{
//the first time will be null
//the others wont
//serialize, compress and post to buffer
string json = JsonConvert.SerializeObject(empresa);
string compressed = Base64Compress(json);
buffer.Post(compressed);
}
conjunto = linha;
}
else if (tipoCampo != "0")
{
conjunto = conjunto + Environment.NewLine + linha;
}
}
}
}
}
}
}
catch (Exception e)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Producer Error: {e.Message}");
Console.ResetColor();
}
Console.ForegroundColor = ConsoleColor.DarkCyan;
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Producer finalized");
Console.ResetColor();
});
try
{
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Starting threads.");
List<Task> tasks = new List<Task>();
tasks.Add(Producer);
tasks.AddRange(Consumers);
Task.WaitAll(tasks.ToArray());
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Threads finalized");
}
catch (Exception e)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Error: {e.Message}");
Console.ResetColor();
}
ts.Cancel();
watch.Stop();
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Total Time: {watch.Elapsed.ToString()}");
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Pushed-{RegistrosProcessados:n0} ({RegistrosProcessados / watch.Elapsed.TotalSeconds:n0}/s | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60):n0}/m | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60 / 60):n0}/h)");
Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] End of process");
Console.ReadLine();
}
private static EmpresaModel Parse(string conjunto)
{
EmpresaModel empresa = null;
if (!string.IsNullOrEmpty(conjunto))
{
string[] linhas = conjunto.Trim().Split(new[] { '\n' });
foreach (string linha in linhas)
{
string cnpj = linha.Substring(3, 14);
if (linha.Substring(0, 1) == "1")
{
//Company
empresa = new EmpresaModel();
empresa.Cnpj = cnpj;
empresa.IndicadorMatrizFilial = linha.Substring(17, 1).Trim();
empresa.RazaoSocial = linha.Substring(18, 150).Trim();
empresa.NomeFantasia = linha.Substring(168, 55).Trim();
empresa.CodigoSituacaoCadastral = linha.Substring(223, 2).Trim();
//empresa.SituacaoCadastral = (string)SituacaoCadastral.FirstOrDefault(x => x.Key == empresa.CodigoSituacaoCadastral).Value;
empresa.DataSituacaoCadastral = linha.Substring(225, 8).Trim();
empresa.CodigoMotivoSituacaoCadastral = linha.Substring(233, 2).Trim();
//empresa.MotivoSituacaoCadastral = (string)MotivoSituacaoCadastral.FirstOrDefault(x => x.Key == empresa.CodigoMotivoSituacaoCadastral).Value;
empresa.CidadeExterior = linha.Substring(235, 55).Trim();
empresa.CodigoPais = linha.Substring(290, 3).Trim();
empresa.Pais = linha.Substring(293, 70).Trim();
empresa.CodigoNaturezaJuridica = linha.Substring(363, 3).Trim() + "-" + linha.Substring(366, 1).Trim();
//empresa.NaturezaJuridica = (string)NaturezaJuridica.FirstOrDefault(x => x.Key == empresa.CodigoNaturezaJuridica).Value;
empresa.DataInicioAtividade = linha.Substring(367, 8).Trim();
empresa.IdCnae = linha.Substring(375, 7).Trim();
empresa.TipoLogradouro = linha.Substring(382, 20).Trim();
empresa.Logradouro = linha.Substring(402, 60).Trim();
empresa.Numero = linha.Substring(462, 6).Trim();
empresa.Complemento = linha.Substring(468, 156).Trim();
empresa.Bairro = linha.Substring(624, 50).Trim();
empresa.Cep = linha.Substring(674, 8).Trim();
empresa.UF = linha.Substring(682, 2).Trim();
empresa.CodigoMunicipio = linha.Substring(684, 4).Trim();
empresa.Municipio = linha.Substring(688, 50).Trim();
empresa.DDD1 = linha.Substring(738, 4).Trim();
empresa.Telefone1 = linha.Substring(742, 8).Trim();
empresa.DDD2 = linha.Substring(750, 4).Trim();
empresa.Telefone2 = linha.Substring(754, 8).Trim();
empresa.DDDFax = linha.Substring(762, 4).Trim();
empresa.TelefoneFax = linha.Substring(766, 8).Trim();
empresa.Email = linha.Substring(774, 115).Trim();
empresa.CodigoQualificacaoResponsavel = linha.Substring(889, 2).Trim();
empresa.CapitalSocial = linha.Substring(891, 14).Trim();
empresa.CodigoPorteEmpresa = linha.Substring(905, 2).Trim();
empresa.CodigoOpcaoSimplesNacional = linha.Substring(907, 1).Trim();
empresa.DataOpcaoSimples = linha.Substring(908, 8).Trim();
empresa.OptanteMei = linha.Substring(924, 1).Trim();
empresa.SituacaoEspecial = linha.Substring(925, 23).Trim();
empresa.DataSituacaoEspecial = linha.Substring(948, 8).Trim();
}
else if (linha.Substring(0, 1) == "2")
{
//Partners
QuadroSocietarioModel qsa = new QuadroSocietarioModel();
qsa.Cnpj = linha.Substring(3, 14).Trim();
qsa.IdentificadorSocio = linha.Substring(17, 1).Trim();
qsa.NomeSocio = linha.Substring(18, 150).Trim();
qsa.CnpjCpfSocio = linha.Substring(168, 14).Trim();
qsa.CodigoQualificacaoSocio = linha.Substring(182, 2).Trim();
//qsa.QualificacaoSocio = (string)QualificacaoResponsavelSocio.FirstOrDefault(x => x.Key == qsa.CodigoQualificacaoSocio).Value;
qsa.PercentualCapitalSocial = linha.Substring(184, 5).Trim();
qsa.DataEntradaSociedade = linha.Substring(189, 8).Trim();
qsa.CodigoPais = linha.Substring(197, 3).Trim();
qsa.Pais = linha.Substring(200, 70).Trim();
qsa.CpfRepresentanteLegal = linha.Substring(270, 11).Trim();
qsa.NomeRepresentante = linha.Substring(281, 60).Trim();
qsa.CodigoQualificacaoRepresentanteLegal = linha.Substring(341, 2).Trim();
empresa?.QuadroSocietario.Add(qsa);
}
else if (linha.Substring(0, 1) == "6")
{
//Activity
string[] cnaes =
Split(linha.Substring(17, 693).Trim(), 7)
.Where(x => x != "0000000")
.Where(x => !string.IsNullOrEmpty(x.Trim()))
//.Select(x => "cnae/" + x)
.ToArray();
foreach (string cnae in cnaes)
{
CnaeSecundarioModel cnaeSecundario = new CnaeSecundarioModel();
cnaeSecundario.Cnpj = cnpj;
cnaeSecundario.Cnae = cnae;
empresa?.CnaesSecundarios.Add(cnaeSecundario);
}
}
}
}
return empresa;
}
private static IEnumerable<string> Split(string str, int chunkSize)
{
return Enumerable.Range(0, str.Length / chunkSize)
.Select(i => str.Substring(i * chunkSize, chunkSize));
}
private static string Base64Compress(string s)
{
byte[] inputBytes = Encoding.UTF8.GetBytes(s);
using (var outputStream = new MemoryStream())
{
using (var gZipStream = new System.IO.Compression.GZipStream(outputStream, System.IO.Compression.CompressionMode.Compress))
gZipStream.Write(inputBytes, 0, inputBytes.Length);
var outputBytes = outputStream.ToArray();
var outputbase64 = Convert.ToBase64String(outputBytes);
return outputbase64;
}
}
public static async Task SendMessagesAsync(QueueClient queue, IEnumerable<string> messages)
{
await Task.WhenAll(
from partition in Partitioner.Create(messages).GetPartitions(500)
select Task.Run(async delegate
{
using (partition)
while (partition.MoveNext())
await queue.SendMessageAsync(partition.Current);
}));
}
}
}
here are the entities
using System.Collections.Generic;
namespace BaseReceita.Producer.Model
{
public class EmpresaModel
{
public EmpresaModel()
{
QuadroSocietario = new HashSet<QuadroSocietarioModel>();
CnaesSecundarios = new HashSet<CnaeSecundarioModel>();
}
public string Cnpj { get; set; }
public string IndicadorMatrizFilial { get; set; }
public string RazaoSocial { get; set; }
public string NomeFantasia { get; set; }
public string CodigoSituacaoCadastral { get; set; }
public string DataSituacaoCadastral { get; set; }
public string CodigoMotivoSituacaoCadastral { get; set; }
public string CidadeExterior { get; set; }
public string CodigoPais { get; set; }
public string Pais { get; set; }
public string CodigoNaturezaJuridica { get; set; }
public string DataInicioAtividade { get; set; }
public string IdCnae { get; set; }
public string TipoLogradouro { get; set; }
public string Logradouro { get; set; }
public string Numero { get; set; }
public string Complemento { get; set; }
public string Bairro { get; set; }
public string Cep { get; set; }
public string UF { get; set; }
public string CodigoMunicipio { get; set; }
public string Municipio { get; set; }
public string DDD1 { get; set; }
public string Telefone1 { get; set; }
public string DDD2 { get; set; }
public string Telefone2 { get; set; }
public string DDDFax { get; set; }
public string TelefoneFax { get; set; }
public string Email { get; set; }
public string CodigoQualificacaoResponsavel { get; set; }
public string CapitalSocial { get; set; }
public string CodigoPorteEmpresa { get; set; }
public string CodigoOpcaoSimplesNacional { get; set; }
public string DataOpcaoSimples { get; set; }
public string OptanteMei { get; set; }
public string SituacaoEspecial { get; set; }
public string DataSituacaoEspecial { get; set; }
public HashSet<QuadroSocietarioModel> QuadroSocietario { get; set; }
public HashSet<CnaeSecundarioModel> CnaesSecundarios { get; set; }
}
public class QuadroSocietarioModel
{
public string Cnpj { get; set; }
public string IdentificadorSocio { get; set; }
public string NomeSocio { get; set; }
public string CnpjCpfSocio { get; set; }
public string CodigoQualificacaoSocio { get; set; }
public string PercentualCapitalSocial { get; set; }
public string DataEntradaSociedade { get; set; }
public string CodigoPais { get; set; }
public string Pais { get; set; }
public string CpfRepresentanteLegal { get; set; }
public string NomeRepresentante { get; set; }
public string CodigoQualificacaoRepresentanteLegal { get; set; }
}
public class CnaeSecundarioModel
{
public string Cnpj { get; set; }
public string Cnae { get; set; }
}
}