246 years of weather in one .dfc file

I had been looking for an excuse to point the S3 connector at something big and public, and NOAA handed me one. Its Global Historical Climatology Network collects daily weather observations from land stations around the world, and the daily archive sits on a public S3 bucket that anyone can read without an account. One station in it, ITE00100554 in Milan, has a record that starts on 1 January 1763. That is not a typo.

That station's entire history is a single CSV object in the bucket, 8,048,041 bytes of it. Below I read it straight from S3, write it into a .dfc file, and start asking it questions: what is actually in the file, how the mean daily maximum moved by decade, which ten days were the hottest, and what a rolling year looks like for two elements at once. Every command is here in the order I ran it, with the output it printed. There is no database and no ETL server anywhere in this. It is one console project, four packages, and about ninety lines.

Where the data comes from

GHCN-Daily is listed on the AWS Registry of Open Data at registry.opendata.aws/noaa-ghcn. The bucket is noaa-ghcn-pds in us-east-1 and it is readable anonymously, so nothing in this post needs an access key. That is a big part of why I picked it: you can run every line here without signing up for anything.

The bucket's own readme.txt describes two CSV layouts. by_year holds one file per year covering every station, and by_station holds one file per station covering its whole period of record. Both use the same eight columns:

ID,DATE,ELEMENT,DATA_VALUE,M_FLAG,Q_FLAG,S_FLAG,OBS_TIME
ITE00100554,17630101,TMAX,-36,,,E,

ELEMENT says what was measured and DATA_VALUE is the measurement. The readme gives you the units: TMAX and TMIN are maximum and minimum temperature in tenths of degrees Celsius, and PRCP is precipitation in tenths of a millimeter. So that first row is a maximum of minus 3.6 degrees on 1 January 1763.

Pick your run size on purpose. These are the object sizes the bucket returned when I wrote this:

I used the station file. It is small enough to run in a few seconds on a laptop and long enough that the questions get interesting, which is the combination I wanted. The same code works on any of the others by changing one string.

The station's own line in the bucket's ghcnd-stations.txt fills in the rest: ITE00100554, latitude 45.4717, longitude 9.1892, elevation 150 meters, name MILAN.

What you need

.NET 8.0 or later, on Windows, macOS, or Linux, and a new console project:

dotnet new console -o ghcn
cd ghcn
dotnet add package Datafication.Core
dotnet add package Datafication.S3Connector
dotnet add package Datafication.Storage.Velocity
dotnet add package Datafication.ParquetConnector

That resolves to 1.0.9 of each. The S3 connector brings AWSSDK.S3 along with it. The SDK is free to use for organizations under five developers and under $500,000 in annual revenue, and for open-source projects under the same funding limit. The details are on the SDK product page.

Step one: look before you load

The temptation with an unfamiliar object is to load it into something and see what happens. Look at it first. The S3 connector takes a configuration, hands back a DataBlock, and Info() prints the schema the loader inferred.

using Datafication.Connectors.S3Connector;
using Datafication.Core.Data;
using Datafication.Core.Sinks;

var s3 = new S3ConnectorConfiguration
{
    Region = "us-east-1",
    BucketName = "noaa-ghcn-pds",
    ObjectKey = "csv/by_station/ITE00100554.csv"
};

var connector = new S3DataConnector(s3);
var peek = await connector.GetDataAsync();
Console.WriteLine(await peek.Info().TextTableAsync());
Console.WriteLine(await peek.Head(5).TextTableAsync(5));
peek.Dispose();
connector.Dispose();

There are no credentials anywhere in that. Leaving AccessKeyId and SecretAccessKey null is what makes the read anonymous. The connector looks at the key's extension, sees .csv, and hands the parsing to the CSV connector. That is the shape of the whole connector set, which Seven connectors, one shape covers in more detail.

  |         Column|          Label|       Type|     Non-Null Count|     Null Count
 0|             ID|             ID|     String|             234470|              0
 1|           DATE|           DATE|      Int32|             234470|              0
 2|        ELEMENT|        ELEMENT|     String|             234470|              0
 3|     DATA_VALUE|     DATA_VALUE|      Int32|             234470|              0
 4|         M_FLAG|         M_FLAG|     String|                  0|         234470
 5|         Q_FLAG|         Q_FLAG|     String|                991|         233479
 6|         S_FLAG|         S_FLAG|     String|             234470|              0
 7|       OBS_TIME|       OBS_TIME|     String|             234470|              0

  |              ID|         DATE|     ELEMENT|     DATA_VALUE|     M_FLAG|     Q_FLAG|     S_FLA...
 0|     ITE00100554|     17630101|        TMAX|            -36|     <null>|     <null>|          ...
 1|     ITE00100554|     17630102|        TMAX|            -26|     <null>|     <null>|          ...
 2|     ITE00100554|     17630103|        TMAX|             -9|     <null>|     <null>|          ...
 3|     ITE00100554|     17630104|        TMAX|             -4|     <null>|     <null>|          ...
 4|     ITE00100554|     17630105|        TMAX|             21|     <null>|     <null>|          ...

234,470 rows, and DATE and DATA_VALUE both came back as Int32, which is what I was hoping for: dates in this file are integers of the form YYYYMMDD, and arithmetic on an integer is cheap. Two of the four flag columns are worth a look before you build anything on them. M_FLAG is empty in every single row of this station's record, and Q_FLAG, the quality flag, is set on 991 rows out of 234,470.

Step two: write the object into a .dfc file

The peek pulled the whole object into memory, which is fine for eight megabytes and not fine as a habit. The streaming path never does that. GetStorageDataAsync takes a storage target and a batch size, reads the object in batches, and appends each batch to the target as it goes.

using Datafication.Storage.Velocity;

var options = new VelocityOptions
{
    DefaultCompression = VelocityCompressionType.LZ4,
    AutoCompactionEnabled = false
};

using (var store = new VelocityDataBlock("milan.dfc", options))
{
    var loader = new S3DataConnector(s3);
    await loader.GetStorageDataAsync(store, batchSize: 50000);
    await store.FlushAsync();
    loader.Dispose();

    var stats = await store.GetStorageStatsAsync();
    Console.WriteLine($"{stats.ActiveRows} rows, {stats.StorageFiles} files, {stats.EstimatedSizeBytes} bytes");
}

234470 rows, 5 files, 4799122 bytes

Five files, because five batches of 50,000 rows produced five segments, and each segment is its own file sitting next to the one you named. GetStorageStatsAsync counts them, and they add up to exactly what is on disk:

The last two rows come from the one-call form, which takes a DataBlock you already have in memory and writes it in a single shot:

var saved = await VelocityDataBlock.SaveAsync("milan-one-file.dfc", peek, options);
await saved.FlushAsync();
saved.Dispose();

Two numbers in that table caught me off guard. The first is how much of the work compression is doing. DefaultCompression is None unless you set it, and an uncompressed DFC file of this data comes out more than twice the size of the CSV it came from. That is not a defect, it is the trade you are making. Four of the eight columns are text that is empty or nearly empty in the CSV, where an empty field costs you one comma, and in a columnar file every row takes a slot in every column whether it holds anything or not. The second is that the same rows cost 9,037 bytes more spread across five segments than written in one.

The layout inside each of those files, and why it is shaped that way, is the subject of Designing DFC.

Step three: open the file and count what is in it

From here on the object storage is done with. OpenAsync opens the file, and from that point every question is a plan you build with method calls and run with Execute().

var block = await VelocityDataBlock.OpenAsync("milan.dfc");
Console.WriteLine($"{block.RowCount} rows");

var byElement = block
    .GroupByAggregate("ELEMENT", "DATA_VALUE", AggregationType.Count)
    .Sort(SortDirection.Descending, "count_DATA_VALUE")
    .Execute();
Console.WriteLine(await byElement.TextTableAsync());
block.ClearQueryPlan();

var firstDay = block.Min("DATE").Execute();
block.ClearQueryPlan();
var lastDay = block.Max("DATE").Execute();
block.ClearQueryPlan();
Console.WriteLine(await firstDay.TextTableAsync(1));
Console.WriteLine(await lastDay.TextTableAsync(1));

234470 rows
  |     ELEMENT|     count_DATA_VALUE
 0|        TMIN|                89697
 1|        TMAX|                89686
 2|        PRCP|                55087

  |     Column|        Value
 0|       DATE|     17630101

  |     Column|        Value
 0|       DATE|     20081201

Three elements and 246 calendar years. The result column is called count_DATA_VALUE because GroupByAggregate names its output {prefix}_{column}, with the prefix taken from the aggregation. The two temperature series start together in 1763, and precipitation does not show up until 1858.

ClearQueryPlan() is the one piece of bookkeeping in all of this. The plan accumulates on the block, so a block you reuse for a second question needs the first one cleared off it. Miss it and your second question is still carrying the first one around.

Step four: the record, decade by decade

The question worth asking of a 246-year record is how it moved. That needs two derived columns, a decade and a temperature in real degrees, and this is where the boundary between the file and the result starts to matter.

var tmax = block
    .Where("ELEMENT", "TMAX", ComparisonOperator.Equals)
    .Select("DATE", "DATA_VALUE")
    .Execute();
block.ClearQueryPlan();

var decades = tmax
    .Compute("Decade", "FLOOR(DATE / 100000) * 10")
    .Compute("TempC", "DATA_VALUE / 10.0")
    .GroupByAggregate("Decade", new Dictionary<string, AggregationType>
    {
        { "TempC", AggregationType.Mean },
        { "DATA_VALUE", AggregationType.Count }
    })
    .Sort(SortDirection.Ascending, "Decade");
Console.WriteLine(await decades.TextTableAsync(25));

The filter and the projection run against the file, which is the cheap part: 89,686 rows of the 234,470 come back, two columns wide. The two computed columns and the grouping then run on the DataBlock that Execute() handed back.

That split is not a style choice, and it tripped me up the first time. Every method in the query plan names columns that have to exist in the file, so a column you create inside a plan is not visible to a later step of the same plan. Compute puts its column on the result. Group it, sort it, or window it there.

The expression language behind Compute is arithmetic, comparison, logic, CASE WHEN, and about forty built-in functions, FLOOR among them. Dates being integers of the form YYYYMMDD is the whole reason FLOOR(DATE / 100000) * 10 gives you a decade. Group by FLOOR(DATE / 10000) instead and the same six lines give you the per-year series drawn at the top of this post.

Here are the 25 rows it prints, with avg_TempC rounded to two decimals:

Two of those bars are not ten years. The 1760s bar is 1763 to 1769, and the 2000s bar stops on 30 November 2008, where the station's record in this file ends. The eight complete years from 2000 to 2007 average 20.52 degrees over 2,862 days, so the shape of the last bar survives the correction. It just is not a decade. Everything in between is dense: no year in the record has fewer than 300 daily maxima in it.

The bar I keep coming back to is the 1810s. At 15.15 degrees it is the coolest full decade in the record, cooler than the 1800s at 15.52 and the 1820s at 15.86 sitting on either side of it. This file will not tell you why, and I am not going to guess at it in a post about file formats. That is the fun of a record this long, though. You notice something, and then you go find out.

Step five: the ten hottest days

Sorting and taking the top of a sort is a plan like any other, and it is the clearest place to see why you want the file doing the work rather than memory, because only ten rows ever have to come back.

var hottest = block
    .Where("ELEMENT", "TMAX", ComparisonOperator.Equals)
    .Select("DATE", "DATA_VALUE")
    .Sort(SortDirection.Descending, "DATA_VALUE")
    .Head(10)
    .Execute()
    .Compute("TempC", "DATA_VALUE / 10.0");
Console.WriteLine(await hottest.TextTableAsync(10));
block.ClearQueryPlan();

  |         DATE|     DATA_VALUE|     TempC
 0|     20030811|            411|      41.1
 1|     20030805|            403|      40.3
 2|     20060721|            403|      40.3
 3|     20060722|            403|      40.3
 4|     20030812|            402|      40.2
 5|     20030810|            401|      40.1
 6|     20030804|            398|      39.8
 7|     20030806|            396|      39.6
 8|     20030813|            396|      39.6
 9|     20060723|            395|      39.5

Ten days out of 89,686, and they come from two spells: seven days in August 2003 and three in July 2006. The Compute on the end converts the ten rows that came back, not the 89,686 that were scanned.

The filter is written as Where("ELEMENT", "TMAX", ComparisonOperator.Equals) rather than a pattern match on purpose. Equality is the form the file can answer straight out of its string dictionary. WhereContains, WhereStartsWith and WhereEndsWith are there when you need them, and they work through the rows.

Step six: a rolling year, one window per element

A 365-day moving average is the usual way to look at a daily series without the seasons drowning out everything else. The catch here is that the file interleaves elements. For most dates there is a TMAX row and a TMIN row, so a window that just walks the rows in order would happily average maxima and minima together. That is what partitionByColumns is for.

var since1900 = block
    .Where("DATE", 19000101, ComparisonOperator.GreaterThanOrEqual)
    .Where("ELEMENT", "PRCP", ComparisonOperator.NotEquals)
    .Select("DATE", "ELEMENT", "DATA_VALUE")
    .Execute();
block.ClearQueryPlan();

var rolling = since1900
    .Compute("TempC", "DATA_VALUE / 10.0")
    .Window("TempC", WindowFunctionType.MovingAverage, windowSize: 365,
            resultColumnName: "Rolling365C", orderByColumn: "DATE",
            partitionByColumns: new[] { "ELEMENT" })
    .Select("DATE", "ELEMENT", "TempC", "Rolling365C");
Console.WriteLine($"{rolling.RowCount} rows windowed");
Console.WriteLine(await rolling.Tail(4).TextTableAsync(4));

79442 rows windowed
  |         DATE|     ELEMENT|     TempC|            Rolling365C
 0|     20081127|        TMIN|       3.3|     13.466849315068567
 1|     20081128|        TMIN|       1.1|      13.44493150684939
 2|     20081129|        TMIN|       1.9|     13.429589041095964
 3|     20081130|        TMIN|       3.1|     13.413972602739799

Two filters, both against columns that exist in the file, cut 234,470 rows down to the 79,442 temperature rows since 1900. The window then runs per element and orders each partition by DATE, so Rolling365C on the last TMIN row is the mean minimum of the previous year of minima, 13.41 degrees. On the last TMAX row it is 21.58. Results come back partition by partition, which is why the tail of this frame is all TMIN.

Step seven: hand the answer to something else

The answer is 25 rows. Somebody else wants it in the tools they already use, which in practice means Parquet.

var parquet = await decades.ParquetSinkAsync();
await File.WriteAllBytesAsync("milan-decades.parquet", parquet);
Console.WriteLine($"{parquet.Length} bytes of Parquet");

1006 bytes of Parquet

The sink hands back a byte[] instead of writing a file itself, so those bytes can go to a stream, a blob, or an HTTP response without ever touching a disk.

What to watch for

The whole program above runs in about 6.5 seconds on my machine, an iMac with 16 logical cores running macOS 15.7.4 and .NET 8.0.23, and roughly half of that is the two reads of the object across the network. That is one run on one machine and not a benchmark, which is the only honest way to read a number like it.

Four things I would want to know before pointing this at my own data.

Dispose the block before you move or delete the file. Velocity memory-maps its files, and on Windows a File.Move or a delete can fail with IOException or UnauthorizedAccessException while the block or a cursor is still alive. Dispose first, and retry with backoff if you are racing something else.

Match the method to the key. GetDataAsync() reads one object and throws NotSupportedException if the key turns out to be a prefix. Prefixes need GetStorageDataAsync and AllowMultipleSegments = true.

Take the batch size seriously. It is the memory ceiling of the load, and it decides how many segments you end up with. 50,000 rows on this object gives five segments. A larger batch gives you fewer, larger ones.

Check the flags before you trust a row. GHCN publishes quality flags in the file and this post ignores them, which is fine for the shape of a curve and not fine for a claim about one particular day.

The whole program

using Datafication.Connectors.S3Connector;
using Datafication.Core.Data;
using Datafication.Core.Sinks;
using Datafication.Sinks.Connectors.ParquetConnector;
using Datafication.Storage.Velocity;

var s3 = new S3ConnectorConfiguration
{
    Region = "us-east-1",
    BucketName = "noaa-ghcn-pds",
    ObjectKey = "csv/by_station/ITE00100554.csv"
};

// 1. Look before you load.
var connector = new S3DataConnector(s3);
var peek = await connector.GetDataAsync();
Console.WriteLine(await peek.Info().TextTableAsync());
Console.WriteLine(await peek.Head(5).TextTableAsync(5));
peek.Dispose();
connector.Dispose();

// 2. Write the whole object to DFC, 50,000 rows at a time.
var options = new VelocityOptions
{
    DefaultCompression = VelocityCompressionType.LZ4,
    AutoCompactionEnabled = false
};

using (var store = new VelocityDataBlock("milan.dfc", options))
{
    var loader = new S3DataConnector(s3);
    await loader.GetStorageDataAsync(store, batchSize: 50000);
    await store.FlushAsync();
    loader.Dispose();

    var stats = await store.GetStorageStatsAsync();
    Console.WriteLine($"{stats.ActiveRows} rows, {stats.StorageFiles} files, {stats.EstimatedSizeBytes} bytes");
}

// 3. Open the file and count what is in it.
var block = await VelocityDataBlock.OpenAsync("milan.dfc");
Console.WriteLine($"{block.RowCount} rows");

var byElement = block
    .GroupByAggregate("ELEMENT", "DATA_VALUE", AggregationType.Count)
    .Sort(SortDirection.Descending, "count_DATA_VALUE")
    .Execute();
Console.WriteLine(await byElement.TextTableAsync());
block.ClearQueryPlan();

var firstDay = block.Min("DATE").Execute();
block.ClearQueryPlan();
var lastDay = block.Max("DATE").Execute();
block.ClearQueryPlan();
Console.WriteLine(await firstDay.TextTableAsync(1));
Console.WriteLine(await lastDay.TextTableAsync(1));

// 4. Every daily maximum, by decade.
var tmax = block
    .Where("ELEMENT", "TMAX", ComparisonOperator.Equals)
    .Select("DATE", "DATA_VALUE")
    .Execute();
block.ClearQueryPlan();

var decades = tmax
    .Compute("Decade", "FLOOR(DATE / 100000) * 10")
    .Compute("TempC", "DATA_VALUE / 10.0")
    .GroupByAggregate("Decade", new Dictionary<string, AggregationType>
    {
        { "TempC", AggregationType.Mean },
        { "DATA_VALUE", AggregationType.Count }
    })
    .Sort(SortDirection.Ascending, "Decade");
Console.WriteLine(await decades.TextTableAsync(25));

// 5. The ten hottest days in the record.
var hottest = block
    .Where("ELEMENT", "TMAX", ComparisonOperator.Equals)
    .Select("DATE", "DATA_VALUE")
    .Sort(SortDirection.Descending, "DATA_VALUE")
    .Head(10)
    .Execute()
    .Compute("TempC", "DATA_VALUE / 10.0");
Console.WriteLine(await hottest.TextTableAsync(10));
block.ClearQueryPlan();

// 6. A rolling year, one window per element.
var since1900 = block
    .Where("DATE", 19000101, ComparisonOperator.GreaterThanOrEqual)
    .Where("ELEMENT", "PRCP", ComparisonOperator.NotEquals)
    .Select("DATE", "ELEMENT", "DATA_VALUE")
    .Execute();
block.ClearQueryPlan();

var rolling = since1900
    .Compute("TempC", "DATA_VALUE / 10.0")
    .Window("TempC", WindowFunctionType.MovingAverage, windowSize: 365,
            resultColumnName: "Rolling365C", orderByColumn: "DATE",
            partitionByColumns: new[] { "ELEMENT" })
    .Select("DATE", "ELEMENT", "TempC", "Rolling365C");
Console.WriteLine($"{rolling.RowCount} rows windowed");
Console.WriteLine(await rolling.Tail(4).TextTableAsync(4));

// 7. Hand the answer to something else.
var parquet = await decades.ParquetSinkAsync();
await File.WriteAllBytesAsync("milan-decades.parquet", parquet);
Console.WriteLine($"{parquet.Length} bytes of Parquet");

block.Dispose();

Ninety-odd lines, four packages, and a 4.8 MB file that answers questions about 246 years of weather in about the time it takes to print them. What I like about it is what is missing: no server to stand up, no schema to declare in advance, no import step living separately from the program that asks the questions. The file is the database, and it sits right next to the code that reads it. If you point this at a station of your own and something in the numbers surprises you, that is the kind of thing I would like to hear about.