Load a CSV, clean it, group it, write Parquet

Every pipeline I've ever written started the same way. Somebody hands you a CSV, and the CSV isn't as clean as they think it is.

So that's what we're doing here. A CSV of orders goes in, a Parquet file of revenue by region comes out, and we stop and look at the data after every step in between. Give it about ten minutes. Everything uses the packages published on the ninth and nothing else.

You need the .NET 8 SDK (or later) and a terminal. Start with a new console project and three packages.

dotnet new console -o orders-pipeline
cd orders-pipeline
dotnet add package Datafication.Core
dotnet add package Datafication.CsvConnector
dotnet add package Datafication.ParquetConnector

Save this as orders.csv next to the .csproj. It has one duplicated order, one missing price, and one missing date. That's on purpose. A tidy sample file teaches you nothing about the parts of a pipeline that actually give you trouble.

OrderId,OrderDate,Region,Category,UnitPrice,Quantity
1001,2025-01-14,North,Widgets,19.99,3
1002,2025-02-03,South,Gadgets,149.00,1
1003,2025-02-27,North,Widgets,19.99,12
1004,2025-04-11,East,Gadgets,,5
1005,2025-05-02,North,Sprockets,4.25,7
1006,2025-06-18,South,Widgets,19.99,4
1003,2025-02-27,North,Widgets,19.99,12
1007,2025-07-09,West,Gadgets,149.00,1
1008,2025-08-21,North,Sprockets,4.25,25
1009,,East,Widgets,19.99,6
1010,2025-10-30,South,Sprockets,4.25,9
1011,2025-11-12,North,Gadgets,149.00,3
1012,2025-12-05,West,Widgets,19.99,8
1013,2025-03-22,East,Sprockets,4.25,14

Fourteen rows in, four rows and one Parquet file out.

Load it and look at it

First get it into memory and see what the connector made of it.

using Datafication.Core.Data;
using Datafication.Core.Sinks;
using Datafication.Extensions.Connectors.CsvConnector;

var orders = await DataBlock.Connector.LoadCsvAsync("orders.csv");
Console.WriteLine($"Loaded {orders.RowCount} rows");

var info = orders.Info();
Console.WriteLine(await info.TextTableAsync(10));

A relative path resolves against the current directory, which under dotnet run is the project folder. If the file lives on a web server instead, an http:// or https:// address goes in the same argument.

Info() is worth a moment, because it isn't a printer. It hands back a DataBlock of five columns, so you can filter or sort it like any other block. TextTableAsync, from Datafication.Core.Sinks, is the piece that turns a block into console output. Run it and here's what you see.

  |        Column|         Label|         Type|     Non-Null Count|     Null Count
 0|       OrderId|       OrderId|        Int32|                 14|              0
 1|     OrderDate|     OrderDate|     DateTime|                 13|              1
 2|        Region|        Region|       String|                 14|              0
 3|      Category|      Category|       String|                 14|              0
 4|     UnitPrice|     UnitPrice|       Double|                 13|              1
 5|      Quantity|      Quantity|        Int32|                 14|              0

Column	What it holds
`Column`	The column name
`Label`	Its display label, which defaults to the name
`Type`	The CLR type the connector inferred
`Non-Null Count`	Values that are not null
`Null Count`	Values that are

The CSV connector inferred a type for every column without being told: the dates parsed as DateTime, the prices as Double. And there are the two holes, sitting in the null counts as one missing date and one missing price.

Clean it

Three operations, and the order matters more than it looks.

var clean = orders
    .DropDuplicates(KeepDuplicateMode.First, "OrderId")
    .FillNulls(FillMethod.Mean, "UnitPrice")
    .DropNulls(DropNullMode.Any);

Console.WriteLine($"{clean.RowCount} rows after cleaning");

Deduplicate first, so the repeated order does not pull the mean sideways. DropDuplicates compares only the columns you name, and KeepDuplicateMode.First keeps the earlier row (Last and None are the alternatives).

FillNulls(FillMethod.Mean, ...) replaces nulls in the named column with the column mean. ForwardFill, BackwardFill, Median, Mode, LinearInterpolation and ConstantValue are the other methods, and which one is right depends on what the column actually means.

DropNulls then removes any row still holding a null anywhere, which here is the order with no date. It takes a mode and nothing else: Any drops a row with any null, All drops only rows that are null in every column. There's no column list on it, which is the part people go looking for.

Fourteen rows in, one duplicate gone, one row without a date gone: 12 rows after cleaning.

Compute two columns

Now the part you came for.

if (!clean.ValidateExpression("Quantity * UnitPrice", out var error))
{
    Console.WriteLine($"Bad expression: {error}");
    return;
}

var priced = clean
    .Compute("Revenue", "Quantity * UnitPrice")
    .Compute("Quarter", "QUARTER(OrderDate)");

Console.WriteLine(await priced.Select("OrderId", "Region", "Revenue", "Quarter").Head(4).TextTableAsync(4));

Compute adds a column from an expression string. The language covers arithmetic, comparison, boolean logic, CASE WHEN, and about forty functions across math, dates, and strings, QUARTER among them.

ValidateExpression checks an expression without running it. You don't strictly need it here, since the expression is sitting right there in the file. The moment one comes from configuration or from something a user typed, that check is what you want standing between them and a stack trace.

  |     OrderId|     Region|                Revenue|     Quarter
 0|        1001|      North|                  59.97|           1
 1|        1002|      South|                    149|           1
 2|        1003|      North|                 239.88|           1
 3|        1004|       East|     234.97916666666669|           2

Look at that fourth row. It's the order whose price we filled with the mean, and the long decimal is the fill showing through. Nothing went wrong, but that number is a guess wearing a lot of digits. Round a filled column, or leave those rows out, depending on what the number is for.

Group, sort, and name the output columns

One filter, one group, one sort, and we have the answer we wanted.

var byRegion = priced
    .Where("Quantity", 2, ComparisonOperator.GreaterThan)
    .GroupByAggregate("Region", new Dictionary<string, AggregationType>
    {
        { "Revenue", AggregationType.Sum },
        { "Quantity", AggregationType.Sum },
        { "OrderId", AggregationType.Count }
    })
    .Sort(SortDirection.Descending, "sum_Revenue");

Console.WriteLine(await byRegion.TextTableAsync(20));

Where takes a column, a value, and a comparison: Equals, NotEquals, GreaterThan, GreaterThanOrEqual, LessThan, LessThanOrEqual, Contains, StartsWith, EndsWith. Chain calls for an AND, and use WhereIn when you have a set of values.

The sort there refers to a column that did not exist a line earlier, which catches everybody once. GroupByAggregate names its output columns from the aggregation:

Aggregation	Output column
`Count`	`count_<column>`
`Sum`	`sum_<column>`
`Mean`	`avg_<column>`
`Min`, `Max`	`min_<column>`, `max_<column>`
`StandardDeviation`, `Variance`	`std_<column>`, `var_<column>`

  |     Region|           sum_Revenue|     sum_Quantity|     count_OrderId
 0|      North|                882.85|               50|                 5
 1|       East|     294.4791666666667|               19|                 2
 2|       West|                159.92|                8|                 1
 3|      South|                118.21|               13|                 2

Write it as Parquet

Last step.

using Datafication.Sinks.Connectors.ParquetConnector;

var parquet = await byRegion.ParquetSinkAsync();
await File.WriteAllBytesAsync("revenue-by-region.parquet", parquet);
Console.WriteLine($"Wrote revenue-by-region.parquet, {parquet.Length} bytes");

ParquetSinkAsync returns byte[], not a path. It takes an optional compression method and defaults to Snappy. Writing the bytes is your call, and that's on purpose: when the destination is a stream, a blob, or an HTTP response instead of a file, you don't want a sink that insists on a filename.

Wrote revenue-by-region.parquet, 880 bytes

Reading it back adds a column

LoadParquetAsync takes a Uri, so a local file goes in as new Uri(Path.GetFullPath("revenue-by-region.parquet")), with using Datafication.Extensions.Connectors.ParquetConnector; for the extension method. The block that comes back has one column more than you wrote: the connector adds RowGroup, holding the Parquet row group each row came from. Drop it with RemoveColumn("RowGroup") or leave it out with Select, and don't be surprised when a round trip fails a column-count assertion.

The whole program

Here it is in one piece, in the order you'd run it.

using Datafication.Core.Data;
using Datafication.Core.Sinks;
using Datafication.Extensions.Connectors.CsvConnector;
using Datafication.Sinks.Connectors.ParquetConnector;

var orders = await DataBlock.Connector.LoadCsvAsync("orders.csv");
Console.WriteLine($"Loaded {orders.RowCount} rows");

var info = orders.Info();
Console.WriteLine(await info.TextTableAsync(10));

var clean = orders
    .DropDuplicates(KeepDuplicateMode.First, "OrderId")
    .FillNulls(FillMethod.Mean, "UnitPrice")
    .DropNulls(DropNullMode.Any);

Console.WriteLine($"{clean.RowCount} rows after cleaning");

if (!clean.ValidateExpression("Quantity * UnitPrice", out var error))
{
    Console.WriteLine($"Bad expression: {error}");
    return;
}

var priced = clean
    .Compute("Revenue", "Quantity * UnitPrice")
    .Compute("Quarter", "QUARTER(OrderDate)");

Console.WriteLine(await priced.Select("OrderId", "Region", "Revenue", "Quarter").Head(4).TextTableAsync(4));

var byRegion = priced
    .Where("Quantity", 2, ComparisonOperator.GreaterThan)
    .GroupByAggregate("Region", new Dictionary<string, AggregationType>
    {
        { "Revenue", AggregationType.Sum },
        { "Quantity", AggregationType.Sum },
        { "OrderId", AggregationType.Count }
    })
    .Sort(SortDirection.Descending, "sum_Revenue");

Console.WriteLine(await byRegion.TextTableAsync(20));

var parquet = await byRegion.ParquetSinkAsync();
await File.WriteAllBytesAsync("revenue-by-region.parquet", parquet);
Console.WriteLine($"Wrote revenue-by-region.parquet, {parquet.Length} bytes");

dotnet run, and the four output blocks above are what you get.

Two things that catch people

Load into a variable before you chain. await binds to the whole expression, so await DataBlock.Connector.LoadCsvAsync("orders.csv").Select("Region") does not compile. Assign the load, then chain off the variable. I still get this one wrong more often than I'd like.

The other one is that nothing here mutated anything. orders still holds all fourteen rows at the end of the program, duplicate and nulls included, because DropDuplicates, Compute, Where and the rest each return a new block. That's what makes it safe to keep the raw block around and branch several pipelines off it. It's also why a long chain on a large block is worth a thought: each step is a new block until the last one goes out of scope.

Point it at one of your own CSVs and see how far you get before the file surprises you. If a step you expected to find isn't there, that's the note I want.