| name | add-datalake-consumer |
| description | Adds an event consumer that writes to Azure Data Lake (Parquet) following BI_SALES_RISK plan. Creates events/consumers/[Name]DataLakeCollector.ts subscribing to RabbitMQ, building Parquet rows, writing to /path_prefix/year=YYYY/month=MM/day=DD/. Use when adding DataLakeCollector in logging or similar “event to Data Lake” pipelines. |
Add Data Lake Consumer
Event consumer that subscribes to RabbitMQ and writes to Azure Data Lake (Parquet). Pattern: logging’s DataLakeCollector for risk.evaluated (BI_SALES_RISK_IMPLEMENTATION_PLAN §3.5, §9.1). BI Sales Risk: Paths and Parquet columns MUST match documentation/requirements/BI_SALES_RISK_DATA_LAKE_LAYOUT.md (§2.1 risk.evaluated, §2.2 ml_outcomes, §4 config).
1. Consumer
Path: src/events/consumers/[Name]DataLakeCollector.ts
EventConsumer with queue, exchange: coder_events, bindings: e.g. ['risk.evaluated','ml.prediction.completed','opportunity.updated','forecast.generated'].- Handler: map event to row. For risk.evaluated use columns in Data Lake Layout §2.1. Build path:
{path_prefix}/year={YYYY}/month={MM}/day={DD}/... (Layout §1).
- Write via
@azure/storage-blob (BlockBlob) or @azure/storage-blob + parquetjs (or Arrow) for Parquet. Buffer/batch by time or count if needed.
- Config:
data_lake.connection_string, data_lake.container, data_lake.path_prefix (e.g. /risk_evaluations).
2. Config
config/default.yaml:
data_lake:
connection_string: ${DATA_LAKE_CONNECTION_STRING}
container: ${DATA_LAKE_CONTAINER:-risk}
path_prefix: ${DATA_LAKE_PATH_PREFIX:-/risk_evaluations}
rabbitmq:
url: ${RABBITMQ_URL}
exchange: coder_events
queue: [module]_data_lake
bindings:
- risk.evaluated
- ml.prediction.completed
config/schema.json: add data_lake with connection_string, container, path_prefix.
3. Server
In server.ts: await dataLakeCollector.start() after RabbitMQ connect.
4. Checklist
