Tiny ETL
Purpose of this package is to create a small ETL.
https://rpgeeganage.github.io/tiny-etl/doc/
TypeScript Doc:Basic Usage
import { TinyEtl } from 'tiny-etl';
const tinyEtl = TinyEtl.getInstance({
// Input stream, Input file path
input: ...,
// Output stream, Output file path
output: ...
});
// Add transformer
tinyEtl.addTransformer(...);
// Add transformer
tinyEtl.addTransformer(...);
// Call Process method to initiate the ETL
tinyEtl.process();
// Listen to the 'finish' event of the output stream to for the end of process
(tinyEtl.output as fs.WriteStream).on('finish', () => {
....
});
Conditional branching
IF-ELSE
Transformers
Sample code: Sample-3
Input parameter:
interface IfElseParameters {
// Transformer to run if the predicate function returns true
ifTrue: Transform;
// Transformer to run if the predicate function returns false
ifFalse: Transform;
// Predicate function. chunk and encoding will be passed the transformer function in IfElse
predicate: (chunk: any, encoding?: string) => boolean | Promise<boolean>;
// Set this to true, if the predicate function is a Async/Await or a Promise
isAsyncPredicate?: boolean;
}
Usage:
import { ControlFlow } from 'tiny-etl';
const sampleControlFlow = ControlFlow.IfElse.getControlFlow({
ifTrue: ...,
ifFalse: ...,
predicate: (chunk: any, encoding?: string) => {
return ...
}
} as ControlFlow.IfElseParameters );
...
tinyEtl.addTransformer(sampleControlFlow);
...
tinyEtl.process();
Writers
Sample code: Sample-4
Input parameter:
interface IfElseParameters {
// Writer to run if the predicate function returns true
ifTrue: Writable;
// Writer to run if the predicate function returns false
ifFalse: Writable;
// Predicate function. chunk and encoding will be passed the Writer function in IfElse
predicate: (chunk: any, encoding?: string) => boolean | Promise<boolean>;
// Set this to true, if the predicate function is a Async/Await or a Promise
isAsyncPredicate?: boolean;
}
Usage:
import { ControlFlow, TinyEtl } from 'tiny-etl';
const sampleControlFlowWriter = ControlFlow.Output.IfElse.getControlFlow({
ifTrue: ...,
ifFalse: ...,
predicate: (chunk: any, encoding?: string) => {
return ...
}
} as ControlFlow.Output.IfElseParameters );
const tinyEtl = TinyEtl.getInstance({
// Input stream, Input file path
input: ...,
output: sampleControlFlowWriter
});
Multi-Conditions
Transformers
Sample code: Sample-3
Input parameter: Single return value parameter
interface SingleCondition<R> {
// Return value of the condition
ReturnValue: R;
// Transformer for the return value
Transformer: Transform;
}
Multi-Condition parameter
interface MultiConditionParameters<R> {
// Transformer to run if the predicate function returns true
conditions: SingleCondition[];
// Predicate function. chunk and encoding will be passed to each transformer function in conditions
predicate: (chunk: any, encoding?: string) => R | Promise<R>;
// Set this to true, if the predicate function is a Async/Await or a Promise
isAsyncPredicate?: boolean;
}
Usage:
import { ControlFlow } from 'tiny-etl';
const sampleControlFlow = ControlFlow.MultiCondition.getControlFlow({
conditions: [...],
predicate: (chunk: any, encoding?: string): R => {
if (...) return R
if (...) return R
if (...) return R
}
} as ControlFlow.MultiConditionParameters );
...
tinyEtl.addTransformer(sampleControlFlow);
...
tinyEtl.process();
Writers
Sample code: Sample-5
Input parameter: Single return value parameter
interface SingleCondition<R> {
// Return value of the condition
ReturnValue: R;
// Writer for the return value
Writer: Writable;
}
Multi-Condition parameter
interface MultiConditionParameters<R> {
// Writer to run if the predicate function returns true
conditions: SingleCondition[];
// Predicate function. chunk and encoding will be passed the writer function in condtion
predicate: (chunk: any, encoding?: string) => R | Promise<R>;
// Set this to true, if the predicate function is a Async/Await or a Promise
isAsyncPredicate?: boolean;
}
Usage:
import { ControlFlow, TinyEtl } from 'tiny-etl';
const sampleControlFlowWriter = ControlFlow.Output.MultiCondition.getControlFlow({
conditions: [...],
predicate: (chunk: any, encoding?: string): R => {
if (...) return R
if (...) return R
if (...) return R
}
} as ControlFlow.Output.MultiConditionParameters );
const tinyEtl = TinyEtl.getInstance({
// Input stream, Input file path
input: ...,
output: sampleControlFlowWriter
});
Writing a custom transformer
import { Transform, TransformCallback} from 'stream';
export class SampleTransformer extends Transform {
constructor() {
super();
}
_transform(chunk: any, encoding: string, callback: TransformCallback) {
... // Do the transforming
callback(
// Error
...
,
// Transformed chunk
...)
);
}
}
// Add transformer
tinyEtl.addTransformer(SampleTransformer);
Writing a custom Writer
import { TransformCallback, Writable, WritableOptions } from 'stream';
export class SampleTransformer extends Writable {
constructor(options: WritableOptions) {
super(options);
}
_write(chunk: any, encoding: string, callback: TransformCallback) {
... // Do the transforming
callback(
// Error
...
,
// Transformed chunk
...)
);
}
_final(cb: (error?: Error | null) => void) {
cb(
// Error
...
);
}
}
// Add transformer
tinyEtl.addTransformer(SampleTransformer);
Examples
-
Encrypting and decryption.
npm run sample-1
-
Parse CSV to JSON without conditional branching
npm run sample-2
-
Parse CSV to JSON with conditional branching
npm run sample-3
-
Conditional Writer
npm run sample-4
-
Multi Conditional Writer
npm run sample-5
Run all the samples.
npm run all-samples