JSPM

@hazeljs/data

0.8.2
  • ESM via JSPM
  • ES Module Entrypoint
  • Export Map
  • Keywords
  • License
  • Repository URL
  • TypeScript Types
  • README
  • Created
  • Published
  • Downloads 393
  • Score
    100M100P100Q96670F
  • License Apache-2.0

Data Processing & ETL for HazelJS framework

Package Exports

  • @hazeljs/data
  • @hazeljs/data/dist/index.js

This package does not declare an exports field, so the exports above have been automatically detected and optimized by JSPM instead. If any package subpath is missing, it is recommended to post an issue to the original package (@hazeljs/data) to support the "exports" field. If that is not possible, create a JSPM override to customize the exports field for this package.

Readme

@hazeljs/data

Data Processing & ETL for HazelJS - pipelines, schema validation, streaming, data quality, and more.

npm version npm downloads License: Apache-2.0

Features

  • Pipelines – Declarative ETL with @Pipeline, @Transform, @Validate decorators
  • Schema validation – Fluent Schema API (string, number, boolean, date, object, array, literal, union) with .optional(), .nullable(), .default(), .transform(), .refine(), Infer<T>, .toJsonSchema()
  • Pipeline options – Conditional steps (when), per-step retry, timeout, dead letter queue (DLQ)
  • PipelineBuilder – Programmatic pipelines with .branch(), .parallel(), .catch(), .toSchema()
  • ETL service – Execute multi-step pipelines with executeBatch, onStepComplete
  • Stream processing – StreamService, StreamProcessor with tumbling/sliding/session windows and stream join
  • Built-in transformers – trimString, toLowerCase, toUpperCase, parseJson, stringifyJson, pick, omit, renameKeys
  • Data quality – QualityService with completeness, notNull, uniqueness, range, pattern, referentialIntegrity, profile(), detectAnomalies()
  • Connectors – DataSource/DataSink (MemorySource, MemorySink, CsvSource, HttpSource)
  • PII decorators – @Mask, @Redact, @Encrypt, @Decrypt for sensitive data
  • Test utilities – SchemaFaker, PipelineTestHarness, MockSource, MockSink
  • Flink integration – Optional Apache Flink deployment for distributed stream processing

Installation

npm install @hazeljs/data @hazeljs/core

Quick Start

1. Import DataModule

import { HazelApp } from '@hazeljs/core';
import { DataModule } from '@hazeljs/data';

const app = new HazelApp({
  imports: [DataModule.forRoot()],
});

app.listen(3000);

2. Define a pipeline with decorators

import { Injectable } from '@hazeljs/core';
import {
  Pipeline,
  PipelineBase,
  Transform,
  Validate,
  ETLService,
  Schema,
  Infer,
} from '@hazeljs/data';

const OrderSchema = Schema.object({
  id: Schema.string().min(1),
  customerId: Schema.string().min(1),
  items: Schema.array(
    Schema.object({
      sku: Schema.string().min(1),
      qty: Schema.number().min(1),
      price: Schema.number().min(0),
    })
  ),
  status: Schema.string().oneOf(['pending', 'paid', 'shipped', 'delivered', 'cancelled']),
  createdAt: Schema.string().min(1),
});

type Order = Infer<typeof OrderSchema>;

@Pipeline('order-processing')
@Injectable()
export class OrderProcessingPipeline extends PipelineBase {
  constructor(etlService: ETLService) {
    super(etlService);
  }

  @Transform({ step: 1, name: 'normalize' })
  async normalize(data: unknown): Promise<Order> {
    return { ...(data as Order), status: String((data as Order).status).toLowerCase() };
  }

  @Validate({ step: 2, name: 'validate', schema: OrderSchema })
  async validate(data: Order): Promise<Order> {
    return data;
  }

  @Transform({ step: 3, name: 'enrich' })
  async enrich(data: Order): Promise<Order & { total: number; tax: number }> {
    const items = data.items ?? [];
    const subtotal = items.reduce((sum, i) => sum + i.qty * i.price, 0);
    const tax = subtotal * 0.1;
    return { ...data, subtotal, tax, total: subtotal + tax };
  }
}

3. Execute from a controller or service

import { Controller, Post, Body, Inject } from '@hazeljs/core';
import { OrderProcessingPipeline } from './pipelines/order-processing.pipeline';

@Controller('data')
export class DataController {
  constructor(@Inject(OrderProcessingPipeline) private pipeline: OrderProcessingPipeline) {}

  @Post('pipeline/orders')
  async processOrder(@Body() body: unknown) {
    const result = await this.pipeline.execute(body);
    return { ok: true, data: result };
  }
}

Schema validation

Build schemas with the fluent API. Full type inference via Infer<T>:

import { Schema, Infer, SchemaValidator } from '@hazeljs/data';

const UserSchema = Schema.object({
  email: Schema.string().email(),
  name: Schema.string().min(1).max(200),
  age: Schema.number().min(0).max(150),
  role: Schema.string().oneOf(['user', 'admin', 'moderator', 'guest']),
  active: Schema.boolean().default(true),
});

type User = Infer<typeof UserSchema>;

// Validate (throws on failure)
const validator = new SchemaValidator();
const user = validator.validate(UserSchema, rawData);

// Safe validate (returns result)
const result = validator.safeValidate(UserSchema, rawData);
if (result.success) {
  const user = result.data;
} else {
  console.error(result.errors);
}

Schema types and modifiers

Type Example
Schema.string() .email(), .url(), .min(), .max(), .uuid(), .oneOf(), .pattern(), .required(), .trim()
Schema.number() .min(), .max(), .integer(), .positive(), .negative(), .multipleOf()
Schema.boolean() .default()
Schema.date() .min(), .max(), .default()
Schema.object({...}) .strict(), .pick(), .omit(), .extend()
Schema.array(itemSchema) .min(), .max(), .nonempty()
Schema.literal(value) Literal values
Schema.union([a, b]) Discriminated unions
Modifiers .optional(), .nullable(), .default(), .transform(), .refine(), .refineAsync()

Pipeline options

Steps support conditional execution, retry, timeout, and DLQ:

@Transform({
  step: 2,
  name: 'enrich',
  when: (data) => (data as { type: string }).type === 'order',
  retry: { attempts: 3, delay: 500, backoff: 'exponential' },
  timeoutMs: 5000,
  dlq: { handler: (item, err, step) => logger.error('DLQ', { item, err, step }) },
})
async enrich(data: unknown) {
  return { ...data, enriched: true };
}

PipelineBuilder (programmatic pipelines)

Build pipelines in code without decorators:

import { PipelineBuilder } from '@hazeljs/data';

const pipeline = new PipelineBuilder('orders')
  .addTransform('normalize', (d) => ({
    ...d,
    email: (d as { email: string }).email?.toLowerCase(),
  }))
  .branch(
    'classify',
    (d) => (d as { type: string }).type === 'premium',
    (b) => b.addTransform('enrichPremium', enrichPremium),
    (b) => b.addTransform('enrichStandard', enrichStandard)
  )
  .parallel('enrich', [(d) => ({ ...d, a: 1 }), (d) => ({ ...d, b: 2 })])
  .catch((data, err) => ({ ...data, error: err.message }));

const result = await pipeline.execute(rawData);

Batch and stream processing

import { StreamService, StreamProcessor } from '@hazeljs/data';

// Batch
const results = await streamService.processBatch(pipeline, items);

// Streaming with windowing
const processor = new StreamProcessor(etlService);
for await (const batch of processor.tumblingWindow(source, 60_000)) {
  console.log(batch.items, batch.windowStart, batch.windowEnd);
}
// Also: slidingWindow, sessionWindow, joinStreams

Data quality

import { QualityService } from '@hazeljs/data';

const qualityService = new QualityService();

qualityService.registerCheck('completeness', qualityService.completeness(['id', 'email']));
qualityService.registerCheck('notNull', qualityService.notNull(['id']));
qualityService.registerCheck('uniqueness', qualityService.uniqueness(['id']));
qualityService.registerCheck('range', qualityService.range('age', { min: 0, max: 120 }));
qualityService.registerCheck('pattern', qualityService.pattern('phone', /^\d{10}$/));
qualityService.registerCheck(
  'ref',
  qualityService.referentialIntegrity('status', ['active', 'inactive'])
);

const report = await qualityService.runChecks('users', records);
const profile = qualityService.profile('users', records);
const anomalies = qualityService.detectAnomalies(records, ['value'], 2);

PII decorators

import { Transform, Mask, Redact } from '@hazeljs/data';

@Transform({ step: 1, name: 'sanitize' })
@Mask({ fields: ['email', 'ssn'], showLast: 4 })
sanitize(data: User) {
  return data; // email/ssn already masked
}

@Transform({ step: 2, name: 'redact' })
@Redact({ fields: ['internalId'] })
redact(data: Record<string, unknown>) {
  return data; // internalId removed
}

Test utilities

import { SchemaFaker, PipelineTestHarness, MockSource, MockSink } from '@hazeljs/data';

const fake = SchemaFaker.generate(UserSchema);
const many = SchemaFaker.generateMany(UserSchema, 10);

const harness = PipelineTestHarness.create(etlService, pipeline);
const { result, events } = await harness.run(input);
await harness.runAndAssertSuccess(input);

const source = new MockSource([{ x: 1 }]);
const sink = new MockSink();

Built-in transformers

Transformer Description
trimString Trim whitespace from strings
toLowerCase / toUpperCase Case conversion
parseJson / stringifyJson JSON parsing and serialization
pick Select specific keys from objects
omit Remove specific keys from objects
renameKeys Rename object keys 
DataModule.forRoot({
  flink: {
    url: process.env.FLINK_REST_URL ?? 'http://localhost:8081',
    timeout: 30000,
  },
});

Example

See hazeljs-data-starter for a full example with order and user pipelines, PipelineBuilder, PII decorators, quality profiling, anomaly detection, and REST API.