Forem: Ayush

📈Scaling Node.js: Core Principles for High-Throughput Data Pipelines

Ayush — Sun, 24 Aug 2025 15:34:38 +0000

At Biofourmis, we handle a wide range of patient data, from low-frequency episodic submissions to high-throughput, continuous streams of data arriving at sub-second frequencies.

For a particularly challenging use case, I was tasked with creating a pipeline to handle thousands of concurrent patients, each sending a 150 KB data blob every 10 seconds. This pipeline also needed to withstand bursts of up to 8 calls per second per patient. All of this had to be achieved with an eye on maintaining low costs, both in terms of infrastructure and developer hours.

Given our team's expertise is primarily in Node.js, I set out to create and test the limits of what a platform made with Node.js could reliably handle. I also decided against direct Kafka broker exposure for now, as I wanted a service that could be independently modified and easily extended to handle other ingestion use cases in the future.

TL;DR 🚀

The Requirement🤯: To build a high-throughput Node.js service to handle massive patient data streams (150 KB blobs) from thousands of concurrent devices up to 700 calls/sec, while keeping costs low.
The Solution✨: My approach combined strategic tool choices with a deep understanding of Node.js and the V8 engine:
- Fastify: Chosen for its superior speed and low overhead compared to Express, which is due to its schema-based approach that helps V8 optimize code from the start.
- Streams: Prevented event loop from being blocked by using json-stream-stringify to process large JSON payloads in chunks, maintaining high concurrency.
- node-rdkafka: Chose this battle-tested wrapper on top of native C/C++ librdkafka library for its exceptional throughput over other clients.
- V8Memory Optimizations: Prevented V8's hidden class de-optimizations by ensuring objects adhered to a fixed schema and handled responsibly. Also used flatstr library to make strings more memory-efficient before being converted to a Buffer.
Result and Analysis✅: The service proved to be highly efficient and cost-effective:
- Peak Load: Handled up to 700 calls/sec during stress tests.
- Normal Resource Usage: Stayed low at 90-125 MB of memory and only 50-150 millicores of CPU.

This approach demonstrated that Node.js, when architected with performance in mind, can be a powerful and viable option for building high-throughput data pipelines.

The Requirement

The service's core job was to perform a simple, high-frequency task:

API call -> Validate data -> Serialize Data -> Produce to Kafka

The Solution: A Deep Dive into Key Components

Here's how I built a Node.js service that not only meets these demanding requirements but also remains incredibly efficient.

1. Fastify: The High-Speed Gateway⚡

Fastify is a web framework known for its speed and low overhead. Fastify's core design philosophy leverages V8 optimizations that are often overlooked, allowing it to achieve up to 4x the throughput of Express.

Fastify's schema-based approach is one of the key factors behind its performance, as I'll explain further when we discuss memory optimization.

2. Streams: Event Loop's Best Friend🌊

A common pitfall with Node.js is blocking the event loop. Operations like JSON.stringify() on a large object can be computationally expensive and may lead to dropped requests in a high-concurrency environment.

To avoid this, I used streams with the json-stream-stringify library. Instead of processing the entire JSON object at once, this library serializes the data in chunks, ensuring the event loop remains free to handle incoming requests.

let stringjson = "";
const jsonStream = new JsonStreamStringify(very-large-payload);
jsonStream.once("error", (err) => {
    reject(CustomError);
    // delete message ID from message tracker
});
jsonStream.on("data", (chunk) => {
    stringjson += chunk;
});
jsonStream.on("end", () => {
    try {
        flatstr(stringjson);
        // convert to Buffer and produce to Kafka
        resolve();
    } catch (err) {
        rej(err);
    }
});

3. node-rdkafka: The Battle-Tested Connector💪

For the Kafka integration, a client was needed that could handle the immense throughput. While KafkaJS is a popular choice, my tests showed that node-rdkafka was superior for this particular use case.

node-rdkafka is a high-performance Node.js client that wraps librdkafka, Apache Kafka's native C/C++ library. This native binding provides exceptional speed and efficiency.

A drawback is that it can occasionally be challenging to install and develop with locally, but for a throughput-intensive application, the performance gains are well worth it.

4. Memory Optimization: The Flatstr and Pass-by-Reference Approach🧠

To keep the service lean and cost-effective, I focused on memory optimization next.

V8's Hidden Classes: A key insight was understanding V8's hidden classes. Every time you alter an object by adding a new property, V8 creates a new hidden class, which can severely de-optimize your code. Ensuring that objects adhered to a pre-defined schema, were not unnecessarily copied, and were passed by reference to maintain a predictable, optimized object structure helped immensely with performance & resource optimization.

Fastify's schema-based approach handles this well and is a key factor to its performance. By defining the data structures upfront with JSON Schema, it allows the V8 engine to create optimized hidden classes for objects, avoiding costly runtime de-optimizations.

Flatstr: I also used the flatstr library. This module helps with a specific V8 optimization issue: when a new string is created by concatenating others, it leads to bloated memory-inefficient objects, internally represented as a tree. flatstr prunes these internal structures, making the string into a compact array before it's processed by an external streaming source or converted to a Buffer, thereby saving precious memory and compute resources.

Results and Analysis 📊

The results of this approach were highly encouraging. The service's resource usage remains incredibly low, making it an exceptionally cheap service to run.

Normal Load: 10 calls/sec
Peak Load Observed: 230 calls/sec
Stress Test: Up to 700 calls/sec
Memory Usage: 90 MB (normal) to 125 MB (peak)
CPU Usage: 50 millicore (normal) to 150 millicore (peak)

Conclusion ✨

In this project, I set out to prove that Node.js, when architected with a deep understanding of its event loop and the V8 engine, is a powerful and viable platform for high-throughput, data-intensive workloads. By strategically choosing tools like Fastify and node-rdkafka and applying key optimizations with streams and flatstr, I was able to build an incredibly efficient and cost-effective service.

Ultimately, the metrics speak for themselves: the service handles hundreds of requests per second with minimal resource consumption. Node.js is a robust and highly capable choice for some complex, real-world data pipelines.

🧠 GenAI as a Backend Engineer: Part 3 — RAG with LlamaIndex

Ayush — Sun, 10 Aug 2025 16:00:47 +0000

Next Up: Retrieval-Augmented Generation (RAG) made simple 🚀

In Part 2, we built a semantic search engine with embeddings + Qdrant. Now, we’ll hook that up to an LLM so your app can answer questions instead of just returning matching docs.

By the end, you’ll:

📖 Understand RAG and why it matters.
⚙️ Learn how LlamaIndex helps you build RAG pipelines quickly.
🏗 Build: Raw text → Vector DB → LLM → Answer.

What is RAG? 🤔

RAG = Retrieval-Augmented Generation.

Retrieve 🔍: Fetch relevant chunks from your own data using semantic search.
Augment ➕: Add that data to the prompt for the LLM.
Generate ✨: Get a grounded, accurate answer.

Why? LLMs are frozen in time. They are trained on a generic data, not your data. RAG lets them use your current, domain-specific data and generate responses.

LlamaIndex vs LangChain ⚖️

Feature	LangChain	LlamaIndex
Focus	Agents, tools, flexibility	RAG & retrieval pipelines
Setup	More verbose, more options	Leaner, less boilerplate
Best for	Complex, multi-tool workflows	Quick, retrieval-focused RAG

Here, we go with LlamaIndex — less config, easier to build. They are however, not mutually exclusive and can be mixed & matched.

📖 More on concepts: LlamaIndex Overview

Building It 🛠

Install dependencies 📦

python -m venv venv
source venv/bin/activate
pip install llama-index sentence-transformers qdrant-client langchain_community llama-index-vector-stores-qdrant llama-index-embeddings-langchain llama-index-llms-gemini

Scenario A: Naïve RAG (full doc as context)

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Settings, StorageContext
from llama_index.embeddings.langchain import LangchainEmbedding
from llama_index.vector_stores.qdrant import QdrantVectorStore
from qdrant_client import QdrantClient
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from llama_index.llms.gemini import Gemini, Gemini
import os

# Keep this as none to see the context creation in response
api_key = os.environ.get('GEMINI_API_KEY', None)

# Loads full docs, embeds, and queries without chunking

# Step 1: Load your files (txt for now)
documents = SimpleDirectoryReader("./tmp").load_data()

# Step 2: Use HuggingFace embeddings (SBERT)
# Interface conversion so llama_index can understand embeddings
embed_model = LangchainEmbedding(HuggingFaceBgeEmbeddings(model_name="all-MiniLM-L6-v2"))

# Step 3: Connect to Qdrant + create storage context for llamaindex
qdrant = QdrantClient(host="localhost", port=6333)
vector_store = QdrantVectorStore(client=qdrant, collection_name="rag_docs")

storage_context = StorageContext.from_defaults(vector_store=vector_store)

# Step 4: Create index to query on
index = VectorStoreIndex.from_documents(
  documents,
  storage_context=storage_context,
  embed_model=embed_model
)

Settings.llm = Gemini(api_key=api_key) if api_key else None
# Internally calls retreiver as shown in rag_tokens
query_engine = index.as_query_engine()

# Step 4: Create context based on index response, send to LLM
# Works, but not efficient — passes entire text to the LLM
response = query_engine.query("What is milk?")
print(response)

response = query_engine.query("What is the color of this item?")
print(response)

response = query_engine.query("Is it edible?")
print(response)

Scenario B: Smarter RAG (chunk + retrieve)

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Settings, StorageContext
from llama_index.embeddings.langchain import LangchainEmbedding
from llama_index.core.node_parser import SimpleNodeParser
from llama_index.vector_stores.qdrant import QdrantVectorStore
from qdrant_client import QdrantClient
from langchain_community.embeddings import HuggingFaceBgeEmbeddings 
from llama_index.llms.gemini import Gemini, Gemini
from textwrap import dedent
import google.generativeai as genai
import google.generativeai.generative_models as genaimodels
import sys
import os

api_key = os.environ.get('GEMINI_API_KEY', None)

documents = SimpleDirectoryReader("./tmp").load_data()

embed_model = LangchainEmbedding(HuggingFaceBgeEmbeddings(model_name="all-MiniLM-L6-v2"))

qdrant = QdrantClient(host="localhost", port=6333)
vector_store = QdrantVectorStore(client=qdrant, collection_name="rag_docs_chunk")

storage_context = StorageContext.from_defaults(vector_store=vector_store)

# Splits docs into 200-token chunks with 50-token overlap
parser = SimpleNodeParser.from_defaults(chunk_size=200, chunk_overlap=50)
nodes = parser.get_nodes_from_documents(documents)

index = VectorStoreIndex(nodes, storage_context=storage_context, embed_model=embed_model)
retreiver = index.as_retriever(similarity_top_k=3)

question = "Health benefits of turmeric?"
retreived_nodes = retreiver.retrieve(question)

print ("=======Nodes retreived=========")
for i, node in enumerate(retreived_nodes):
  print(f"--- Chunk {i} ---")
  print(f"(Score: {node.score})")
  print(node)

if (api_key is None):
  print ("Cannot run gemini workflows. API Key not set")
  sys.exit()

# Both retrieve top-k relevant chunks, then passes only those to the LLM

# =======Self pompt building section=========
def build_prompt(question, nodes):
  context_text = "\n\n".join(
    [f"Chunk{i+1}:\n{chunk.get_content()}" for i, chunk in enumerate(nodes)]
  )
  return dedent(f"""
  You are an expert assistant. Use only the provided context to answer the question.
  If the answer is not found in the context, say "I don't know."

  Context:
  {context_text}

  Question:
  {question}

  Answer:
  """)

genai.configure(api_key=api_key)
prompt = build_prompt(question, retreived_nodes)

model = genaimodels.GenerativeModel()
response = model.generate_content(prompt)
print("\n--- Final Answer ---")
print(response)
# =======Self pompt building section=========

# =======Auto pompt section=========
Settings.llm = Gemini(api_key=api_key) # None
query_engine = index.as_query_engine()
response = query_engine.query("Health benefits of turmeric?")
print(response)
# =======Auto pompt section=========

Step 4 — How the Pipeline Works 🔄

Chunking 🪓 — Break docs into small segments.
Embedding 🧩 — Convert chunks to semantic vectors.
Vector Store 📚 — Save vectors + metadata in Qdrant.
Retrieval 🔍 — Search for top-k similar chunks.
Prompting 🗨 — Inject chunks into the LLM query.

TL;DR Takeaways 📌

RAG = more accurate, up-to-date answers.
Chunking keeps context relevant and token count low.
LlamaIndex is great for quick RAG setups.

If using the files in repo, you should see chunks instead of the whole file:

=======Nodes retreived=========
--- Chunk 0 ---
(Score: 0.72586405)
Node ID: 473e9e27-8052-4347-8108-6dc496f46912
Text: It's also used as a natural food coloring and in various
beverages and teas.   Fresh turmeric root can be grated and added to
dishes, while dried and powdered turmeric is widely available.
Traditional and Medicinal Uses:  Turmeric has a rich history in
traditional medicine systems like Ayurveda and traditional Chinese
medicine.   It has been u...

💻 GitHub Repo

🔗 Other Parts:

🧠 GenAI as a Backend Engineer: Part 2 - Vector DBs

Ayush — Sat, 09 Aug 2025 22:15:04 +0000

🚀 Next Up: Vector Databases (Hands-On!)

We’re about to build a tiny semantic search engine from scratch.

The secret ingredient? Vector databases — a tool that’s surprisingly easy to grasp, but powerful enough to make modern AI work.

By the end of this article, you’ll:

Understand what embeddings are (and why they’re magic)
Learn what a vector DB does
Spin up Qdrant in Docker
Store and search your own embeddings
See how this forms the foundation for RAG (Retrieval-Augmented Generation)

Step 1 — From Words to Numbers: Embeddings

Computers don’t understand “dog” or “car” like we do.

Instead, we turn these into embeddings — long lists of numbers that capture meaning and relationships between concepts.

Example:

"dog"   → [0.1, 0.6, -0.4, ...]  
"puppy" → vector very close to "dog"  
"car"   → far away from both

Different embedding models (like OpenAI or SentenceTransformers) can convert:

Text → a 768- or 1536-dimensional vector
Images → a visual vector

📖 Want to go deeper?

Step 2 — Why a Vector Database?

Once you have embeddings, you need a way to:

Store them (there can be millions)
Quickly find which ones are most similar to a new query

That’s exactly what a Vector Database does.

Instead of “WHERE id = 123” lookups, it supports similarity search — finding the closest k vectors to your query in high-dimensional space.

Key terms you’ll run into:

ANN Search — Approximate Nearest Neighbor, fast search in huge vector spaces
Index — special data structure to speed up similarity lookups
Top-K Search — get the k most similar vectors (e.g., top 5 related docs)

Some popular choices:

Tool	Notes
FAISS	Local, in-memory, blazing fast
Qdrant	Rust-based, open-source, persistent — we’ll use this
Pinecone / Weaviate	Fully managed cloud services

📖 More on concepts: Qdrant Overview | Qdrant Concepts

🎥 Optional: FAISS Top-K Search explained

Step 3 — Setting Up Qdrant

We’ll run Qdrant locally using Docker:

docker run -p 6333:6333 qdrant/qdrant

(If Docker isn’t an option, try Qdrant Cloud Free Tier).

You can also explore the Qdrant dashboard locally by visiting http://localhost:6333/dashboard in your browser.

Step 4 — Creating Our Project

Let’s set up a Python environment for our experiment:

mkdir vector && cd vector
python -m venv venv
.env\Scriptsctivate

pip install openai qdrant-client numpy sentence-transformers

Step 5 — Storing Embeddings in Qdrant

We’ll embed a few sample sentences and store them in Qdrant:

# embed_and_store.py
from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance
import uuid

# Sample documents
docs = [
    "Apple is a fruit",
    "Dogs are loyal animals",
    "SpaceX launches rockets",
    "Oranges are citrus",
    "Cats are independent"
]

# Connect to Qdrant
client = QdrantClient(host="localhost", port=6333)

# Create (or reset) a collection
client.recreate_collection(
    collection_name="embeddings",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)

# Load embedding model
embed_model = SentenceTransformer("all-MiniLM-L6-v2")
def get_embedding(txt): return embed_model.encode(txt)

# Store documents
points = [
    PointStruct(id=str(uuid.uuid4()), vector=get_embedding(txt), payload={"text": txt})
    for txt in docs
]

client.upsert(collection_name="embeddings", points=points)
print("Stored docs.")

Step 6 — Running a Similarity Search

Now, let’s query for something related to rockets:

# embed_and_store.py
query = get_embedding("Tell me about rockets")
hits = client.search(
    collection_name="embeddings",
    query_vector=query,
    limit=2
)
for hit in hits:
    print(hit.payload["text"], "Score:", hit.score)

Example output:

# python embed_and_store.py
SpaceX launches rockets Score: 0.92
Dogs are loyal animals Score: 0.12

Step 7 — From Search to AI: RAG

At this point, you’ve built the backbone of a semantic search system.

But right now, you’re only returning similar documents — not answers.

That’s where RAG (Retrieval-Augmented Generation) comes in:

Fetch the top results from your vector DB
Feed them into a large language model (LLM)
Let the model generate an answer using that context

In the next part, we’ll hook Qdrant up to an LLM so it can respond to natural language questions with real context.

💡 Key Takeaways:

Embeddings turn meaning into numbers
Vector DBs store and search embeddings efficiently
Qdrant + SentenceTransformers is an easy, powerful local setup
RAG is the magic step that turns similarity search into a conversational AI

💻 GitHub Repo

🔗 Other Parts:

🧠 GenAI as a Backend Engineer: Part 1 - Model Serving

Ayush — Wed, 06 Aug 2025 20:38:32 +0000

This series is meant to serve as a guide towards better understanding and getting started with concepts surrounding AI. I am writing this as a Backend Engineer trying to level-up.

I am using google (yep, old school), and AI tools to form a roadmap, ask questions and understand the concepts. The goal is to document the steps along the way.

🤔 Questions that will get answered by the end, hopefully:

What are LLMs, Agents, DAGS, RAGs, Vector DBs etc
How do these things really work? Not the Math or the Neural nets but rather, all the flows using LLMs etc
More importantly — how can I, as a backend engineer, get started and contribute to building AI tools?

If you're also exploring, or already working in the field — do share your feedback, mistakes, or suggestions in the comments.

🗺️ Current plan:

Model Serving --> Airflow --> Vector DBs --> RAG Style Q&A + Llama Index

🚀 Part 1: Model Serving with FastAPI & TorchVision

In this step, I learned how a model is served behind an API — that's it. The same models from the earlier "ML" days (e.g., classification models) — now exposed cleanly via API.

Serving is about making the model available for real-time or batch predictions, efficiently, securely, and at scale.

🔹 What is a Model?

A model is the core of an ML system — a program trained on data to recognize patterns and make predictions.

🔹 What is Model Serving?

Model Serving is the process of putting that trained model behind an API (e.g., FastAPI), so it can take input and return predictions (inference).

Instead of bundling the model inside every client app, you host it once, centrally.

🔍 Real-World Examples:

🖼️ Image → API → Model returns: "cat" or "dog"
💬 Chatbot message → API → LLM replies
📄 Transaction → Fraud model → "fraud" or "legit"

📚 Key Terms I Came Across

Inference → Running the model on new (unseen) input data
Model Hosting → Putting the model on a server (local or cloud) and exposing an API

📝 Read: Model Serving 101 (Paywalled...)

🔥 Key Takeaways from the Article:

Model serving introduces a distinct set of challenges compared to a typical CRUD backend. It’s as if the heavy-lifting data pipelines we used to run in the background now need to respond to client requests in real-time, with strict performance and scalability requirements.
Key factors to balance:
- 🚀 Throughput – predictions/sec
- ⏱️ Latency – response time
- 💰 Cost – infra & compute
3 Fundamental Deployment Types:

Online Real-Time Inference
Asynchronous Inference
Offline Batch Transform

🧰 Tools Used

⚡ FastAPI

A high-performance Python web framework.

📖 Official Tutorial: FastAPI Docs

🖼️ TorchVision

A interesting PyTorch library that provides:

Pretrained computer vision models (like resnet18, mobilenet) used for image or object classification
Tools for image transformations and loading

💡 Why it’s great: You don’t need to train from scratch. You can just load and serve a powerful image model in minutes.

📖 Read: TorchVision Basics

⚙️ Step-by-Step: Serving a Model via API

🔹 Step 1: Setup Environment

mkdir model-serving && cd model-serving
python3 -m venv venv
venv\Scripts\activate  # On Mac/Linux: source venv/bin/activate
pip install fastapi uvicorn torch torchvision pillow requests

🔹 Step 2: Create Model Loader

📄 model.py

import torch
from torchvision import models, transforms
from PIL import Image
import requests

model = models.resnet18(weights=models.ResNet18_Weights.DEFAULT)
model.eval()

transform = transforms.Compose([
  transforms.Resize((224, 224)),
  transforms.ToTensor(),
  transforms.Normalize(
    mean=[0.485, 0.456, 0.406],
    std=[0.229, 0.224, 0.225],
  )
])

LABELS_URL = "https://raw.githubusercontent.com/pytorch/hub/master/imagenet_classes.txt"
response = requests.get(LABELS_URL)
labels = [line.strip() for line in response.text.splitlines()]

def predict(image_path):
    image = Image.open(image_path).convert("RGB")
    input_tensor = transform(image).unsqueeze(0)

    with torch.no_grad():
        output = model(input_tensor)
    pred_index = output.argmax().item()
    return labels[pred_index]

def predict_topk(img_path):
    image = Image.open(img_path).convert("RGB")
    input_tensor = transform(image).unsqueeze(0)

    with torch.no_grad():
        output = model(input_tensor)
    probs = torch.nn.functional.softmax(output[0], dim=0)
    top_p, top_i = torch.topk(probs, 5)
    top_labels = [(labels[idx], round(prob.item(), 4)) for idx, prob in zip(top_i, top_p)]
    return top_labels

🔎 Interesting learning: models.ResNet18_Weights.DEFAULT loads a model pre-trained on 1000 categories. These labels come from a public file maintained by PyTorch (based on the ImageNet dataset). The model outputs a probability distribution over these categories, and the index with the highest score maps to the predicted label.

🔍 I have included both predict and predict_topk methods to demonstrate how you can work with the model's output. While predict gives you just the top result, predict_topk provides the top 5 predictions along with confidence scores. This is useful when you want more insight into what the model "thinks" the image could be, especially in ambiguous cases.

🔹 Step 3: Create FastAPI App

📄 app.py

from fastapi import FastAPI, UploadFile, File
from model import predict
import shutil

app = FastAPI()

@app.post("/predict")
async def classify_image(file: UploadFile = File(...)):
    temp_path = f"/tmp/{file.filename}"
    with open(temp_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    result = predict(temp_path)  # or predict_topk(temp_path)
    return {"prediction": result}

🔹 Step 4: Run the API

uvicorn app:app --reload

Go to: http://127.0.0.1:8000/docs

📤 Upload any image → get a prediction

✅ That’s it! You’ve served your first model. Now you can integrate this into real-world applications or scale it using cloud services.

💻 GitHub Repo

🔗 Other Parts:

🪜 Coming Up Next

Next, I plan to explore ~~Apache Airflow~~ and how it's used for ML workflows and pipelines — one layer deeper each time 💡

Due to technical issues in Windows, this topic had to be pushed for later. Take a look at RAG and VectorDB instead.

🔥 1BRC in Node.js: From 12 Minutes to 35 Seconds

Ayush — Tue, 29 Jul 2025 15:54:14 +0000

I took on the 1 Billion Row Challenge (1BRC) — but with Node.js. The "event loop" language that is not the first choice when crunching raw numbers.

🧠 TL;DR

Stage	Time
🐢 Baseline	12:06
🔥 Final	0:35
📈 Speedup	1957%

I squeezed out sizeable performance gains out of Node with buffer math, manual parsing, worker threads — even byte-level micro-ops.

🎯 The Challenge

You're given a file with 1 billion lines, each like:

StationName;Temperature\n

You need to compute per station:

Minimum temperature
Maximum temperature
Average temperature

And you need to do it fast.

⚙️ System Configuration & Setup

Here’s my system details that was used for benchmarking and optimizations:

Machine: Windows 11 (x64)
CPU: 2.5 GHz, 10 cores, 12 logical threads, (Intel i5, 12th Gen)
RAM: 8 GB
Disk: 256GB NVMe SSD
Node.js version: v22.x (LTS)

⛽ Just Reading the File (No Work)

To measure the disk I/O floor, I timed a pure file read — no parsing or processing.

⏱️ 13 seconds

This was my physical lower limit; my goal. Here's the minimal benchmark:

const fs = require("fs");

const filePath = process.argv[2];
const bufferSize = 64 * 1024; // 64 KB
const buffer = Buffer.alloc(bufferSize);

const fd = fs.openSync(filePath, "r");

let totalBytesRead = 0;
let bytesRead = 0;

console.time("diskRead");

do {
  bytesRead = fs.readSync(fd, buffer, 0, bufferSize, null);
  totalBytesRead += bytesRead;
} while (bytesRead > 0);

console.timeEnd("diskRead");

fs.closeSync(fd);

console.log(`Read ${totalBytesRead} bytes`);
console.log(`Read speed: ${(totalBytesRead / (1024 * 1024)).toFixed(2)} MB`);

🧪 Starting Baseline

The first working version provided in repo:

Single thread
Used readline with for...await
Parsed strings with split(';'), used parseFloat & .toFixed()
Used Map for aggregation

⏱️ 12:06

Let’s optimize.

🧵 Parallelism — Low hanging fruit

1 billion CPU bound calculations on a multi-cores machine? Worker Threads were the first upgrade.

The trick here was to divide the file equally while keeping the lines intact. This was done with the below method.

const calculateOffset = async (start, end, fileHandle) => {
  const { buffer } = await fileHandle.read({
    buffer: Buffer.alloc(MAX_LINE_LENGTH),
    length: MAX_LINE_LENGTH,
    position: end,
  });
  const diff = buffer.indexOf(10);
  return { start, end, diff };
};

After, each worker got their slice to process. 🚀

Workers	Time
With 4 Workers	2:35
With 12 Workers (system core count)	1:41
With 14, 18 Workers	Perf degradation

🧠 Learning: Since the data is locally sourced, the context switching overhead between threads started having a negative impact. And worker threads make a HUGE difference.

⚡Loop optimizations

Once threading was established, I started chipping away at the main loop:

Switched to .on('data') from readline iterator - ⏱️1:19

for await (const line of lineStream) {}
// to
lineStream.on("line", (line) => {}

Switched to manual byte stream parsing readStream.on('data') - ⏱️1:02

Here, I started working with the byte chunk processing using processChunk method. Still using higher-order functions like toString and parseFloat

Improvements that gave minor gains - ⏱️1:00
- Replaced repetitive chunk[i] reference with const c = chunk[i]
- Removed .toFixed()

At this point, I started searching for what processing takes the most time & resources. I searched the web and tried to make sense of flame graphs (which I got working at this point, finally)

Consistent answer - string processing & float arithmetic

Replaced parseFloat with integer math - ⏱️0:35✅

Replaced with byte-based float parsing (scaled to tenths):

const parseBufferToDigit = (byte) => byte - 0x30;

const parseNumber = (length) => {
  if (number[0] === 0x2d) { // negative
    return length === 5
      ? -(parseBufferToDigit(number[1]) * 100 +
          parseBufferToDigit(number[2]) * 10 +
          parseBufferToDigit(number[4]))
      : -(parseBufferToDigit(number[1]) * 10 +
          parseBufferToDigit(number[3]));
  } else {
    return length === 3
      ? (parseBufferToDigit(number[0]) * 10 +
         parseBufferToDigit(number[2]))
      : (parseBufferToDigit(number[0]) * 100 +
         parseBufferToDigit(number[1]) * 10 +
         parseBufferToDigit(number[3]));
  }
};

❌ What Didn't Work

I tried some other things that failed (but worth noting):

⛔ Hashed station names: Collisions or performance. A fast function that concatenated/hashed ascii codes was much faster than .toString() (0:19s), but had a large amount of collisions. A larger hash absolutely hammered the performance. So, no straight forward solution found.
⛔ Float32Array, Int32Array for aggregation instead of objects: interestingly degraded performance
⛔ Using >12 threads (my core count): overhead cancelled benefit
⛔ Replaced Math.min/max with ternary comparison: lost ~2s

💻 Windows + WSL Gotchas

(not so) Fun roadblocks along the way:

Generator script failed with Java 24, specifically needed Java 21
Maven build failed: used the flag -Dlicense.skip to bypass license plugin
Switched to WSL for scripts and time: the path performance was horrible — sync file read in CMD: 22s vs WSL: 4 minutes 😵

12 threads benchmarking was +1:00 minute till I figured this out. Switched to Powershell to use Measure-Command {}

clinic profiler DID NOT work in PowerShell: Switched to CMD. This one took a looooong time. Ended up relying on a single console.time which worked consistently.

📢 Acknowledgments

🔍 This project draws inspiration from Edgar-P-Yan’s excellent 1BRC repo.
Some parsing techniques were adapted from his implementation.
This was an independent learning project, and I did not submit this to the official 1brc leaderboard.

🔗 GitHub

📦 Code & scripts:

👉 GitHub Repo
🏎️ 1BRC Node Repo

🧵 Final Thoughts

This wasn’t just about speeding up Node.js (ok, maybe it was) — but it was also about discovering what makes it tick.

CPU-bound work? Use worker_threads
Strings are expensive — avoid until the last moment
Floats are expensive — avoid in hot path, convert later
Every byte and cycle matters (literally)

I walked away with crashes, a laptop trying to take off 🚁 and a deep satisfaction.

Yet, I feel there is still more to be done here. I will keep trying to push these numbers so feel free to share insights, ideas or observations.

Thanks for reading!

A beginners guide to data streaming with Kafka, NodeJs and Socket.io

Ayush — Wed, 17 Feb 2021 18:34:42 +0000

Through this small guide, my intention is to enable anyone to understand and begin their journey of working with Kafka in a few minutes.
After following this guide and using the tools mentioned, which is not the only way to achieve the output, you should have a working example on your hands and the understanding to use more advanced concepts.

We shall start by building the individual modules, step-by-step and integrate all of them in the end. So, let's get started!

1. Setting up Kafka

This is where we set up Kafka in our system. So, head down to the official Apache Kafka quick-start guide. Follow the steps till Step 3 and create a topic with any desired name. In my examples I have used the name topic_stream.

2. Creating a Kafka Consumer

Glad you are back! Let's now use the Kafka-node consumer to see how a consumer functions. A consumer simply reads messages from the Kafka topic.
Below, we have created a Consumer that subscribes to topic_stream and prints the data that is produced.
I have setup the Kafka client on port 9092 in step 1. You should change the configuration according to your settings.

const kafka = require('kafka-node');

const Consumer = kafka.Consumer,
  client = new kafka.KafkaClient('localhost:9092'),
  consumer = new Consumer(
  client, [ { topic: 'topic_stream', partition: 0 } ], { autoCommit: false });

  consumer.on('message', function (message) {
    console.log(message);
  });

Pretty straightforward, right?

3. Creating a Socket API

Let's now setup a WebSocket API in our NodeJs service so we start streaming data to a webpage. We use Socket.io in conjunction with express. In a file called server_consumer.js create the following:

const express = require('express');
const port = 3000;
const app = express();

const server = app.listen(port, () => {
  console.log(`Listening on port ${server.address().port}`);
});
const io = require('socket.io')(server, {
  cors: {
    origin: '*',
  }
});

io.on('connection', client => {
  console.log('Connected', client);

  client.on('event', data => { 
    console.log('Event triggered by client')
   });

  client.on('disconnect', () => { 
    console.log('Client disconnected');
   });
});

4. Creating the Consumer, Socket API

We now know what a consumer is and how we intend to use the Socket.io API. Let's bring it all together now. Edit the server_consumer.js such that it now contains the Consumer as well:

const kafka = require('kafka-node');
const express = require('express');
const port = 3000;
const app = express();

const Consumer = kafka.Consumer,
 client = new kafka.KafkaClient('localhost:9092'),
 consumer = new Consumer(
 client, [ { topic: 'topic_stream', partition: 0 } ], { autoCommit: false });

const server = app.listen(port, () => {
  console.log(`Listening on port ${server.address().port}`);
});
const io = require('socket.io')(server, {
  cors: {
    origin: '*',
  }
});

io.on('connection', client => {
  console.log('Connected', client);

consumer.on('message', function (message) {
    client.emit('request', message.value);
  });

client.on('disconnect', () => { 
    console.log('Client disconnected');
 });
});

5. Creating the frontend using Chart JS

On our web browser, we will create a webpage that invokes the socket API and displays the data using Chart JS. You may use D3.js or any other libraries that you are comfortable with to display the data.
Let's create a simple index.html:

<html>
<head>
 <script src="https://code.jquery.com/jquery-3.5.1.slim.min.js"></script>
 <script src="https://cdn.jsdelivr.net/npm/chart.js@2.8.0"></script>
 <script src="https://cdn.socket.io/socket.io-3.0.1.min.js"></script>
 <script type="text/javascript" src="index.js"></script>
</head>
<body>
  <canvas id="myChart" width='300' height='100'></canvas>
</body>
</html>

In our index.js, we will make a call to our socket API and initiate listeners for data:

$(document).ready(function(){
  const count = 10;
  const data = {
    labels : [...Array(count).keys()].map(i => i.toString()),
    datasets : [
      {
        label: '# - Streamed number',
        backgroundColor: "rgba(50,220,220,0.5)",
        data : Array(count).fill(0),
      }
    ]
  }
  const updateData = function(newVal){
    const labels = data["labels"];
    const dataSetInitial = data["datasets"][0]["data"];
    count++;
    labels.push(count.toString());
    labels.shift();
    const newData = Math.floor(newVal);
    dataSetInitial.push(newData);
    dataSetInitial.shift();
  };
  const ctx = document.getElementById("myChart").getContext("2d");
  const chart = new Chart(ctx, {
    type: 'line',
    data,
    options: {animation : false}
  });

  function webSocketInvoke() {
    var socket = io('http://localhost:3000');
    socket.on('event', (value) => {
      updateData(value);
      chart.update();
    });
  }
  webSocketInvoke();
 });

We will now open the index.html and wait for the data to stream through.

6: Producing data

Let's get to the final step where we produce data that can then be streamed to our setup. I will mention with 2 options to do this:

Follow the instruction given in Step 4 of the Kafka quick-start guide. That's it.
If you would like to explore the kafka-node library further, you may experiment and create a producer.js file and run it using your node environment; example file given here.

That's it! You should now have a fully functional Apache Kafka stream at your hands and the understanding on how the end-to-end flow works.

Github link for the project: https://github.com/yuvraj2112/kafka_socket_stream

Compress and upload a file using Pako in Vue.js and Node.js

Ayush — Mon, 23 Mar 2020 11:40:42 +0000

Compressing a file before uploading seemed imperative.

The candidates were starting to flow through our system at Brilliant Hire. That ecstatic sense of accomplishment though, brought along a major concern. The upload time was insanely slow and the file size, especially the audio recordings, large! Attempts were ongoing to reduce the size of the live wav recordings but even then, all the file uploads by the candidate had to be fast! Compressing a file before uploading seemed imperative.

That's when Pako came to the rescue. We were able to reduce the files to half their size just by using the lowest level of compression and it did not even block the execution thread for the file size we were targeting. A win!

Just the same way, here's how you can upload a compressed file in Vue.js and uncompress the file in Node.js before you perhaps, further pipe it to S3.

Compressing a file in Vue.js

Here, we will allow a user to upload a file using the input tag. After a file upload is triggered, we shall call the onChange method which will compress the file before forwarding it to the upload method. This method will make the upload happen to our file handling API.

<input type="file" :accept="allowedMimes" ref="inputFile" @change="onChange"/>

import pako from 'pako';

function onChange() {
  const data = new FormData();
  const file = this.$refs.inputFile.files[0];

  const reader = new FileReader();
  reader.onload = (e) => {
    const fileAsArray = new Uint8Array((e.target! as any).result as ArrayBuffer);
    const compressedFileArray = pako.deflate(fileAsArray);
    const compressedFile = compressedFileArray.buffer;
    const dataToUpload = new Blob([compressedFile], {type: file.type});
    const fileToUpload = new Blob([dataToUpload], {type: file.type});
    data.append('file', fileToUpload, file.name);
    this.upload(data);
  };
  reader.readAsArrayBuffer(file);
}

What's happening here: The FileReader converts the file into an ArrayBuffer that is converted into a compressed or deflated file. The file is then sent to the upload method where it will be sent to our API.

Uncompressing and piping the file in Node.js

We will be using Busboy to handle the file in our back-end which runs on Node.js.

To keep this piece on point I will point you to my other write-up about How to Upload a file in Node.js. Here, you will find step-by-step instructions on how to handle a file using Busboy. I will be referring to the hooks mentioned in this post.

If you are back here, I will assume you have read or you already know how to use Busboy. Therefore, we shall get started with the task of uncompressing our file using Pako.

Step 1: We initiate our Pako instance and streams. We also initiate the ReadStream. Pako will push our uncompressed file to this stream to be piped further. This will all happen inside the Busboys onFile event.


const pako = require('pako');

busboy.on('file', async (fieldname, file, filename, encoding, mimetype) => {
  const fileReadStream = new Readable({
    read(size) {
      if (!size) this.push(null);
      else this.push();
    },
  });
  const inflate = new pako.Inflate(); // you can customize Pako here
  inflate.onData = (dat) => {
    fileReadStream.push(dat);
  };
  inflate.onEnd = () => {
    fileReadStream.push(null);
  };

  let nextChunk = null;

  file.on('data', async (data) => {
    // our Pako gets data from here
  });
  file.on('end', () => {
    // we tell Pako we reached the end here
  });
});

Step 2: We now use the file.on('data') hook to pipe our file to Pako. We use a buffer variable called nextChunk since Pako requires a true to be sent as a parameter once the file end is reached in file.on(end) hook. Below is how.


file.on('data', async (data) => {
  if (nextChunk) {
    inflate.push(nextChunk, false);
  }
  nextChunk = data;
});
file.on('end', () => {
  inflate.push(nextChunk, true);
});

This should be enough for you to get started with file compression for your own application. For more details you can of course, check out the library's documentation here or ask me!

Hope it helped!

Upload Files to S3 in Node.js

Ayush — Wed, 15 Jan 2020 11:49:02 +0000

My excitement to implement file upload didn't take too long to turn into a struggle, then dread but finally, a victorious push to the finish. This is my humble attempt to help you skip the line and jump directly to the third phase.

If you are someone who is here for a specific piece of information, you may skip to any of the below:

1. Upload file as a whole using Multer

2. Upload in chunks with Busboy

Let's begin by making an uploader for Vue.js

First off, let's allow our user to upload a file using Vue.js so that it may reach our API.

For that, we starts with the tag:

<input type="file" :accept="allowedMimes" ref="inputFile" @change="onChange"/>

The above input tag allows a user to upload a single file. Once a file is selected, the onChange method is called with the file data.

The onChange method looks like below:

function onChange() {
  const data = new FormData();
  for (const [key, value] of Object.entries(this.options)) {
    data.append(key, value);
  }

  const file = this.$refs.inputFile.files[0];
  data.append('file', fileToUpload, file.name);
  const {data: res} = await axios.post(API`/files`, data);
}

With this, our front-end is good to go and now, we are ready to send our file off to S3.

Multer-S3 saves the day

This approach will let you upload a file directly to AWS S3, without having to do anything in between.

When to use this approach:

You want to pipe your data to a location in your S3 bucket without modifying or accessing the file bytes. In short, this method will pipe your whole file without you having to do anything.

Here's how the basic skeleton looks like. It contains your multer declaration and the API endpoint.

const upload = multer({});

router.post('/file', upload.single('file'), async (req, res) => {

});

We start by specifying the upload method:

const multer = require('multer');
const multerS3 = require('multer-s3');

const upload = multer({
  storage: multerS3({
    s3, // instance of your S3 bucket
    contentDisposition: 'attachment',
    contentType: multerS3.AUTO_CONTENT_TYPE,
    bucket(req, file, callb) {
      // logic to dynamically select bucket
      // or a simple `bucket: __bucket-name__,`
      callb(null, '_my_bucket_');
    },
    metadata(req, file, cb) {
      cb(null, {
        'X-Content-Type-Options': 'nosniff',
        'Content-Security-Policy': 'default-src none; sandbox',
        'X-Content-Security-Policy': 'default-src none; sandbox',
      });
    },
    async key(req, file, abCallback) {
      try {
        // logic to dynamically select key or destination
        abCallback(null, ' _dest/key_');
      } catch (err) {
        abCallback(err);
      }
    },
  }),
  limits: {}, // object with custom limits like file size,
  fileFilter: filterFiles, // method returns true or false after filtering the file
});

We then pass it as a middleware to our API end-point.

router.post('/file', upload.single('file'), async (req, res) => {
    // you can access all the FormData variables here using req.file._var_name
});

This is it! All the data pertaining to your S3 upload will be available under the req.file variable.

With that, we have successfully uploaded your file to s3, the easy way.

When save the day with Busboy

Then comes a situation where you want to have access of the bytes you are piping to your S3 bucket, before the actual upload happens. You might want to compress them, uncompress them, check for virus, or fulfil any other endless requirements. I decided to use Busboy here, it's a tried, tested and an easy to use library. Other options you may go for are libraries like Formidable or Multiparty.

When to use this approach:

You want to access the file chunks, modify them or use them before you pipe them to your S3 bucket.

Here's how the basic structure looks like. It again, contains the basic definition along with our usual API endpoint.

const busboyUpload = (req) => {};

router.post('/file', async (req, res) => {
});

So, let's dive right in. The Busboy is called as a method from our API with the request as its parameter as defined below.

router.post('/file', async (req, res) => {
  try {
    const uploadedFileData = await busboyUpload(req);
    req.file = uploadedFileData;
    res.sendStatus(200);
  } catch (err) {
    res.sendStatus(500);
  }
}

Our Busboy uploader will be set up in a simple and straight forward manner.

We start by returning a Promise and initiate our Busboy instance along with the basic structure.

const busboyUpload = (req) => new Promise((resolve, reject) => {
  const busboy = new Busboy({});
});

We then define an array that will help us check whether the upload has finished or not. This will allow us to return a suitable response.

const fileUploadPromise = [];

In this next step, we will work on the actual file. We define the listener that executes when a file is encountered.

busboy.on('file', async (fieldname, file, filename, encoding, mimetype) => {
  // check for conditions and set your logic here
  // s3Bucket = '_Bucket_';
  // s3Key = '_Key_';
  // check file size and file type here
});

Inside the onFile listener above, we will upload to S3 using Read and PassThrough stream. The way our streams and S3 upload will be defined is:

const { Readable, PassThrough } = require('stream');
const s3 = require('@/utils/awsConnect').getS3();

const passToS3 = new PassThrough();
const fileReadStream = new Readable({
  read(size) {
    if (!size) this.push(null);
    else this.push();
  },
});
fileUploadPromise.push(new Promise((res, rej) => {
  s3.upload({
    Bucket: bucket,
    Key: key,
    Body: passToS3,
    contentDisposition: 'attachment',
  }, (err, data) => {
    if (err) {
      rej();
    } else {
      res({ ...data, originalname: filename, mimetype });
    }
  });
}));
fileReadStream.pipe(passToS3);

Whats happening here: We create the Read stream, pass it to PassThrough and after creating PassThrough we pipe it to the S3 upload function. Before beginning the upload, we push it as a Promise to the fileUploadPromise array we created earlier.

To begin the file upload, we define the following listeners inside our onFile listener. On a chunk/data event, we push the same to the Read stream that will in turn push it to our S3.

file.on('data', async (data) => {
  fileReadStream.push(Buffer.from(nextChunk));
});
file.on('end', () => {
  fileReadStream.push(null);
});

Lastly, we define our onFinish event, pipe the request to BusBoy, sit back and relax. You will notice, we wait for the fileUploadPromise to complete here before we send a response back.

busboy.on('finish', () => {
  Promise.all(fileUploadPromise).then((data) => {
    resolve(data[0]);
  })
    .catch((err) => {
      reject(err);
    });
});
req.pipe(busboy);

In the end this is how your BusBoyUpload structure should look like.

const busboyUpload = (req) => new Promise((resolve, reject) => {
  const busboy = new Busboy({ });
  busboy.on('file', async (fieldname, file, filename, encoding, mimetype) => {
    fileReadStream.pipe(passToS3);
    file.on('data', async (data) => {
    });
    file.on('end', () => {
    });
  });
  busboy.on('finish', () => {
  });
  req.pipe(busboy);
});

With this, you are well set to upload files to S3 the right way.

Or, you could even use the npm package I created: https://www.npmjs.com/package/@losttracker/s3-uploader

Thanks for reading! :)