377 lines
16 KiB
Markdown
377 lines
16 KiB
Markdown
# Guide 9: MinIO Data Lake with Parquet
|
|
|
|
## What Are MinIO, Parquet, and a Data Lake?
|
|
|
|
### MinIO — S3-Compatible Object Storage
|
|
|
|
**Object storage** is a way to store files (called "objects") in a flat structure identified by keys (like file paths). Unlike a filesystem with directories, object storage uses a single "bucket" (container) and key strings like `observations/2026/06/24/partition-0-offset-0000001000.parquet`.
|
|
|
|
**Amazon S3** is the most widely-used object storage service. **MinIO** is an open-source server that implements the same S3 API, so you can develop locally with MinIO and deploy to AWS S3 later without changing your code. The application uses the Minio .NET SDK, which speaks the S3 protocol.
|
|
|
|
### Parquet — Columnar File Format
|
|
|
|
**Parquet** is a file format designed for analytics. While JSON and CSV store data row-by-row, Parquet stores data column-by-column:
|
|
|
|
```
|
|
CSV/JSON (row-oriented): Parquet (column-oriented):
|
|
┌─────┬────────┬───────┐ ┌─────────────────────────┐
|
|
│ id │ code │ value │ │ id: [1, 2, 3, 4, ...] │
|
|
├─────┼────────┼───────┤ │ code: [HR, BP, HR, ...] │
|
|
│ 1 │ HR │ 80 │ │ value: [80, 120, 85, ...]│
|
|
│ 2 │ BP │ 120 │ └─────────────────────────┘
|
|
│ 3 │ HR │ 85 │
|
|
└─────┴────────┴───────┘
|
|
```
|
|
|
|
Why columnar? Analytics queries usually read a few columns from many rows ("give me all heart rate values"). Columnar storage lets the query engine read only the `code` and `value` columns, skipping everything else. Parquet also compresses data within each column (similar values compress well), so files are much smaller than JSON — often 10-50x.
|
|
|
|
Tools like Apache Spark, AWS Athena, Pandas, and DuckDB can read Parquet files natively and query them with SQL.
|
|
|
|
### Data Lake — Long-Term Analytics Storage
|
|
|
|
A **data lake** is a centralized storage repository where you dump raw data in its original form for later analysis. Unlike a database (which is optimized for real-time transactional queries), a data lake is optimized for batch analytics — "analyze all observations from the past 6 months" or "what percentage of critical alerts were acknowledged within 5 minutes across all departments?"
|
|
|
|
The typical pattern: application databases hold recent data for real-time operations, while the data lake holds historical data for research, compliance auditing, and machine learning.
|
|
|
|
---
|
|
|
|
## Why a Data Lake in This Project?
|
|
|
|
PostgreSQL and Elasticsearch serve real-time operational needs. But clinical data has long-term value — research on sepsis detection accuracy, compliance audits, training machine learning models. The data lake captures a complete, immutable history of all events in a format optimized for large-scale analytics, stored cheaply in object storage.
|
|
|
|
---
|
|
|
|
## Architecture Overview
|
|
|
|
```
|
|
Kafka Topics DataLakeWriterService MinIO (S3)
|
|
┌─────────────────┐ ┌──────────────────┐
|
|
│observation. │─┐ │ vigilcare/ │
|
|
│recorded │ │ ┌──────────────────────┐ │ │
|
|
├─────────────────┤ ├────►│ In-memory buffer │ │ observations/ │
|
|
│alert. │ │ │ (by topic + date + │ flush │ 2026/06/24/ │
|
|
│generated │ │ │ partition) │────────►│ part-0.pqt │
|
|
├─────────────────┤ │ │ │ │ │
|
|
│encounter.status. │─┘ │ Flush when: │ │ alerts/ │
|
|
│changed │ │ - 1000 events buffer │ │ 2026/06/24/ │
|
|
└─────────────────┘ │ - 5 min elapsed │ │ part-0.pqt │
|
|
└──────────────────────┘ │ │
|
|
│ encounters/ │
|
|
│ 2026/06/24/ │
|
|
│ part-0.pqt │
|
|
└──────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Configuration
|
|
|
|
### MinIO Client
|
|
|
|
```csharp
|
|
public static class MinioClientFactory
|
|
{
|
|
public static IMinioClient Build(MinioOptions opts)
|
|
{
|
|
var client = new MinioClient()
|
|
.WithEndpoint(opts.Endpoint)
|
|
.WithCredentials(opts.AccessKey, opts.SecretKey);
|
|
if (opts.UseSSL) client = client.WithSSL();
|
|
return client.Build();
|
|
}
|
|
}
|
|
```
|
|
|
|
```csharp
|
|
public sealed class MinioOptions
|
|
{
|
|
public string Endpoint { get; init; } = "localhost:9005";
|
|
public string AccessKey { get; init; } = "minioadmin";
|
|
public string SecretKey { get; init; } = "minioadmin";
|
|
public string BucketName { get; init; } = "vigilcare";
|
|
public bool UseSSL { get; init; } = false;
|
|
}
|
|
```
|
|
|
|
### Data Lake Options
|
|
|
|
```csharp
|
|
public sealed class DataLakeOptions
|
|
{
|
|
public int FlushCount { get; init; } = 1_000; // max events before forced flush
|
|
public int FlushIntervalSeconds { get; init; } = 300; // max time before forced flush (5 min)
|
|
public string BucketName { get; init; } = "vigilcare";
|
|
}
|
|
```
|
|
|
|
---
|
|
|
|
## The DataLakeWriterService
|
|
|
|
This is a Kafka consumer that buffers events in memory and flushes them to MinIO as Parquet files.
|
|
|
|
### Consumer Setup
|
|
|
|
```csharp
|
|
protected override async Task ExecuteAsync(CancellationToken ct)
|
|
{
|
|
var consumerConfig = new ConsumerConfig
|
|
{
|
|
BootstrapServers = _kafkaOptions.BootstrapServers,
|
|
GroupId = "data-lake-writer",
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|
EnableAutoCommit = false,
|
|
};
|
|
|
|
using var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
|
|
consumer.Subscribe(new[]
|
|
{
|
|
_kafkaOptions.Topics.ObservationRecorded,
|
|
_kafkaOptions.Topics.AlertGenerated,
|
|
_kafkaOptions.Topics.EncounterStatusChanged,
|
|
});
|
|
```
|
|
|
|
### Buffering Strategy
|
|
|
|
Instead of writing one Parquet file per event (too many small files), the service buffers events and flushes in batches:
|
|
|
|
```csharp
|
|
// Buffer key groups events into one Parquet file
|
|
private record BufferKey(string Topic, string DatePath, int Partition);
|
|
|
|
// Track the highest offset per partition for safe commits
|
|
private readonly Dictionary<TopicPartition, TopicPartitionOffset> _highWatermarks = new();
|
|
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
var result = consumer.Consume(TimeSpan.FromMilliseconds(500));
|
|
|
|
if (result is not null)
|
|
AddToBuffer(result);
|
|
|
|
var totalBuffered = _buffer.Values.Sum(v => v.Count);
|
|
var shouldFlushCount = totalBuffered >= _opts.FlushCount; // 1000 events
|
|
var shouldFlushTime = elapsed >= TimeSpan.FromSeconds(300); // 5 minutes
|
|
|
|
if ((shouldFlushCount || shouldFlushTime) && totalBuffered > 0)
|
|
{
|
|
await FlushAsync(consumer, ct);
|
|
lastFlush = DateTimeOffset.UtcNow;
|
|
}
|
|
}
|
|
```
|
|
|
|
Events are grouped by **(topic, date, partition)**. All heart rate observations from June 24th on Kafka partition 0 end up in one Parquet file: `observations/2026/06/24/partition-0-offset-0000001000.parquet`.
|
|
|
|
### Date Partitioning
|
|
|
|
The date in the file path comes from the event's timestamp (e.g., `recordedAt` for observations), not the wall clock. This is important: analytics tools like AWS Athena and Apache Spark can "partition-prune" — when you query "observations from June 24th," they only read files from the `2026/06/24/` folder, skipping everything else:
|
|
|
|
```csharp
|
|
public static string ExtractDatePath(string topic, string payload, KafkaTopicOptions topics)
|
|
{
|
|
var ts = topic switch
|
|
{
|
|
var t when t == topics.ObservationRecorded => GetTimestamp(doc, "recordedAt"),
|
|
var t when t == topics.AlertGenerated => GetTimestamp(doc, "triggeredAt"),
|
|
var t when t == topics.EncounterStatusChanged => GetTimestamp(doc, "changedAt"),
|
|
_ => DateTimeOffset.UtcNow,
|
|
};
|
|
return $"{ts.Year:D4}/{ts.Month:D2}/{ts.Day:D2}";
|
|
}
|
|
```
|
|
|
|
### Partial-Commit Safety
|
|
|
|
**This is the most important safety feature of the data lake writer.** When flushing, some file uploads might succeed while others fail (e.g., MinIO is temporarily unreachable for one partition). The service only commits Kafka offsets for partitions where ALL uploads succeeded:
|
|
|
|
```csharp
|
|
private async Task FlushAsync(IConsumer<string, string> consumer, CancellationToken ct)
|
|
{
|
|
var failedPartitions = new HashSet<int>();
|
|
|
|
foreach (var (key, events) in _buffer)
|
|
{
|
|
try
|
|
{
|
|
var bytes = await BuildParquetAsync(key.Topic, events, key.Partition);
|
|
await UploadToMinioAsync(objectKey, bytes, ct);
|
|
flushedKeys.Add(key);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
failedPartitions.Add(key.Partition);
|
|
}
|
|
}
|
|
|
|
// Only commit offsets for partitions with no failures
|
|
var safeOffsets = _highWatermarks
|
|
.Where(kv => !failedPartitions.Contains(kv.Key.Partition))
|
|
.Select(kv => kv.Value)
|
|
.ToList();
|
|
|
|
if (safeOffsets.Count > 0)
|
|
consumer.Commit(safeOffsets);
|
|
|
|
// Retain failed buffers for retry on next flush
|
|
foreach (var key in flushedKeys)
|
|
_buffer.Remove(key);
|
|
}
|
|
```
|
|
|
|
If partition 2 fails to upload but partitions 0, 1, 3, 4, 5 succeed: offsets for 0, 1, 3, 4, 5 are committed. Partition 2's buffer is retained and retried on the next flush cycle. If the service crashes before the next flush, Kafka re-delivers partition 2's events from the last committed offset — no data is lost.
|
|
|
|
### Shutdown Flush
|
|
|
|
On graceful shutdown, the service flushes any remaining buffered events:
|
|
|
|
```csharp
|
|
finally
|
|
{
|
|
if (_buffer.Values.Sum(v => v.Count) > 0)
|
|
{
|
|
try { await FlushAsync(consumer, CancellationToken.None); }
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex,
|
|
"DataLakeWriter shutdown flush failed — some events may be re-read on next start");
|
|
}
|
|
}
|
|
consumer.Close();
|
|
}
|
|
```
|
|
|
|
`CancellationToken.None` is used instead of the stopping token because the host shutdown might cancel the token before MinIO uploads finish.
|
|
|
|
---
|
|
|
|
## Building Parquet Files
|
|
|
|
The `ParquetFileBuilder` uses the `Parquet.Net` library to create Parquet files in memory:
|
|
|
|
```csharp
|
|
public static async Task<byte[]> BuildObservationsAsync(IReadOnlyList<ObservationRow> rows)
|
|
{
|
|
var schema = new ParquetSchema(
|
|
new DataField<string>("observation_id"),
|
|
new DataField<string>("encounter_id"),
|
|
new DataField<string>("patient_id"),
|
|
new DataField<string>("mrn"),
|
|
new DataField<string>("observation_code"),
|
|
new DataField<double>("value"),
|
|
new DataField<string>("unit"),
|
|
new DataField<string>("source"),
|
|
new DataField<string>("recorded_at"),
|
|
new DataField<int>("kafka_partition"),
|
|
new DataField<long>("kafka_offset")
|
|
);
|
|
|
|
using var ms = new MemoryStream();
|
|
using (var writer = await ParquetWriter.CreateAsync(schema, ms))
|
|
using (var rg = writer.CreateRowGroup())
|
|
{
|
|
var f = schema.DataFields;
|
|
await rg.WriteColumnAsync(new DataColumn(f[0], rows.Select(r => r.ObservationId).ToArray()));
|
|
await rg.WriteColumnAsync(new DataColumn(f[1], rows.Select(r => r.EncounterId).ToArray()));
|
|
// ... one WriteColumnAsync per field
|
|
}
|
|
return ms.ToArray();
|
|
}
|
|
```
|
|
|
|
Parquet is columnar, so you write one column at a time (all observation IDs, then all encounter IDs, etc.), not one row at a time.
|
|
|
|
The `kafka_partition` and `kafka_offset` fields provide **lineage** — given any Parquet row, you can trace it back to the exact Kafka message it came from. This is useful for debugging and auditing.
|
|
|
|
---
|
|
|
|
## Parsing Kafka Events
|
|
|
|
The `DataLakeEventParser` extracts fields from Kafka JSON payloads into typed row objects:
|
|
|
|
```csharp
|
|
public static ObservationRow ParseObservationRow(string payload, long offset, int partition)
|
|
{
|
|
var d = JsonDocument.Parse(payload).RootElement;
|
|
return new ObservationRow(
|
|
ObservationId : GetString(d, "observationId"),
|
|
EncounterId : GetString(d, "encounterId"),
|
|
PatientId : GetString(d, "patientId"),
|
|
ObservationCode : GetString(d, "observationCode", "code"), // fallback name
|
|
Value : GetDouble(d, "value"),
|
|
RecordedAt : GetTimestampString(d, "recordedAt"),
|
|
KafkaPartition : partition,
|
|
KafkaOffset : offset);
|
|
}
|
|
```
|
|
|
|
The parser is tolerant of field name variations (e.g., `"observationCode"` or `"code"`) for backwards compatibility with older event formats.
|
|
|
|
---
|
|
|
|
## Object Key Structure
|
|
|
|
```
|
|
vigilcare/ ← bucket
|
|
├── observations/ ← topic-derived folder
|
|
│ ├── 2026/06/23/ ← date partition from event timestamp
|
|
│ │ ├── partition-0-offset-0000001000.parquet
|
|
│ │ └── partition-3-offset-0000002500.parquet
|
|
│ └── 2026/06/24/
|
|
│ └── partition-0-offset-0000005000.parquet
|
|
├── alerts/
|
|
│ └── 2026/06/24/
|
|
│ └── partition-1-offset-0000000100.parquet
|
|
├── encounters/
|
|
│ └── 2026/06/24/
|
|
│ └── partition-2-offset-0000000050.parquet
|
|
└── discharge-summaries/ ← from DischargeSummaryWorkerService
|
|
└── {encounterId}/
|
|
└── summary.pdf
|
|
```
|
|
|
|
The key encodes: what type of data, when it happened, which Kafka partition, and the starting offset. This makes each file uniquely identifiable and traceable.
|
|
|
|
---
|
|
|
|
## Querying the Data Lake
|
|
|
|
Once data is in MinIO as Parquet files, you can query it using analytics tools:
|
|
|
|
**With DuckDB (local, fast):**
|
|
```sql
|
|
SELECT observation_code, AVG(value), COUNT(*)
|
|
FROM read_parquet('s3://vigilcare/observations/2026/06/*/partition-*.parquet')
|
|
WHERE observation_code = 'HEART_RATE'
|
|
GROUP BY observation_code;
|
|
```
|
|
|
|
**With Pandas (Python):**
|
|
```python
|
|
import pandas as pd
|
|
df = pd.read_parquet('s3://vigilcare/observations/2026/06/24/')
|
|
high_hr = df[df['observation_code'] == 'HEART_RATE'][df['value'] > 120]
|
|
```
|
|
|
|
**With AWS Athena (serverless SQL):**
|
|
```sql
|
|
SELECT COUNT(DISTINCT patient_id)
|
|
FROM vigilcare.observations
|
|
WHERE observation_code = 'SPO2' AND value < 92
|
|
AND recorded_at BETWEEN '2026-06-01' AND '2026-06-30';
|
|
```
|
|
|
|
The date-partitioned folder structure enables **partition pruning** — the query engine only reads files from the date range you're querying, making queries over specific time periods very fast even with years of historical data.
|
|
|
|
---
|
|
|
|
## Key Takeaways
|
|
|
|
- **Buffer-then-flush reduces file count**: One Parquet file per flush (up to 1000 events) instead of one file per event. Fewer, larger files are better for analytics tools.
|
|
- **Date partitioning enables fast queries**: Analytics tools skip irrelevant date folders entirely. Querying "last 7 days" reads 7 folders, not millions of files.
|
|
- **Partial-commit safety prevents data loss**: Only commit Kafka offsets for partitions where all uploads succeeded. Failed partitions are retried.
|
|
- **Kafka lineage enables tracing**: Every Parquet row includes the Kafka partition and offset it came from, creating an audit trail from source to storage.
|
|
- **Parquet + object storage is the industry standard**: This pattern (event stream → columnar files → S3-compatible storage) is how most data platforms work. The same Parquet files work with Spark, Athena, Pandas, DuckDB, and dozens of other tools.
|
|
- **MinIO is a local S3**: Develop against MinIO, deploy to AWS S3 with zero code changes. Only the endpoint and credentials change.
|