
import Meta from "../../components/Meta"; 
import Layout from '../../layouts/Layout'

<Meta title="Data Engineering Fundamentals" tags="AWS, Data Engineering, Data Lake, Data Warehouse, DEA-C01" />

# Data Engineering Fundamentals

## Introduction

This is the first post in a series on Data Engineering with AWS. In this post, we will cover some basic terminology related to data engineering that can be useful as a foundation before diving into AWS services related to data management.

### Types of data

In data engineering, the types of data we can deal with are usually categorized into one of the main type:

* `structured` - this is highly organized data, stored in predefined format with clear and consistent schema. This type of data can be easily searchable, indexed and the relation between data is well defined. Usually, this type of data is represented by relational databases, spreadsheets or some reports (financial, sales, transaction, inventory, etc).

* `semi-structured` - this type of data does not fit into a regular and well defined schema, but still has some patterns, tags or metadata that describe records. The structure is flexible and can evolve over time. Usually, this type of data is represented by JSON/XML files, NoSQL databases, logs or emails.

* `unstructured` - data that has no predefined format or structure, making it the most difficult to store or analyze. This type of data is usually rich in information, but requires significant effort to process it to extract what's relevant. Usually, this can be represented by web pages, multimedia files, documents, social media posts, pdf documents.

### 5 V's of Big Data

The data that you, as a data engineer, will be handling can be described by 5 characteristics:

* `Volume` - refers to amount of data that needs to be ingested, transformed and stored from different sources (IoT devices, sensors, transactions, logs, applications, etc.)

* `Velocity` - refers to the speed at which data is generated and needs to be processed. When dealing with velocity, we can distinguish three main categories: real time, near real time (usually it is when a delay is between 1m-15m) or batch processing (when the delay is above 1h).

* `Variety` - refers to the diversity of data type (`structured`, `semi-structured` or `unstructured`)

* `Veracity` - refers to the accuracy and reliability of data. Veracity addresses the challenges of data quality, inconsistencies, which can impact the validity of analysis

* `Value` - refers to the insights and benefits that can be extracted from data. It emphasizes the tangible business outcomes and actionable intelligence that result from processing and transforming ingested raw data.

Based on these characteristics, you can design and implement a data pipeline that aligns with AWS best practices, ensuring scalability, efficiency, and cost-effectiveness.

### Date Lake vs Data Warehouse

`Date Lake` and `Data Warehouse` are storage for data with different characteristics and needs.

* `Date Lake` - a cost-effective storage solution for massive amounts of raw, unprocessed data of all types (structured, semi-structured, and unstructured). It applies a schema only when the data is read (schema-on-read). Data lakes are designed for big data analytics, including machine learning, artificial intelligence, and advanced data exploration. Minimal processing is required before data is stored. Common platforms providing data lake functionality include Hadoop, Amazon S3, Azure Data Lake, and Google Cloud Storage.

* `Data Warehouse`- a more expensive storage solution designed for processed and structured data with a well-defined schema. It focuses on optimized query performance rather than raw data storage. Data is typically transformed and processed before being loaded (ETL: Extract, Transform, Load). Data warehouses are ideal for performing historical queries and generating business insights. Common platforms providing data warehouse functionality include traditional database management systems (DBMS), Amazon Redshift, Google BigQuery, Snowflake or Microsoft Azure Synapse.

### Data Mesh

`Data Mesh` is an emerging paradigm for managing and handling data, where data is treated as a product. This means that each set of data, such as a dataset or a data pipeline, is owned and maintained by a dedicated team responsible for its quality, security and availability. By adopting this approach, organizations aim to decentralize data ownership and distribute responsibilities across different teams (usually aligned with specific business domains).

### Data Lineage

`Data Lineage` refers to the visual representation that traces the flow of data from its source to its destination, including all transformations applied along the way. The primary goal of data lineage is to help organizations maintain, monitor, debug, and validate their data pipelines. It provides a clear understanding of where the data originates, how it is transformed, and how it is ultimately used. This visibility can aid in troubleshooting transformation failures, monitoring pipeline efficiency, and identifying potential sensitive data leaks.

In AWS data lineage can be built sing Neptune, AWS Glue and [Spline agent](https://aws.amazon.com/blogs/big-data/build-data-lineage-for-data-lakes-using-aws-glue-amazon-neptune-and-spline/).

### Data Sampling Techniques

`Data Sampling` is a technique used by data engineers to create a representative subset of a large and complex dataset while preserving the same characteristics and patterns as the original. This technique is particularly useful for preparing testing datasets that help reduce storage and processing costs while validating and testing data pipelines. 

There is plenty of sampling methods, but we will explore a few based on the probability:

* `Simple random sampling` - every member of a population has an equal chance of being selected. In order to conduct this type of sampling, tools such as random number generators can be used.

![image](/images/aws-dea-c01-01/random_sampling.png)

* `Systematic sampling` - in systematic sampling, every member of a population is assigned a consecutive number. To conduct this type of sampling, you randomly select a starting point within the first interval (based on the population size and the required sample size) and then select individuals at regular intervals.

Suppose you have a population of 30 items and need a sample of 5 items. First, calculate the sampling interval: divide the population size (30) by the sample size (5), which gives an interval of 6. Next, randomly choose a starting point between 1 and 6. If the starting point is 4, your sample would include the following items: 4, 10, 16, 22, and 28.

![image](/images/aws-dea-c01-01/systematic_sampling.png)

* `Stratified sampling` - in stratified sampling, the population is divided into groups, or strata, based on specific characteristics. The number of individuals selected from each stratum is determined by the size of the strata relative to the total population. Individuals are then selected from each stratum using random or systematic sampling to create the final sample.

Suppose you have a population of 30 items, consisting of 20 orange items and 10 blue items, and you need a sample of 5 items. First, divide the population into two strata: 20 orange items and 10 blue items. Next, calculate how many items to select from each stratum. Based on the proportion of each stratum in the total population, select 3 items from the orange group and 2 items from the blue group.

![image](/images/aws-dea-c01-01/stratified_sampling.png)

* `Cluster sampling` - in cluster sampling, the population is divided into subgroups, or clusters, where each cluster has characteristics that closely resemble those of the overall population. Instead of selecting individuals from each cluster, entire clusters are randomly chosen, and all members within the selected clusters are included in the sample.

![image](/images/aws-dea-c01-01/cluster_sampling.png)

### Data Skew

`Data Skew` refers to a situation where data (e.g., traffic or load) is unevenly distributed across nodes, partitions, clusters, or similar structures. This phenomenon is sometimes referred to as the "hot partition" or "celebrity problem." It often occurs when the chosen key pattern for distributing data is not optimal.

Data skew can be either a temporary issue—such as high traffic caused by backfilling data when an IoT device that was offline for an extended period comes back online—or a persistent problem due to structural or design flaws in the distribution strategy.

Addressing `Data Skew` highly depends on the root cause of the problem. Here are some example solution that can be applied in certain cases:

* `Salting keys` - adding a random or calculated prefix (salt) to the keys in the dataset to distribute data more evenly across partitions. This ensures that dominant data is spread across different clusters or partitions, preventing bottlenecks and improving load balancing.
* `Range Partitioning` - using custom logic to explicitly specify ranges for partitioning based on data distribution. This approach helps distribute hot keys across a wider range of partitions, ensuring better load balancing.
* `Scaling Horizontally` - Increasing the number of nodes or partitions to reduce the load on individual nodes.

### Data Validation and Profiling

Data Validation and Profiling is a process that ensures our data is complete and can be used in analysis. There are several dimensions we can use to validate and profile our data:

* `Completeness` - validates whether all required data is present and ensures that no critical information is missing. Missing data can result in invalid or misleading insights, leading to inaccuracies in reports and data analysis. Completeness validation covers counting null values, checking missing data or calculating percentage of missing values.
* `Consistency` - validates whether data from different sources is consistent across the entire dataset. Inconsistent data can cause confusion and negatively affect insights. Consistency validation involves cross-checking data across sources to ensure the same target scale or standard is used (e.g., Celsius for temperature across all data sources).
* `Accuracy` - validates whether the data is accurate, reliable, and correctly represents what it is supposed to. Inaccurate data can lead to poor decisions and false insights. Accuracy validation involves cross-checking the data against multiple sources and verifying it against known standards and rules.
* `Integrity` - ensures that data maintains its relationships and consistency throughout the entire data pipeline. Integrity can be enforced by preserving relationships between data using foreign keys or other database techniques.

## Summary

In this post, we covered fundamental terminology related to data engineering. In the upcoming posts, we will delve deeper into AWS services, with a focus on designing and implementing data pipelines while adhering to well-architected principles. Stay tuned for more insights!

Data Engineering Fundamentals


import Meta from "../../components/Meta"; 
import Layout from '../../layouts/Layout'

<Meta title="Understanding Node.js - Memory management" tags="Node.js, Memory leak, Memory management, Profiling" />

# Understanding Node.js - Memory management

## Introduction

This is the fourth article in our series on understanding the architecture of Node.js. In this post, we explore how Node.js manages memory, discuss methods to detect memory leaks, and review some of the most common issues that can cause them.

While this is the final article focusing on the core architecture of Node.js, it is by no means the last one in our series about Node.js and its ecosystem.

## Memory management

Memory management involves allocating and deallocating memory. In the case of a Node.js application, this responsibility lies with the V8 engine, which handles it automatically. While the process is largely automated, it is essential to understand how it works to write programs that assist the engine in managing memory efficiently and avoiding memory leaks.

### Memory allocation

Node allocates memory in the following area:

#### Stack
The stack is a dedicated memory area used for managing local variables, function calls, and the execution context of a program. It is organized as a Last In, First Out (LIFO) data structure, which ensures efficient management of data (all operations are performed at the top of the stack).

The stack is automatically managed, meaning:
* Memory for variables is allocated when a function is called.
* The memory is automatically reclaimed when the function ends or the program exits.
* This makes the stack efficient for short-lived, scoped variables.

#### Heap
The heap is a large block of memory reserved for dynamically allocated data such as objects, arrays, and closures. It is divided into the following spaces:

* `Young Generation Space`: A smaller section of the heap designed for newly created objects that are likely to have a short lifespan. This space typically starts at 8 MB (on 32-bit systems) or 16 MB (on 64-bit systems).
* `Old Generation Space`: A larger portion of the heap allocated for objects that need to persist longer in memory. It is significantly bigger compared to the Young Generation Space.
* `Code Space`: A dedicated area in the heap used for storing the code that is being executed.
* `Large Object Space`: A specialized space for objects that are too large to be efficiently stored in either the Young or Old Generation spaces. Allocating such objects here avoids the high performance cost of copying them during garbage collection.

### Memory releasing

Memory releasing is the process of removing unused objects from memory to reclaim space for new ones. In Node.js, this task is managed by a mechanism called the *Garbage Collector*.

The *Garbage Collector* often operates during idle phases when the CPU has fewer tasks to handle. Its operation is divided into two phases:

1. `Scavenging (Minor GC)` – A fast process that scans only the *young generation* of objects, removes unused ones, and promotes longer-lived objects to the old generation. If an object survives two or more scavenging cycles, it is moved to the old generation space.
1. `Mark-Sweep (Major GC)` – A slower garbage collection process that scans the *old generation*. It begins by starting with the root object, traversing all referenced objects, and marking them during the first phase *(mark)*. In the second phase *(sweep)*, it removes all unmarked objects.

Consider following code:

```ts
class Engine {
  constructor(public power: number) {
    this.power = power;
  }
}

class Car {
  private engine: Engine;
  constructor(public model: string, private power: number) {
    this.model = model;
    this.engine = new Engine(power);
  }

  public replaceEngine(newPower: number) {
    this.engine = new Engine(newPower);
  }
}
```

And some code that uses the classes:

```ts
const car1 = new Car(‘car1’, 200)
const car2 = new Car(‘car2’, 400)
const car3 = new Car(‘car3’, 150)
```

we could visualize the object graph as below: 

![image](/images/understanding-nodejs-memory-part1/memory-1.svg)

And now we will call:

```ts
car2.replaceEngine(200)
car3 = null
```

As a result the graph will change to:

![image](/images/understanding-nodejs-memory-part1/memory-2.svg)

For the simplicity, assume that all objects are located now in `Old Generation Space`. In that case, they will be a handled by `Mark-Sweep (Major GC)` phase. In the first phase *(mark)*, the algorithm will start with the root object, traverse all referenced objects and marking them. So, the graph could look like:

![image](/images/understanding-nodejs-memory-part1/memory-3.svg)

All objects referenced from root were marked and colored with blue. The second phase *(sweep)*, will find all unmarked object (green ones) and remove them to reclaim space. After that, the objects graph will look following:

![image](/images/understanding-nodejs-memory-part1/memory-4.svg)

However, in some cases, objects that are no longer needed may still be referenced by the root, even though they could be safely removed to reclaim space. This situation results in a phenomenon known as a memory leak.

## Memory leaks

When a memory leak occurs, your program starts occupying more and more memory, causing the overall performance to degrade. Eventually, the program may crash due to insufficient space for new objects, resulting in an "Out of Memory" error. Below are some common bad coding practices that can lead to memory leaks:s

### Using global variables

Global variables are always referenced by a root object and stay in memory as long as the main program is running.

```ts
type CacheValue = {
  timestamp: number;
  value: number;
};

const cache: Record<string, CacheValue> = {};

app.get("/add/:a/:b", (req, res) => {
  const { a, b } = req.params;
  const cacheKey = `${a}_${b}`;
  if (cache[cacheKey]) {
    const { value } = cache[cacheKey];
    res.send(`Cached result: ${value}`);
    return;
  }
  const value = parseInt(a) + parseInt(b);
  cache[cacheKey] = {
    timestamp: Date.now(),
    value,
  };
  res.send(`Calculated result: ${value}`);
});
```

In the above example, the cache only appends new values, leaving all older or unused values in memory. These values will not be removed by the garbage collector. As a result the cache will start occupying more and more memory.

**Mitigate strategy**

1. **Proactively manage the lifecycle of global variables**: For caches, consider implementing strategies such as time-based data removal or setting a maximum size for the cache to prevent unbounded memory growth.

### Closure based memory leak

The inner function maintains a reference to data from outer function, preventing garbage collection even if the function is no longer needed.

```ts
function generateArray(length: number) {
  const outerData = new Array(length).fill("*");

  return function processData() {
    console.log(`Generated ${outerData.length} elements`);
  };
}

function closureLeak() {
  const functions = [];
  for (let i = 0; i < 1000; i++) {
    const largeArrayGenerator = generateArray(10_000);
    functions.push(largeArrayGenerator);
  }
  console.log("Functions created:", functions.length);
}
```

The closure in the `generateArray` function captures the entire `outerData` array, even though the inner `processData` function does not utilize it directly. This prevents the array from being garbage collected, leading to unnecessary memory consumption.

**Mitigation strategy**

1. **Prevent memory leaks caused by lingering references in closures**: provide a mechanism to release or nullify the references when they are no longer needed. This ensures that large data structures can be garbage collected.

```ts
function generateArraySafe(length: number) {
  let outerData = new Array(length).fill("*");

  function processData() {
    console.log(`Generated ${outerData.length} elements`);
  }

  function cleanup() {
    outerData.length = 0;
  }

  return { processData, cleanup };
}
```

### Timers/intervals based memory leak

Timers and intervals in Node.js can cause memory leaks if not properly managed. When a timer (`setTimeout`) or interval (`setInterval`) is started but not cleared after completing its work, the associated callback function and its resources can remain in memory indefinitely. This prevents the garbage collector from freeing them, leading to potential memory leaks. This issue is particularly significant when the callback function captures resources from its enclosing scope, resulting in a *closure-based memory leak*.

```ts
setInterval(() => {
  // Do something repeatedly
}, 1000);
// If this interval is never cleared, the callback stays in memory.
```

**Mitigation strategy**

1. **Clear timers and intervals when they are no longer needed**: Use `clearTimeout` or `clearInterval` to stop the timer or interval once its purpose is served.

```ts
const intervalId = setInterval(() => {
  // Do something repeatedly
}, 1000);
// release memory when work done
clearInterval(intervalId)
```

2. **Minimize captured resources in closures**: Avoid referencing variables or functions from the outer scope in timer or interval callbacks, unless absolutely necessary, to reduce the risk of *closure-based memory leaks*.

### Unclosed event listeners

The *EventEmitter* class in Node.js has a default limit of 10 listeners per event. Adding more listeners without increasing this limit triggers the following warning:

`(node:57) MaxListenersExceededWarning: Possible EventEmitter memory leak detected`

This warning typically arises when you add listeners repeatedly, such as within a loop or other event-driven processes, without removing them when **they are no longer needed**. 

For example:

```ts
import EventEmitter from "events";

class MyEmitter extends EventEmitter {}

const emitter = new MyEmitter();

for (let i = 0; i < 15; i++) {
  emitter.on("data", (data) => console.log(data));
}
// Warning: MaxListenersExceededWarning
```

**Mitigation strategy**

1. **To avoid memory leaks, always remove listeners when they are no longer needed**. Use either:
* emitter.off("data", listener) (preferred modern API)
* emitter.removeListener("data", listener) (older API, still valid)

Example:

```ts
const listener = (data: string) => console.log(data);
emitter.on("data", listener);

// When the listener is no longer needed:
emitter.off("data", listener);
```

2. **Increase Listener Limit (If Necessary)**. If you intentionally need more than 10 listeners, increase the limit using `setMaxListeners()`:

```ts
emitter.setMaxListeners(20);
```

However, make sure you increases the limit in order to handle all required listeners, not to avoid a warning if you do not expect a higher number.

3. **Use once for One-Time listeners**. For events that need handling only once, use `emitter.once()` instead of `emitter.on()`. For example:

```ts
emitter.once("data", (data) => console.log(data));
```
The above will remove the listener automatically after execution.

## Detecting memory leaks

Node.js provides built-in tools to help analyze and detect memory leaks in your application. These tools allow you to, for instance, trace garbage collection statistics to monitor how memory is managed or generate heap dumps to analyze memory usage over time.

### Tracing garbage collector

In order to trace garbage collector you can start a NodeJS program with `--trace-gc` argument. Create a following program that has a memory leak.

```ts
import express from "express";

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

const memoryLeakArray = [];

function leakMemory() {
  const hugeObject = {
    data: new Array(100_000).fill("*"),
  };
  memoryLeakArray.push(hugeObject);
}

// generate memory leak every 5 seconds
setInterval(leakMemory, 5 * 1000);

app.listen(port, () => {
  console.log(`Example app listening on port ${port}`);
});
```

and start the program with `node --trace-gc dist/index.js`. After a while you should see on the console GC output:

```txt
[40837:0x148008000]       50 ms: Scavenge 4.6 (4.8) -> 4.1 (5.8) MB, 0.3 / 0.0 ms  (average mu = 1.000, current mu = 1.000) allocation failure; 
[40837:0x148008000]       67 ms: Scavenge 5.4 (6.6) -> 4.9 (7.4) MB, 0.3 / 0.0 ms  (average mu = 1.000, current mu = 1.000) allocation failure; 
[40837:0x148008000]       98 ms: Scavenge 5.9 (7.6) -> 5.3 (7.6) MB, 0.2 / 0.0 ms  (average mu = 1.000, current mu = 1.000) allocation failure; 
[40837:0x148008000]      119 ms: Scavenge 6.2 (7.6) -> 5.7 (10.4) MB, 0.3 / 0.0 ms  (average mu = 1.000, current mu = 1.000) allocation failure; 
[40837:0x148008000]      149 ms: Scavenge 8.2 (10.5) -> 7.2 (11.3) MB, 0.4 / 0.0 ms  (average mu = 1.000, current mu = 1.000) allocation failure; 
[40837:0x148008000]      165 ms: Scavenge 8.6 (11.3) -> 7.8 (12.0) MB, 0.5 / 0.0 ms  (average mu = 1.000, current mu = 1.000) allocation failure; 
Example app listening on port 3000
[40837:0x148008000]     8286 ms: Mark-sweep (reduce) 9.4 (12.8) -> 7.4 (9.9) MB, 2.7 / 0.0 ms  (+ 6.4 ms in 23 steps since start of marking, biggest step 3.6 ms, walltime since start of marking 10 ms) (average mu = 1.000, current mu = 1.000) finalize incremental marking via task; GC in old space requested
[40837:0x148008000]     8898 ms: Mark-sweep (reduce) 7.4 (9.9) -> 7.4 (9.4) MB, 3.5 / 0.0 ms  (+ 7.0 ms in 40 steps since start of marking, biggest step 0.4 ms, walltime since start of marking 11 ms) (average mu = 0.983, current mu = 0.983) finalize incremental marking via task; GC in old space requested
[40837:0x148008000]    15175 ms: Scavenge 8.2 (10.2) -> 8.2 (10.2) MB, 1.2 / 0.0 ms  (average mu = 0.983, current mu = 0.983) allocation failure; 
[40837:0x148008000]    20177 ms: Scavenge 8.9 (11.0) -> 8.9 (11.0) MB, 2.9 / 0.0 ms  (average mu = 0.983, current mu = 0.983) allocation failure; 
```

Based on the first line the values corresponds to:

* `[40837:0x148008000]` - PID of the running process and JS heap instance     
* `50 ms` - the time since the process started in ms
* `Scavenge` - type/phase of a garbage collector 
* `4.6` - heap used before GC in MB
* `(4.8)` - total heap before GC in MB
* `4.1` - heap used after GC in MB
* `(5.8)` - total heap after GC in MB
* `0.3 / 0.0 ms  (average mu = 1.000, current mu = 1.000)` - time taken to run GC
* `allocation failure` - Reason for GC

Using the GC output you can see how your program manages the memory. 

### Analyzing of a heapdump

However, if it turns out that your program has a memory leak, you can use embedded tools to dump the heap to a file over time and perform an analysis to identify the cause.

Using `v8` module you can do a heap snapshot. Let's extend our previous program with the following functionality:

```ts
import * as v8 from "v8";

//
function createHeapSnapshot() {
  const snapshotStream = v8.getHeapSnapshot();
  const fileName = `${Date.now()}.heapsnapshot`;
  const fileStream = fs.createWriteStream(fileName);
  snapshotStream.pipe(fileStream);
  console.log(`Heap snapshot created: ${fileName}`);
}

// Run the function every 5 seconds
setInterval(leakMemory, 5 * 1000);

// Create heap snapshot every 10 seconds
setInterval(createHeapSnapshot, 10 * 1000);
```

And run program for a while (to have at least two heap snapshots). If you have two heap snapshots available, you can now analyze them. We will use `Chrome` internat browser (as it uses also v8 engine and has built in tool for heap analysis). 
1. Open Chrome **DevTools**
2. Navigate to the **Memory tab**
3. In the **Memory tab**, click the **Load profile** button and select first (older) file with heap snapshot
4. Load the second file 
5. Finally you should end up with similar view:

![image](/images/understanding-nodejs-memory-part1/dev-tools-1.png)

6. There are numerous options available for analyzing the data; however, we will focus on comparing objects created between the two points in time when the files were generated. To proceed, please select the second file, ensure the "Summary" option remains selected, and choose "Objects allocated between xxx and yyy," as shown in the image below:

![image](/images/understanding-nodejs-memory-part1/dev-tools-2.png)

The aspect that should catch our attention is the very first two lines of the report. While the *Shallow Size* of the data object does not occupy much space, the *Retained Size* is significant and is almost equal to the *Shallow* and *Retained Size* of the second entry (Array).

The *Shallow Size* indicates the amount of memory used directly by an object itself, excluding any references to other objects. The *Retained Size* represents the total memory that would be freed if a given object were garbage collected, including the memory used by all objects it keeps alive (directly or indirectly).

Based on this, we can assume that the data object may hold references to other objects, potentially causing memory usage to grow. Most likely, it is an array. We can now select the second row and expand it to gather more details, such as the object's name and where it is allocated.

![image](/images/understanding-nodejs-memory-part1/dev-tools-3.png)

Expanding the object reveals that it contains the * values. It is an array named `memoryLeakArray`, created and populated within the context of the `leakMemory` function.

We can now examine our code to identify and resolve the cause of the memory growth or leak. In this case, the issue was straightforward to locate because the code was simple. However, in a real-world scenario, it can be more complex. Nevertheless, the steps we followed will help you better understand memory allocation in your program.

# Summary

This concludes the series of articles on the core architecture of Node.js. I hope you found this article, along with the entire series, useful. The goal was to provide you with a better understanding of CPU and memory management in Node.js applications, enabling you to create robust and efficient applications.

Understanding Node.js - Memory management


import Meta from "../../components/Meta"; 
import Layout from '../../layouts/Layout'

<Meta title="Understanding Node.js - Profiling" tags="Node.js, Event Loop, CPU, Profiling" />

# Understanding Node.js - Profiling

## Introduction

This is the third and final article in our series on the architecture of Node.js, focusing specifically on threading and CPU management. Our upcoming articles will shift to topics related to memory management and profiling.

In this post, we’ll explore Node.js's built-in capabilities for profiling and identifying performance bottlenecks. These tools help detect code inefficiencies that may slow down the application and degrade overall performance.

## Problem

Please take a look at the following code (it returns the size of a requested file passed on as a query parameter). 

```ts
import express from "express";
import { randomBytes } from "node:crypto";

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

app.get("/content", (req, res) => {
  const file = req.query.file as string;
  console.log(`Reading file: ${file}`);
  const content = fs.readFileSync(file);
  res.send({
    message: "File read",
    length: content.length,
  });
});

app.listen(port, () => {
  console.log(`Example app listening on port ${port}`);
});
```

At first glance, it does not look as something that can cause any performance issues. So, let us try to benchmark the endpoint with `ab` tool. Start the application and run following in the terminal:

```bash
ab -k -c 20 -n 300 "http://localhost:3000/content?file=file.zip" 
```

After a while, you should see the results:

```txt
Benchmarking localhost (be patient)
Completed 100 requests
Completed 200 requests
Completed 300 requests
Finished 300 requests


Server Software:        
Server Hostname:        localhost
Server Port:            3000

Document Path:          /content?file=file.zip
Document Length:        42 bytes

Concurrency Level:      20
Time taken for tests:   26.227 seconds
Complete requests:      300
Failed requests:        0
Keep-Alive requests:    300
Total transferred:      83100 bytes
HTML transferred:       12600 bytes
Requests per second:    11.44 [#/sec] (mean)
Time per request:       1748.483 [ms] (mean)
Time per request:       87.424 [ms] (mean, across all concurrent requests)
Transfer rate:          3.09 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    0   0.5      0       2
Processing:    82 1688 266.2   1748    1933
Waiting:       82 1688 266.3   1748    1933
Total:         82 1688 266.0   1748    1936

Percentage of the requests served within a certain time (ms)
  50%   1748
  66%   1752
  75%   1759
  80%   1763
  90%   1770
  95%   1775
  98%   1781
  99%   1783
 100%   1936 (longest request)
```

The key takeaways from the output show the it around `26 seconds` to complete all the requests. The mean is about `1.7 requests` per second and more then `50%` of requests took longer then `1.7 seconds`. It's not bad (in my case, the file zip was about 600MB), but let's see if there is something we could improve.

## Profiling


We will run the same test, but this time with the application in profiling mode. To do this, modify the scripts section in your `package.json` file and add the following line:

```json
"start:profiling": "tsc && node --prof dist/index-profiling.js"
```

Here, `index-profiling.js` is the script we want to profile, and the `--prof` flag enables V8 profiling, generating a profiling log file.

Start the application and run the benchmark again. Once the test completes, close the application. This will generate a log file named something like `isolate-0xnnnnnnnnnnnn-v8.log`, which may be hard to read. To convert this file into a more readable format, run:

```bash
node --prof-process isolate-0xnnnnnnnnnnnn-v8.log > profiling_result.txt
```

Open `profiling_result.txt` in your preferred text editor and scroll to the `[Summary]` section. You’ll see a breakdown similar to this:

```txt
 [Summary]:
   ticks  total  nonlib   name
     56    2.8%    3.0%  JavaScript
   1834   93.3%   96.9%  C++
    819   41.7%   43.3%  GC
     74    3.8%          Shared libraries
      2    0.1%          Unaccounted
```

As shown, most of the time was spent in the `C++` section, with a significant portion in `Garbage Collection (GC)` as well. 

We will focus on the C++ section, which might be related to the Node.js module we are using. Please scroll down to the `[Bottom up (heavy) profile]` section to see the most resource-intensive operations. There is already something that should catch our attention:

```txt
   ticks parent  name
    867   44.1%  t std::__1::__shared_ptr_emplace<node::ThreadsafeCopyOnWrite<std::__1::map...., 
    304   35.1%    Function: ^readSync node:fs:701:18
    304  100.0%      Function: ^tryReadSync node:fs:440:21
    304  100.0%        Function: ^readFileSync node:fs:461:22
    295   97.0%          Function: ^<anonymous> /Users/marcinmalinowski/Projects/Priv/event-loop/dist/index-profiling.js:10:21
    295  100.0%            Function: ^handle /Users/marcinmalinowski/Projects/Priv/event-loop/node_modules/express/lib/router/layer.js:86:49
      9    3.0%          LazyCompile: ~<anonymous> /Users/marcinmalinowski/Projects/Priv/event-loop/dist/index-profiling.js:10:21
```

As we might have expected, we used the synchronous version of the `readFile` function, which blocks the event loop and degrades performance. Since we found the problem in our code, we can try to fix that.

## Solution

In order to fix that we could see if there is an async version of a `readFile` functionality we could use. After checking the documentation, we could change the code to utilize the async version:

```ts
app.get("/content", (req, res) => {
  const file = (req.query.file as string) || "";
  console.log(`Reading file: ${file}`);
  fs.readFile("file.zip", (_, data) => {
    res.send({
      message: "File read",
      size: data.length,
    });
  });
});
```

We can run the server again and launch benchmarking with `ab` tool. After a while we should get the results:

```txt
Benchmarking localhost (be patient)
Completed 100 requests
Completed 200 requests
Completed 300 requests
Finished 300 requests


Server Software:        
Server Hostname:        localhost
Server Port:            3001

Document Path:          /content?file=file.zip
Document Length:        40 bytes

Concurrency Level:      20
Time taken for tests:   14.850 seconds
Complete requests:      300
Failed requests:        0
Keep-Alive requests:    300
Total transferred:      82500 bytes
HTML transferred:       12000 bytes
Requests per second:    20.20 [#/sec] (mean)
Time per request:       989.978 [ms] (mean)
Time per request:       49.499 [ms] (mean, across all concurrent requests)
Transfer rate:          5.43 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    0   0.4      0       2
Processing:   108  980 561.2    816    3385
Waiting:      107  980 561.2    815    3385
Total:        108  980 561.6    816    3387

Percentage of the requests served within a certain time (ms)
  50%    816
  66%    831
  75%    899
  80%    936
  90%   1183
  95%   2921
  98%   2982
  99%   3318
 100%   3387 (longest request)
```

As you can see, this time we achieved nearly a 50% improvement (the test took 15 seconds to run, compared to 26 seconds previously).

## Summary

Although our program was simple, we could easily identify potential issues based on function names. The goal here was to demonstrate how to use built-in Node.js tools to profile CPU usage and optimize the application. I hope you found it interesting! In the next post, we'll focus on memory management, so stay tuned!

Understanding Node.js - Profiling


import Meta from "../../components/Meta"; 
import Layout from '../../layouts/Layout'

<Meta title="Understanding Node.js - Performance optimization" tags="Node.js, Performance optimization, CPU, JS" />

# Understanding Node.js - Performance optimization

## Introduction

This is the second post related to understanding architecture of Node.js. In this article we will focus on understanding how we can optimize and manage heavy CPU tasks to make our app responsive. In previous post we understood the event loop concept and now we can try to utilize that knowledge. Also, we will cover a few best practices how to make and handle promises in efficient way.

## Problem

Let's say we have a simple Express application with two GET endpoints. One endpoint responds quickly with a simple `Hello World!` message, while the other performs a heavy CPU-bound calculation to return a result (in this case, a complex math operation that requires significant processing power). Here’s the example:

```ts
import express from "express";

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

function calculateBlocking() {
  let n = 10_000_000_000;
  let sum = 0;
  for (let i = 1; i < n; i++) {
    sum += i * 0.31232342321;
  }
  let avg = sum / n;
  return avg;
}

app.get("/hello", (req, res) => {
  res.send("Hello World!");
});

app.get("/blocking", (req, res) => {
  const result = calculateBlocking(); // CPU heavy operation
  res.send("Hello from blocking  " + result);
});
```

The problem we are facing now is that is someone would call `/blocking` endpoint, the event loop's thread will be busy and `/hello` would have to wait for `calculateBlocking()` to be completed and the main thread to be released.

Now, we will explore how we can  fix that and make our server more responsive.

### Naive solution

The simplest solution that might come to mind is to wrap the blocking function within a `Promise` and make it `async`. Let's try it and expose the operation with one more endpoint:

```ts
function calculateNonBlocking(): Promise<number> {
  return new Promise((resolve, reject) => {
    const result = calculateBlocking();
    resolve(result);
  });
}

app.get("/non-blocking-naive", async (req, res) => {
  const result = await calculateNonBlocking();
  res.send("Hello from non-blocking  " + result);
});
```

Start the application and open `/non-blocking-naive` in one tab and `/hello` in a second tab.

As you might have expected (or maybe not), this approach doesn’t work. Simply making the code `async` by wrapping it in a `Promise` doesn’t change much. The code still runs synchronously, blocking the event loop and making the application unresponsive. The first takeaway here is that wrapping code in a `Promise` doesn’t automatically make it run asynchronously. 

So, what we can do to make it run as we would expect? Actually, there are two approaches I would like to propose.

### Partitioning

*Partitioning* is a technique that breaks down large tasks into smaller ones, which are then executed in consecutive event loop cycles, one by one, collecting the results once all are completed.

In our current blocking function, we have a long-running loop that executes in a single cycle:

```ts
let n = 10_000_000_000;
let sum = 0;
for (let i = 1; i < n; i++) {
  sum += i * 0.31232342321;
}
```

To improve this, we can split the work to run across multiple cycles, calculating the value for the next 50,000 elements in each cycle:

```ts
let n = 10_000_000_000;
let sum = 0;
for (let i = 1; i < 50000; i++) {
  sum += i * 0.31232342321;
}
for (let i = 50000; i < 100000; i++) {
  sum += i * 0.31232342321;
}
//...
```

This approach can release the event loop after processing each segment, allowing it to handle incoming requests in between. To achieve this, we can use one of the timers functions available in Node.js. Specifically, we can utilize `setImmediate` to ensure that requests are handled after each poll phase. To implement this, we need to introduce a helper function that performs calculations for each segment and schedules the work for the next cycle.

```ts
function calculateNonBlockingWithImmediate(): Promise<number> {
  return new Promise((resolve, reject) => {
    let n = 10_000_000_000;
    const partition = 50_000;
    let sum = 0;

    function calculateSegment(i: number, callback: (result: number) => void) {
      for (let k = i; k < i + partition && k <= n; k++) {
        sum += k * 0.31232342321;
      }

      if (i >= n) {
        callback(sum);
        return;
      }

      setImmediate(() => calculateSegment(i + partition, callback));
    }

    calculateSegment(1, (sum) => {
      let avg = sum / n;
      resolve(avg);
    });
  });
}

app.get("/non-blocking-immediate", async (req, res) => {
  const result = await calculateNonBlockingWithImmediate();
  res.send("Hello from blocking  " + result);
});
```

By running this code, you can access both endpoints simultaneously. Applying task partitioning has made our application more responsive. This is one of the techniques you should use if your application performs heavy CPU operations, as it helps avoid blocking the event loop and ensures that your application remains responsive.

### Off-loading

The other approach is called *off-loading* and the main purpose is to delegate heavy and long running task to a separate thread.

In Node.js, this can be done with [worker threads](https://nodejs.org/api/worker_threads.html) module. 

A Worker Thread is an independent JavaScript execution thread with its own event loop. Each worker runs in isolation. It means it has its own memory, global objects, and runtime without access to main program's variables. However, workers can communicate with each other and with the main thread using message passing.

* `Worker` class is used to create a new worker. As an argument, you need to pass on a path to worker script, e.g. `new Worker('./my-worker.js)`.
* `Messages` - to exchange a communication between main thread and a worker you can use:
  * `on('message', callback)` - to handle message sent
  * `postMessage(data)` - to post a message (either worker or main thread)
* `parentPort` - to reference main program worker

So, let's start with creating a new file, e.g. `worker.ts` and adding a worker content to it

```ts
import { parentPort } from "worker_threads";

parentPort.on("message", (task) => { // receive a request to start working
  console.log("started task: ", task);
  const result = calculateBlocking();
  console.log("completed task: ", task);
  parentPort.postMessage(result); // send the results back to main thread using `parentPort`
});

function calculateBlocking() {
  let n = 10_000_000_000;
  let sum = 0;
  for (let i = 1; i < n; i++) {
    sum += i * 0.31232342321;
  }
  let avg = sum / n;
  return avg;
}
```

and use the worker in our express application

```ts
import express from "express";
import { Worker } from "worker_threads";

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

const worker = new Worker("./dist/worker.js");

function calculateWithWorker(): Promise<number> {
  return new Promise((resolve, reject) => {
    worker.postMessage("calculate using worker");
    worker.on("message", (result) => {
      resolve(result);
    });
    worker.on("error", (error) => {
      reject(error);
    });
  });
}

app.get("/hello", (req, res) => {
  res.send("Hello World!");
});

app.get("/non-blocking-worker", async (req, res) => {
  const result = await calculateWithWorker();
  res.send("Hello from blocking  " + result);
});

app.listen(port, () => {
  console.log(`Example app listening on port ${port}`);
});
```

When you run the application, the results will match those in our previous example. The application will still respond to other threads, and the main event loop won’t be blocked.

The main difference between partitioning and offloading lies in where the task is executed. In the partitioning approach, the calculation remains in the same thread as the event loop. This means that if there are many tasks or if parallel processing is required, the single thread may become a bottleneck, potentially impacting application performance as it takes on more work.

In contrast, offloading involves delegating each task to a separate worker, which can also affect system resources. If we anticipate a high volume of heavy tasks, creating a pool of reusable workers is advisable to better manage resource usage and maintain performance.

## Promises

Before we move to the summary, here are a few tips for handling promises that can improve your code’s performance and robustness.

### Use async/await for readability

`async/await` syntax makes asynchronous code more readable and looks like synchronous code. It should be used in favour of `then` promise chaining.

```ts
async function fetchData() {
    try {
        const data = await fetchDataFromDatabase();
        console.log(data);
    } catch (error) {
        console.error(error);
    }
}
```

### Avoid nesting promises whenever possible

Nesting promises can make the code unreadable, so instead of proper chain them or use `async/await` syntax:

```ts
fetchData()
    .then(data => fetchAdditionalData(data)
    .then(moreData => processResults(moreData)))
    .catch(error => console.error(error));

// Instead, do this
fetchData()
    .then(fetchAdditionalData)
    .then(processResults)
    .catch(error => console.error(error));
// or
try {
  const data = await fetchData()
  const moreDate = await fetchAdditionalData(data)
  const result = await processResults(moreData)
} catch (error) {
  console.error(error)
}
```

### Avoid unnecessary waits

If the data can be fetch in parallel, do not use unnecessary `await` and process them in parallel using `Promise.all/Promise.allSettled` utility functions.

```ts

const posts = await fetchUserPosts(userId)
const orders = await fetchUserOrders(userId)

// Instead, do this
const [posts, orders] = await Promise.all([
  fetchUserPosts(userId),
  fetchUserOrders(userId)
])

```

Keep in mind that `Promise.all` fails if any promise rejects. Consider using `Promise.allSettled` if you want all promises to complete, regardless of success or failure.

### Wrap with try/catch/finally

Make sure that your `async/await` code is wrapped into `try/catch` block in order to handle promise rejection correctly. If there are resources to be cleaned up (regardless if succeeded or failed), put that in to `finally` block. 

Keep in mind that unhandled promise rejections can disrupt the event loop and potentially terminate the process. It is also a good practice to add global unhandled promise rejection listener to caught all promises that do not have proper error handling to report and fix them.

```ts
process.on('unhandledRejection', (reason, promise) => {
    console.error('Unhandled promise rejection at:', promise, 'reason:', reason);
});
```

### Wrap callbacks into Promises

To improve code readability, wrap all callbacks in promises to avoid *callback hell* and handle them more efficiently.

```ts

function withCallback(param: string, callback: (result: number) => void) {
  // some logic here
  const result = // ... do some calculation here
  callback(result);
}

function wrapWithPromise(param: string): Promise<number> {
  return new Promise((resolve) => {
    withCallback(param, (result) => resolve(result));
  });
}

// so instead of
withCallback('example parameter', (result: number) => {
  console.log('The calculated value is: ', result)
})

// it can be
const result = await wrapWithPromise('example parameter')
console.log('The calculated value is: ', result)
```

# Summary

I hope you found this post helpful and gained a basic understanding of how to handle CPU-intensive tasks while keeping your application robust and responsive. We explored two approaches to prevent blocking the event loop during computations. Based on your requirements, you should now have a better idea of which approach to choose in different scenarios.

Also, keep in mind that simply wrapping code in a Promise does not make a function asynchronous. This is especially important when adding dependencies or third-party libraries that, theoretically, are supposed to be asynchronous based on their documentation.

In the next post, we’ll explore Node.js’s built-in profiling tools to identify bottlenecks and optimize your application. Stay tuned!

Understanding Node.js - Performance optimization


import Meta from "../../components/Meta"; 
import Layout from '../../layouts/Layout'

<Meta title="Understanding Node.js - Event loop" tags="Node.js, Event Loop, CPU, JS" />

# Understanding Node.js - Event loop

## Introduction

This article is the first in a series exploring the architecture of Node.js, with a focus on threads and memory management. Throughout this series, we will examine the threading model, memory management techniques, and built-in profiling tools available in Node.js to optimize application performance.

In this installment, we’ll concentrate on thread management, addressing the question of how Node.js efficiently handles multiple requests despite its single-threaded architecture.

## Event Loop

Node.js uses the “Single Threaded Event Loop” architecture to handle multiple concurrent clients. Its processing model is based on event-driven architecture with a callback pattern.

The entire "Event Loop" architecture in Node.js is structured around macro and microtask queues. The macrotask queue consists of several main phases, each with distinct responsibilities and designed to handle specific types of tasks. Complementing this, the microtask queue is processed at the end of each macrotask phase, allowing for efficient task prioritization.

Each phase has a FIFO queue of callbacks to be executed. The time a thread spends in each phase depends on factors such as the number of tasks in the queue or the maximum allowed time to be spent in the phase. 

When a task like network communication, file read/write or database access is initiated, Node.js does not wait for the operation to complete. Instead, it delegates the task to the system’s kernel or utilizes additional worker threads through the Node.js `libuv` library, which is responsible for asynchronous operations. The event loop can then continue processing other code, using the main thread.

The high level architecture diagram looks following:

![image](/images/understanding-nodejs-cpu-part1/event_loop.png)

This architecture enables Node.js to handle multiple requests on a single thread with high efficiency and performance.

## Macro tasks

### Timers phase

This phase handles the callbacks for timers that were scheduled using the `setTimeout` and `setInterval` functions. A timer specifies a minimum delay before the callback can be executed, but it does not guarantee an exact execution time. The callback will be triggered as soon as possible after the specified time has elapsed, depending on the state of the event loop and other phases' queues, as well as the operating system.

There is also another timer function, setImmediate, but it is not handled during the Timers phase.

### Pending Callbacks phase

The purpose of this phase is to handle I/O-related callbacks that were deferred to the next cycle of the event loop. This phase processes callbacks for specific types of system operations, particularly those involving non-blocking I/O, such as errors from file system operations or network requests.

### Idle phase

This phase is not related to execution of application code, but rather for Node.js to perform some memory housekeeping or other system related tasks.

### Poll phase

The purpose of this phase is to manage a queue of I/O-related callbacks. It also determines how long to block and poll from the queue. When the queue becomes too long, causing the scheduled timers to be significantly delayed, the process may interrupt the handling of pending callbacks to allow other phases to proceed.

If all the callback from the queue are processed, two things may happen:
1. There are callbacks scheduled with `setImmediate` - it will leave the poll phase and enter check phase to handle `setImmediate` callbacks.
2. There are no callbacks related to timers - it may stay in idle state waiting for new I/O operations.

### Check phase

This is a special phase to handle callbacks scheduled with `setImmediate` timer.

### Close Callbacks phase

The Close Callbacks phase is the last phase in every event loop iteration. It is responsible for cleaning up resources that were closed unexpectedly, such as when socket.destroy() is called or when a file stream closes due to a read/write error. Callbacks registered for these events (often through event listeners like 'close') are executed in this phase and will not be used in subsequent cycles.

Close callbacks are commonly associated with network sockets, file streams, and other resources that require proper cleanup to prevent resource leaks and ensure efficient memory usage.

## Micro tasks

Microtasks are a specific type of task that are handled by the main thread in JavaScript. They are stored in a special microtask queue and are executed at the end of the current script execution, before the event loop moves on to the next phase (which could be handling I/O operations, timers, or other macrotasks).

Microtasks are typically created in response to promise resolutions (using `.then()`) or rejections (using `.catch()`). When a promise is resolved or rejected, its associated callback is queued in the microtask queue, ensuring that it is handled immediately after the current execution context is completed but before any other tasks are processed.

Microtasks are designed to execute small, specific tasks that need to be completed before moving to the next phase of the event loop. This helps maintain a responsive and efficient execution flow in asynchronous programming.

## Examples

### setTimeout vs setImmediate

`setTimeout` and `setImmediate` are timer functions designed to run a callback at a future time. The `setTimeout` callback will be executed once the specified delay has elapsed. `setImmediate`, on the other hand, is designed to execute the callback immediately after the event loop's current poll phase completes.

The execution order of these timers can vary depending on the context in which they are called. If both functions are invoked within the main module, the execution order is not deterministic and may vary based on CPU performance or the system's overall workload. However, if they are called within an asynchronous context (e.g., inside an I/O callback), `setImmediate` will consistently be executed before `setTimeout`.

Create an empty `index.ts` file and run the following code several times:

```ts
setImmediate(() => {
  console.log("hello from setImmediate");
});

setTimeout(() => {
  console.log("hello from setTimeout");
}, 0);
```

As expected, the result is not deterministic and we can see `hello from setImmediate` and `hello from setTimeout` in random order. 

However, if we change the context and run the code inside async context, e.g.

```ts
async function asyncTimers() {
  return new Promise((resolve) => {
    fs.readFile(__filename, (error, data) => {
      setTimeout(() => {
        console.log("hello from setTimeout");
      }, 0);
      setImmediate(() => {
        console.log("hello from setImmediate");
      });
      resolve(true);
    });
  });
}

asyncTimers();
```

the result always would be:

```bash
hello from setImmediate
hello from setTimeout
```

This is because both timers are scheduled during the poll phase. Immediately afterward, any `setImmediate` callbacks that were scheduled are executed.

### Microtasks

We can slightly modify the above code to see how microtasks behaves. Let's create a new function with following content and run it:

```ts
async function microtasks() {
  return new Promise((resolve) => {
    fs.readFile(__filename, (error, data) => {
      setTimeout(() => {
        console.log("hello from setTimeout");
      }, 0);
      setImmediate(() => {
        console.log("hello from setImmediate");
      });
      queueMicrotask(() => {
        console.log("hello from queueMicrotask");
      });
      resolve(true);
    });
  });
}

microtasks().then(() => {
  console.log("hello from then");
});
```

As you may expect, the result is:

```bash
hello from queueMicrotask
hello from then
hello from setImmediate
hello from setTimeout
```

Since microtasks are handled using a FIFO (First In, First Out) queue, the `hello from queueMicrotask` callback was added first, followed by the `hello from then` callback. They were then executed in that exact order. The other timers were executed as described above.

# Summary

This is all about the Node.js event loop. I hope you found it interesting and now have a better understanding of how Node.js handles non-blocking I/O. This is the first in a series of articles where we’ll dive deeper into Node.js architecture. Stay tuned for our next piece, which will focus on performance profiling and optimization.

Understanding Node.js - Event loop


import Meta from "../../components/Meta"; 
import Layout from '../../layouts/Layout'

<Meta title="Partition projection with Amazon Athena" tags="AWS, S3, Athena" />

# Partition projection with Amazon Athena

## Introduction

In this post, we will explore different options for querying data using Amazon Athena. We will examine how Athena works, the partitioning mechanism it uses, and its correlation with the AWS Glue Data Catalog.

## Amazon Athena and Glue Data Catalog

Amazon Athena is a serverless, interactive query tool that allows you to analyze data stored directly in Amazon S3 using SQL. It is built on a distributed query engine, which enables fast querying of large datasets without the need to manage infrastructure. Since Athena is a serverless service, you pay only for the queries executed and the amount of data scanned on S3.

Athena can query structured, semi-structured, and unstructured data in formats such as CSV, TSV, JSON, Parquet, Avro, or ORC. The stored data can be compressed using common formats like bzip, Snappy, or [others](https://docs.aws.amazon.com/athena/latest/ug/compression-formats.html). This data is typically organized into partitions, and queries can directly access these partitions for better performance.

Partitioning is a performance optimization technique that helps reduce the time and cost of executing queries. It directs the query engine to look for specific data segments without requiring a full data scan.

Athena also needs to know the schema of the data being queried. This is done by referencing the appropriate table in the AWS Glue Data Catalog.

## Scheme definition and discovery

AWS Glue provides a crawler service that can scan files stored in S3 to infer the schema and create or update corresponding tables in the Data Catalog. This means you can automate the schema definition process instead of doing it manually. I covered this functionality in my previous posts. In this one, I would like to explore different options for defining schemas and partitions manually.

To define a schema and partitions, you need to run a DDL (Data Definition Language) query in the AWS Athena Console. The syntax for the query is as follows:

```sql
CREATE EXTERNAL TABLE IF NOT EXISTS your_table_name (
    column1 type1,
    column2 type2
    ...
) 
PARTITIONED BY (partition_column partition_type)
ROW FORMAT SERDE 'data_serializer/deserializer'
WITH SERDEPROPERTIES (
    'property_name' = 'property_value'
) 
LOCATION 's3://your-data-bucket/data-folder/';
```
 
* `PARTITIONED BY`: Used to define partitions for your dataset.
* `ROW FORMAT SERDE`: Optional and is used when formats other than TEXT are used. The serializer/deserializer must be selected based on the input [data format](https://docs.aws.amazon.com/athena/latest/ug/serde-reference.html).
* `SERDEPROPERTIES`: Optional and can be used to pass additional parameters to the serializer/deserializer library.
* `LOCATION:` Specifies the data location in S3, formatted as s3://your-data-bucket/data-folder/.

For the other possible options, please take a look in the official [documentation](https://docs.aws.amazon.com/athena/latest/ug/create-table.html)

The above DDL will create a schema definition. However, for Athena to query the data, we also need to update the information about available partitions. This can be done either by running an AWS Glue crawler or by executing DML (Data Manipulation Language) queries. The relevant queries are:
* [ALTER TABLE ADD PARTITION](https://docs.aws.amazon.com/athena/latest/ug/alter-table-add-partition.html) - for non-Hive compatible partitions
* [MSCK REPAIR TABLE](https://docs.aws.amazon.com/athena/latest/ug/msck-repair-table.html) - for Hive compatible partitions

Wondering what makes partitions Hive-compatible or not? We'll cover that shortly.
 
Now, it is time to make our hands dirty and run real examples. 

Lets start with preparing example data sets. It will be small, but the size does not matter here as we want to explore capabilities of Athena only.

Create two files with some JSON content in it. In my case the files are:

data1.txt:
```txt
{"color": "red", "message": "The color is red", "value": 100}
{"color": "pink", "message": "The color is pink", "value": 222}
{"color": "yellow", "message": "The color is yellow", "value": 30}
```

data2.txt:
```txt
{"color": "red", "message": "The color is red", "value": 130}
{"color": "pink", "message": "The color is pink", "value": 252}
{"color": "yellow", "message": "The color is yellow", "value": 44}
{"color": "blue", "message": "The color is blue", "value": -34}
```

Please note that each line is a separate JSON record (no separator between records). Compress them with any type supported by AWS. I will use GZIP.

Go to AWS S3 console and create two buckets. One will be used to store our data and the other one by Athena to store query results. Once the buckets are ready, go to AWS Athena console and configure the second bucket to be used according to [manual](https://docs.aws.amazon.com/athena/latest/ug/query-results-specify-location-console.html).

Now, follow the steps:
1. In the data bucket create following structure: `s://your-bucket/1/year=2024/month=10/day=7` and upload `data1.txt.gz` file there.
2. Go to Athena console and create a schema for a data we uploaded:

```sql
CREATE EXTERNAL TABLE data_1 (
  color string,
  message string,
  value integer
)
PARTITIONED BY (year integer, month integer, day integer)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://athena-exampledata-20241007/1';
```

The above has created a schema for our data. The schema sie defined as `data_1` table and is partitioned by year, month and day value.

3. Run the query to list the data we uploaded.

```sql
SELECT * FROM data_1 WHERE year=2024 AND month=10 AND day=7
```

As you might expect, we have no results. This is because we have not yet created or refreshed the partition information. As mentioned earlier, this is crucial for Athena to know where to look for the data. To resolve this, we need to run one of the queries defined above.

Now, let's explain what a Hive-compatible schema is. When we define a key for our data using the format key1=value1/key2=value2, we are following the Hive naming convention. This convention helps organize the data by defining partition names along with their corresponding values as part of the key.

So, in order to fix the missing part we need to run `MSCK REPAIR TABLE data_1;`. When a query completed, we should see the information about updated metadata.

![image](/images/aws-athena-01/msck_repair.png)

And now we can re-run the query. This time we should see the results.

![image](/images/aws-athena-01/athena_result_1.png)

4. In the next step, create a new partition `s://your-bucket/1/year=2024/month=10/day=8` and upload second file there.

5. Run the query to see the results:

```sql
SELECT * FROM data_1 WHERE year=2024 AND month=10 AND day=8
```

Since the data has appeared in a new partition that Athena does not yet recognize, we need to refresh the metadata by running the MSCK REPAIR TABLE data_1; query. This will update the partition information. After repairing the table, re-run the query, and the results should now be available.

As you can see, updating the schema and partitions for Hive-compatible data is quite straightforward. Now, let's explore how to handle this for a non-Hive-compatible structure.

we will start from the scratch. 
1. Go to S3 console and create following structure: `s://your-bucket/2/2024/10/7` and upload `data1.txt.gz` file there.
2. Create the table that reflects the schema of our data:

```sql
CREATE EXTERNAL TABLE data_2 (
  color string,
  message string,
  value integer
)
PARTITIONED BY (year integer, month integer, day integer)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://athena-exampledata-20241007/2';
```

3. Run the query to list the data we uploaded.

```sql
SELECT * FROM data_2 WHERE year=2024 AND month=10 AND day=7
```

As you may expect, there is no results available. The reason is the same as in previous one. We need to update the metadata. But this time, running `MSCK REPAIR TABLE data_2` would not work as the partition has a different structure, which is not Hive compatible. In this case, we need to use the second query:

```sql
ALTER TABLE data_2 ADD PARTITION (year = 2024, month = 10, day = 7) LOCATION 's3://athena-exampledata-20241007/2/2024/10/7'
```

Once the query completed, we can re-run `SELECT` and the results should be available.

![image](/images/aws-athena-01/athena_result_2.png).

4. If you add more data (in new partitions), such as `data2.txt.gz` in `s3://athena-exampledata-20241007/2/2024/10/7`, you will need to re-run the `ALTER TABLE` command to make the new data visible to Athena.

As you can see, handling data partitioning manually and querying with Athena is fairly straightforward but can be error-prone due to the manual process of updating metadata. If your data is only queried using Athena (and you don’t need to expose it to third-party tools or services, making the AWS Glue Data Catalog optional), we will explore an alternative approach to managing partitions automatically.

## Partition projection

An alternative to using AWS Glue Crawler or manually managing partitions is partition projection. This mechanism can significantly speed up query processing and automate partition management.

With partition projection enabled, Athena computes partition values and locations in memory based on the table definition from the Glue Catalog. It projects the partition values and their corresponding locations according to a defined pattern. Since most of these operations are performed in memory, this approach reduces both the time needed to execute queries and the overhead of updating partition metadata.

Partition projection can be enabled on both, Hive compatible and classic partitions. This is being done by defining extra table properties in format:

```txt
 'projection.column_name1.type'='type',
 'projection.column_name1.range'='value_range',
 'projection.column_name2.type'='type',
 'projection.column_name2.range'='value_range'
 'storage.location.template'='s3://your-data-bucket/key1=${column_name1}/key2=${column_name3}'
```

There might be several column used for projection mechanism. Columns can be any of the supported [types](https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html). 


Follow the steps to test how it works in real example.
1. Go to S3 console and create following structure: `s://your-bucket/3/year=2024/month=10/day=7` and upload `data1.txt.gz` file there.
2. Create the table that reflects the schema of our data with projections enabled for year, month and day:

```sql
CREATE EXTERNAL TABLE IF NOT EXISTS data_3 (
  color string,
  message string,
  value integer
)
PARTITIONED BY (year integer, month integer, day integer)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://athena-exampledata-20241007/3/'
TBLPROPERTIES (
  'projection.year.type'='integer',
  'projection.year.range'='2023,2025',
  'projection.month.type'='integer',
  'projection.month.range'='1,12',
  'projection.day.type'='integer',
  'projection.day.range'='1,31',
  'projection.enabled'='true',
  'storage.location.template'='s3://athena-exampledata-20241007/3/year=${year}/month=${month}/day=${day}'
);
```

3. Run the query to list the data we uploaded.

```sql
SELECT * FROM data_3 WHERE year=2024 AND month=10 AND day=7
```

As you can see, this time Athena returned the results without refreshing/updating Glue data catalog.  That works the same for non-Hive compatible schema. The difference would be only location template, which should match the patter we use to store the data (e.g. `s3://athena-exampledata-20241007/4/${year}/${month}/${day}`).

# Summary

This is all about Athena. I hope you found it interesting and discovered some of the useful features that Athena offers for various ad hoc use cases, such as analyzing logs or other stored data.

Partition projection with Amazon Athena

Fraud detection with Kinesis Data Analytics

Athena to read IoT data (part 3)


import Meta from "../../components/Meta"; 

<Meta title="Athena to read IoT data (part 2)" tags="AWS, IoT NodeJS, Typescript" />

# Athena to read IoT data (part 2)  

## Introduction

This is a second post related to processing IoT data levaraging AWS Analytics services. In previous post we have created an AWS infrastructure to consume IoT Data. This time we will build an IoT device simulator that will connect to MQTT gateway exposed in the AWS account and publish meteo data.

## Prerequisites

1. Infrastructure created in previous [post](https://www.cloudsoft.com.pl/posts/aws-iot-athena-01)
2. NodeJS to develop and run a thing simulator
3. Basic TypeScript and NodeJS knowledge

## Project setup

1. Create an empty typescript project by running following commands:
    * *mkdir meteo-station*
    * *cd meteo-station*
    * *npm init -y*
    * *npm install typescript @types/node ts-node --save-dev*
    * *npx tsc --init --rootDir src --outDir build --target esnext*
  
2. The above set of commands should have created an empty typescript project. Our starting point for the app would be `./src/index.ts`. In that case we need to add a script to compile it an run. Edit package.json and add the below command to `scripts` section. 

```ts
"dev": "npx ts-node ./src/index.ts"
```
  
3. Create a start point for our application by running commands:
    * *mkdir src*
    * *touch src/index.ts*

4. Add example content to src/index.ts 

```ts
console.log("Hello world!")
```

5. Run the app with `npm run dev`. 


    If you can see *Hello world* in the console, it means that your `Typescript` project is ready and we can start building a solution on top of it.

6. But before we go to the next section, we need to add a couple of more dependencies (sdk to connect to IoT Core and some cron lib to send the data periodically). 
    * *npm install aws-iot-device-sdk-v2 cron node-cron*
    * *npm install @types/cron --save-dev*

## MQTT Connection

Let's start with creating an utility class that would be able to establish an mqtt connection. Create an empty `mqtt.ts` file in `./src/utils` directory and export a function `createMqttConnection` with following content:

```ts
// ./src/utils/mqtt.ts

import { iot, mqtt } from "aws-iot-device-sdk-v2";

export const createMqttConnection = (
  endpoint: string,
  clientId: string,
  rootCaPath: string,
  certFilePath: string,
  keyFilePath: string
): mqtt.MqttClientConnection => {
  console.log(`Creating a new mTLS connection to ${endpoint} for client=${clientId}`);

  const mtlsBuilder = iot.AwsIotMqttConnectionConfigBuilder.new_mtls_builder_from_path(certFilePath, keyFilePath);

  mtlsBuilder.with_endpoint(endpoint);
  mtlsBuilder.with_certificate_authority_from_path(undefined, rootCaPath);
  mtlsBuilder.with_clean_session(false);
  mtlsBuilder.with_client_id(clientId);

  return new mqtt.MqttClient().new_connection(mtlsBuilder.build());
};
```

The function will create a new connection for a given thing to a requested endpoint using provided set of X.509 details (certificate, private key and a CA root certificate). The thing certificate and a private key were generated for your thing by a terraform script, but when it comes to the CA root certificate, you need to download the one from [amazon trust service](https://www.amazontrust.com/repository/).

The function itself does not require any comments. Basically, we are using AWS SDK functionality to create new MQTT connection. We do not need any topics here yet, it is just a pure connection. Having it open, we will be able to publish messages on a requested topic.

## Sensor data

The next step would be related to defining a data model for our device. Let's assume we want our meteo station to publish data from a few sensors (like temperature, humidity, atmospheric pressure) and with different units (C, F, mb, mmHg, etc).

To meet the requirements, we could define a meteo sensor in a following way:

```ts
// ./src/iot/types.ts

export class Sensor {
  constructor(
    public readonly name: string, // sensor type (e.g. temperature)
    public readonly unit: string, // unit (e.g. F)
    private readonly min: number, // min value to set a range for random 
    private readonly max: number  // max value to set a range for random 
  ) {}

  // will return random value from min-max range
  get value() { 
    return Math.floor(Math.random() * (this.max - this.min + 1) + this.min);
  }
}
```

We will use a sensor values as a part of payload that will be published from our device. The paylod will also contain thing identifier and the timestamp when the values were collected.

```ts
// ./src/iot/types.ts

export interface Payload {
  deviceId: string;
  timestamp: number;
  values: Partial<Sensor>[];
}
```

We defined `values` as a `Partial` type of `Sensor` as we do not need to publish min/max values.

## Thing connection

Now, we will implement `ThingConnection` class. Its responsibility would be to establish mTLS connecion using its own certificate/key and publish messages on its own topic. Let's start with defining the content:

```ts
// ./src/iot/iot-device.ts

import { mqtt } from "aws-iot-device-sdk-v2";
import { createMqttConnection } from "../utils/mqtt";

export class ThingConnection {
  private topic: string; // topic to publish messages on

  private clientConnection: mqtt.MqttClientConnection; // mqtt connection

  constructor(
      certDir: string,          // directory where the key and cert is stored
      endpoint: string,         // mqtt gateway endpoint
      deviceGroup: string,      // device group
      readonly thingId: string  // thing id
  ) {
    this.topic = `devices/${deviceGroup}/${this.thingId}`; // build a topic name
    this.clientConnection = this.setupMqttConnection(certDir, endpoint); // create a connection 
  }

  /**
   * Creates device mqtt connection
   */
  private setupMqttConnection = (certDir: string, endpoint: string) =>
    createMqttConnection(
      endpoint,
      this.thingId,
      `./root-ca.crt`,
      `${certDir}/${this.thingId}/cert.pem`,
      `${certDir}/${this.thingId}/private.key`
    );
}
```

So far we have create an mqtt connection (but did not connect yet) and defined a topic the messages will be published on. Now, we need to add two more functions (both part of `ThingConnection` class). 

One to connect to the endpoint:

```ts
  // ./src/iot/iot-device.ts

  connect = async () => {
    await this.clientConnection.connect();
    console.log(`Thing ${this.thingId} connected succesfully`);
  };  
```

and the other one to publish a paylod.

```ts
  // ./src/iot/iot-device.ts

  publish = async (payload: string) => {
    console.debug(`Thing ${this.thingId} is publishing results: ${payload}`);
    return this.clientConnection.publish(this.topic, payload, mqtt.QoS.AtLeastOnce);
  };
```

We'd like to see some debug information when new payload is being published. 

And that would be all for the `ThingConnection` class. It meets all the requirements we defined above.

## Meteo station

The next thing to define would be a `MeteoStation` class capable of sending periodically meteo sensor values. It would require a thing connection as well as a set of `Sensor` to publish values for. 

```ts
  // ./src/iot/iot-device.ts

export class MeteoStation {
  constructor(
    private readonly connection: ThingConnection, // thing connection
    private readonly sensors: Sensor[] // sensors to publish values for
  ) {}

  start = async () => {
    await this.connection.connect();
    this.createScheduler().start();
  };

  private createScheduler = () =>
    new CronJob("0 */5 * * * *", async () => {
      await this.publishCurrentValues();
    });
}
```

On `start` the `MeteoStation` will establish a connection with mqtt gateway and start a scheduler. The scheduler will call `publishCurrentValues` method every 5 minutes (based on the cron expression) to send to meteo value. 

Now, it's time to implement `publishCurrentValues` method:

```ts
  // ./src/iot/iot-device.ts

  private publishCurrentValues = async () => {
    const payload: Payload = {
      deviceId: this.connection.thingId,
      timestamp: new Date().getTime(),
      values: this.sensors.map((sample) => {
        return {
          name: sample.name,
          unit: sample.unit,
          value: sample.value,
        };
      }),
    };
    return this.connection.publish(JSON.stringify(payload));
  };
```

We create a `Payload` object with all the reqired data. For every sensor we attach the sensor type, unit and the current value (in our case a random one from a range we have defined). The `Payload` is serialized to string and send to thing's topic.

## Testing

Now, it's time to putting all together to see how it works and produce some data for further steps. 

Go back to our app's starting point (`index.ts`) and define a few const:

```ts
  // ./src/index.ts

import { MeteoStation, ThingConnection } from "./iot/iot-device";
import { Sensor } from "./iot/types";

const ENDPOINT = "<< PUT YOUR MQTT ENDPOINT HERE >>"; // IoT/MQTT gateway endpoint
const THING_GROUP = "meteo-station"; // thing group
const CERT_DIR = "./cert"; // location were we put the cerificates and keys

// sensor types
const THERMOMETER = "Temp";
const BAROMETER = "Bar";
const HUMIDITY = "Humidity";

const createMeteoStation = (thingId: string, sensors: Sensor[]) => {
  const connection = new ThingConnection(`${CERT_DIR}/${THING_GROUP}`, ENDPOINT, THING_GROUP, thingId);
  return new MeteoStation(connection, sensors);
};
```

`createMeteoStation` is a `MeteoStation` factory method. It will create new instance of it.

Let's define a `meteo-001` thing:

```ts
  // ./src/index.ts

const thing1 = createMeteoStation("meteo-001", [
  new Sensor(THERMOMETER, "C", 0, 30),
  new Sensor(BAROMETER, "mb", 900, 1100),
  new Sensor(HUMIDITY, "%", 20, 100),
]);
```

and `meteo-002` that will publish data in different units:

```ts
  // ./src/index.ts
const thing2 = createMeteoStation("meteo-002", [
  new Sensor(THERMOMETER, "F", 32, 100),
  new Sensor(BAROMETER, "mmHg", 675, 825),
  new Sensor(HUMIDITY, "%", 20, 100),
]);
```

And the last, but not least - we need to start both things:


```ts
  // ./src/index.ts
thing1.start().then(() => console.log("Thing1 started"));
thing2.start().then(() => console.log("Thing2 started"));
```

Before you start the simulator, make sure you have all static resources prepared. You need to put the files created by terraform script to `./cert` directory as well as `root-ca.crt` (downloaded from [amazon trust service](https://www.amazontrust.com/repository/) - PEM file) in the root directory of the project.

![image](/images/aws-iot-athena-02/cert_dir.png)

If you have all ready, it's time to start the simulator. Run the simulator with `npm run dev` and you should see:

```bash
Creating a new mTLS connection to xxxxx-ats.iot.eu-west-1.amazonaws.com for client=meteo-001
Creating a new mTLS connection to xxxxx-ats.iot.eu-west-1.amazonaws.com for client=meteo-002
Thing meteo-001 connected succesfully
Thing1 started
Thing meteo-002 connected succesfully
Thing2 started
```

and after every 5 minutes, new data published to the topics:

```bash
Thing meteo-001 is publishing results: {"deviceId":"meteo-001","timestamp":1677340800002,"values":[{"name":"Temp","unit":"C","value":2},{"name":"Bar","unit":"mb","value":928},{"name":"Humidity","unit":"%","value":69}]}
Thing meteo-002 is publishing results: {"deviceId":"meteo-002","timestamp":1677340800003,"values":[{"name":"Temp","unit":"F","value":39},{"name":"Bar","unit":"mmHg","value":796},{"name":"Humidity","unit":"%","value":41}]}
```

Also, you can go to AWS console to make sure our rule we created in the previous step is still working and the raw files are stored in S3.

![image](/images/aws-iot-athena-02/s3.png)

And some metrics from IoT console:

![image](/images/aws-iot-athena-02/iot_monitor.png)

Ale the code is available in [github](https://github.com/mmalinowski/aws-iot-athena-02)

## Next steps

A couple of things to highlight

* A payload send by the simulator is relatively small, up to 200B. In real life it would much larger, but still not large enough to be efficiently handled by Athena. For Athena, files with the size less than 128 MB are consider small and not efficient. 
* Simulators send data in different format, which might get further analysis complicated and not efficient

To make our life easier, we will use ETL jobs to transform, merge, compress our data so that Athena can handle them more efficient. Bear in mind that we have only testing/simulated data, so we will not be able to produce 100 or 200 MB files, but we will discover techniques to merge files and controle the number or size the output files.

Athena to read IoT data (part 2)


import Meta from "../../components/Meta"; 

<Meta title="Athena to read IoT data (part 1)" tags="AWS, S3, IoT Core, Terraform" />

# Athena to read IoT data (part 1)  

## Introduction

This is a first post related to processing IoT data levaraging AWS Analytics services. We will build an end to end solution to store and process IoT device data to make it available for Athena and other Analytics services for further processing.  

High level diagram for the solution look following.
  
![image](/images/aws-iot-athena-01/iot_diagram.png)

These are our main requirements:
1. Use MQTT protocol to establish a connection between a thing and AWS IoT Core.  
    * Why MQTT? MQTT is a lightweight publish/subscribe protocol designed for [Internet of Things](https://mqtt.org/). It integrates very well with AWS IoT Core.
2. Secure a connection with [mTLS](https://en.wikipedia.org/wiki/Mutual_authentication).  
    * Why mTLS? mTLS ensures that the traffic is secure and trusted in both directions, providing additional layer of security for your data.
3. Utilize IoT Rules and Actions to store device's raw data in S3.  
    * Why IoT Rules and Actions?  IoT Rules and Actions give your device the ability to interact with other AWS services out of the box, based on the MQTT payload/topic data.
4. Provision infrastructure with IaC solution.  
    * Why IaC? There is a lot of benefits of using IaC. It makes your solution re-usable, portable, it reduces a risk of human error and much more. And in our case, you can easily **destroy** all the resources after you are done with your work and testing.

You do not need to worry about meteo sensors. We will develop a simulator that will replace a real device and help us in providing real time data.

## Prerequisites

1. Terraform installed to provision the infrastracture
2. Access to AWS account to deploy the solution
3. Understanding of AWS and IaC concept. **We will be creating real AWS services, which can cost real money**
4. NodeJS to develop and run a thing simulator
5. Basic TypeScript and Terraforn knowledge

## Terraform setup

It is a good practice to store Terraform state in a [remote](https://developer.hashicorp.com/terraform/language/state) location. In our case, we will store it in the S3 bucket in an account where our service will be provisioned. If you do not have one ready, you need to create it (eiter via AWS CLI or AWS Console). Or you can keep it simple and store the state in local file system. 

In my case, I chose the remote location and bucket name will be `iot-analytics-terraform-state`. Apart from that, we need to define our cloud provider (AWS) and some default tags that will be attached to every resource created by a provider. Finally, the content of `main.tf` can look as below:

```ts
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 4.0"
    }
  }

  backend "s3" {
    bucket  = "iot-analytics-terraform-state"
    key     = "terraform/iot-analytics/terraform.tfstate"
    region  = "eu-west-1"
    encrypt = "true"
  }
}

provider "aws" {
  region = "eu-west-1"

  default_tags {
    tags = {
      project     = "iot-analytics"
      managed-by  = "Terraform"
    }
  }
}
```

Now you can call `terraform init` to initialize the state and the plugins. Once that is completed, you can run `terraform plan` to verify if your initial setup is ready. If the two commands executed without error you can go to the next step.

## Thing module

Since we want to test the solution with at least two different devices, each one with its own certificate and other parameters, let encapsulate that functionality into a separate module. In order to do so, create a *modules/thing* directory in your project's workspace with 3 files: *variables.tf, thing.tf, outputs.tf*. It should look as below:

![image](/images/aws-iot-athena-01/thing_module.png)

We want our module to be configured with 3 input parameters:
* Thing unique identifier
* A group name the thing should belong to
* A location where the certificate for the thing will be stored

Let's start with putting some content to *variables.tf*:

```ts
variable "thing_group" {
  description = "Thing group name"
  type        = string
}

variable "thing_id" {
  description = "Thing unique id"
  type        = string
}

variable "outputs_path" {
  description = "Certificate output path"
  type        = string
}
```

Once we have defined parametes for the module, let's implement a thing resources now. 

First of all, we need to define a thing and assigne it to the group. For that case, we will use module's variables.

```ts
resource "aws_iot_thing" "thing" {
  name = "${var.thing_group}-${var.thing_id}"
}

resource "aws_iot_thing_group_membership" "thing_group_membership" {
  thing_name             = aws_iot_thing.thing.name
  thing_group_name       = var.thing_group
  override_dynamic_group = true
}
```

The group itself will be created outside the module. The above resource *thing_group_membership* just assignees the thing to existing group by its name.

Now, let's work on relevant resources that will let a thing to connect to AWS and publish messages to a topic. We assume every thing will publish to its own topic. We will define the topic name matching the patter */devices/\{group\}/\{thing-id\}*, so in our case it would be */devices/meteo-station/meteo-001*

```ts
resource "aws_iot_certificate" "thing_certificate" {
  active = true
}

resource "aws_iot_thing_principal_attachment" "cert_attachment" {
  principal = aws_iot_certificate.thing_certificate.arn
  thing     = aws_iot_thing.thing.name
}
```

Yes, that's enough to create a certificate for a thing. 

There is still a few more resources needed to let the thing to connect to MQTT gateway and publish messages to a topic. Let's start with defining a policy based on the [example](https://docs.aws.amazon.com/iot/latest/developerguide/pub-sub-policy.html#pub-sub-specific-topic):

```ts
data "aws_arn" "thing" {
  arn = aws_iot_thing.thing.arn
}

data "aws_iam_policy_document" "thing_policy_document" {
  statement {
    sid = "connect"
    actions = [
      "iot:Connect"
    ]
    resources = [
      "arn:aws:iot:${data.aws_arn.thing.region}:${data.aws_arn.thing.account}:client/${var.thing_id}"
    ]
  }
  statement {
    sid = "communicate"
    actions = [
      "iot:Publish",
      "iot:Receive",
    ]
    resources = [
      "arn:aws:iot:${data.aws_arn.thing.region}:${data.aws_arn.thing.account}:topic/devices/${var.thing_group}/${var.thing_id}"
    ]
  }
  statement {
    sid = "subscribe"
    actions = [
      "iot:Subscribe"
    ]
    resources = [
      "arn:aws:iot:${data.aws_arn.thing.region}:${data.aws_arn.thing.account}:topicfilter/${var.thing_group}/${var.thing_id}"
    ]
  }
}
```

As you can see, the policy is quite strict, we do not use any wildcards here. We allow only a thing to publish to its own topic and not more. 

Now, it's time to attach the document to a policy and bind it with the certificate. With that every device (based one the defined thingId) will be able to use a certificate and publish to its own topic.

```ts
resource "aws_iot_policy" "thing_policy" {
  name   = "thing_policy_${aws_iot_thing.thing.name}"
  policy = data.aws_iam_policy_document.thing_policy_document.json
}

resource "aws_iot_policy_attachment" "policy_attachment" {
  policy = aws_iot_policy.thing_policy.name
  target = aws_iot_certificate.thing_certificate.arn
}
```

We are now approaching to the end of the thing module definition. The last but not least is to define the module's output variables. 

For sure we need the module to export the certificate and private key. To achieve that, we will use the output from the *aws_iot_certificate* resource.

```ts
resource "local_file" "thing_certificate_cert" {
  content  = aws_iot_certificate.thing_certificate.certificate_pem
  filename = "${var.outputs_path}/${var.thing_group}/${var.thing_id}/cert.pem"
}

resource "local_file" "thing_certificate_private_key" {
  content  = aws_iot_certificate.thing_certificate.private_key
  filename = "${var.outputs_path}/${var.thing_group}/${var.thing_id}/private.key"
}
```

It will store the data in the provided location and corresponding directory for a thing group/thing id. 

That's all for the thing module. Now, it's time to setup the IoT Core resources to handle and store our sensors data.

## IoT Core and flow setup

We need to return for a while to the architecture diagram to see what actually we need to provision to meet the requirements. And briefly it is:

1. Bucket to store raw data.
2. Things itself - we just created a module, now we need to use it.
3. Rule to store the sensors data in the bucket

Create a *iot.tf* file and start putting some content in it.

Let's begin with the requirement #1. You need to think about the bucket name that will be unique across the region you provision your resources in.

```ts
resource "aws_s3_bucket" "raw_bucket" {
  bucket = "put.your.fancy.name.here"
}

resource "aws_s3_bucket_acl" "raw_bucket_acl" {
  bucket = aws_s3_bucket.raw_bucket.id
  acl    = "private"
}
```

Yes, that's all needed for our testing/demo purpose. In case of production deployment, you would need to add couple of more resources according to AWS best practices (encryption, storage classes, etc.), but this is a demo so keep it simple.

Now, we can go the the step number 2. That would be also quite straightforward with just a few resources. This is possible due to the fact that we already encapsulated all the logic to provision a single thing. Since we need more then one sensor let's start with creating a variable defining ids of the things to be created.

```ts
locals {
  things = ["meteo-001", "meteo-002"]
}
```

Now we can create a thing group and things attached to that group.

```ts
resource "aws_iot_thing_group" "meteo_stations_group" {
  name = "meteo-station"
}

module "thing" {
  for_each     = toset(local.things)
  source       = "./modules/thing"
  thing_group  = aws_iot_thing_group.meteo_stations_group.name
  thing_id     = each.key
  outputs_path = "./cert"
}
```

We are leveraging *for_each* loop to go through all our sensors and create required resources for it. The certificates will be generated and stored in the *./cert/meteo-station/meteo-001* and *./cert/meteo-station/meteo-002* directories.

And now it's time to create an IoT Rule to store the incomming data in S3 bucket we created in step #1. Before we go to that, we need to create a role with *allow s3:PutObject* that will be assumed by IoT rule. So, here is the role definition

```ts
data "aws_iam_policy_document" "iot_assume_role_policy" {
  statement {
    actions = ["sts:AssumeRole"]

    principals {
      type        = "Service"
      identifiers = ["iot.amazonaws.com"]
    }
  }
}

data "aws_iam_policy_document" "iot_write_to_raw_bucket_policy" {
  statement {
    sid       = "AllowPutObject"
    actions   = ["s3:PutObject"]
    resources = ["${aws_s3_bucket.raw_bucket.arn}/*"]
  }
}

resource "aws_iam_role" "iot_raw_bucket_write_role" {
  name               = "iot-raw-bucket-write-role"
  assume_role_policy = data.aws_iam_policy_document.iot_assume_role_policy.json
}

resource "aws_iam_role_policy" "iot_raw_bucket_write_role_allow_s3_put" {
  name   = "allow-s3-raw-bucket-put"
  role   = aws_iam_role.iot_raw_bucket_write_role.name
  policy = data.aws_iam_policy_document.iot_write_to_raw_bucket_policy.json
}
```

To create a rule and action we will use *aws_iot_topic_rule* [rule](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iot_topic_rule). It has variety of settings. Some of them are self explanatory, but another would need a few words to make them clear. 

What we want to achieve is to have a single rule to read input data from all our devices and store in S3 in a way that would be easy to read and analyze later on. There can be sifferent approaches, but we can start with following *key* patter for the data *s3://\{group-name\}/\{thing-id\}/\{date of arriva\l}/\{timestamp\}.json*, for instance */meteo-sensor/meteo-001/2023-02-09/1676658907184.json*. 

Since a thing will publish a data multiple times an hour, we could intorduce additional level for the key, but as it's just an example and not production case, let's keep it simple.

Firstly, we need to think how we want to achieve the first requirement, a single rule to process data from all devices (in our case, all topic). 

IoT Core supports [SQL-like](https://docs.aws.amazon.com/iot/latest/developerguide/iot-create-rule.html) syntax to read data from the topic. If we go further into documentation, we will find out we can use different wildcard patters to define the FROM clause. Since we know that we want to read from *device/meteo-sensor/meteo-001* and *device/meteo-sensor/meteo-002* we can generalize the topic name to 'device/meteo-sensor/+' (*+* defines single level wildcard). So the final sql statement would be *SELECT \* FROM 'devices/meteo-sensor/+'*.

Since we defined the input for our rule already, we should now focus on our second requirement and define the destination. Again, we will use IoT and SQL [capabilities](https://docs.aws.amazon.com/iot/latest/developerguide/iot-sql-functions.html) to define a key for the MQTT paylod. To cover our requirements we will use following functions:
* topic
* parse_time
* timestamp

The first one, *[topic](https://docs.aws.amazon.com/iot/latest/developerguide/iot-sql-functions.html#iot-function-topic)* let us to get the topic name (if no input parameter provided) or the segment (value from topic name splited by '/') if passed on a number. As we defined our rule with *device/meteo-sensor/+* we have 3 segments, *topic(1)* will return *device*, *topic(2)* will give us the group name and *topic(3)* thing id. Knowing that we can start defining our destination key as

```ts
key = "${topic(2)}/${topic(3)}"
```

That would help us nicely partitioning the incoming data per thing group and identifier.

Now we need to extend the key using *[parse_time](https://docs.aws.amazon.com/iot/latest/developerguide/iot-sql-functions.html#iot-sql-function-parse-time)* and *[timestamp](https://docs.aws.amazon.com/iot/latest/developerguide/iot-sql-functions.html#iot-function-timestamp)* functions. The first one will be needed to define date when the sample arrived. We will use it with the combination of the other one. We also need to make sure the sample will not be overrwriten within a day, so each file needs to be stored with unique name. We could achieve that using some uuid generator, but we can also utilize the *timestamp* function. To sum up, we can extend our key to the following formula:

```ts
key = "${topic(2)}/${topic(3)}/${parse_time("yyyy-MM-dd", timestamp())}/${timestamp()}.json"
```

And we can put all together into single resource (please note the double $$ - needed to properly be parsed by TF):

```ts
resource "aws_iot_topic_rule" "raw_bucket_rule" {
  name        = "raw_bucket_rule"
  description = "A rule to store raw payload in S3 bucket"
  enabled     = true
  sql         = "SELECT * FROM 'devices/${aws_iot_thing_group.meteo_stations_group.name}/+'"
  sql_version = "2016-03-23"

  s3 {
    bucket_name = aws_s3_bucket.raw_bucket.id
    role_arn    = aws_iam_role.iot_raw_bucket_write_role.arn
    key         = "$${topic(2)}/$${topic(3)}/$${parse_time(\"yyyy-MM-dd\", timestamp())}/$${timestamp()}.json"
  }
}
```

All the code is available on [github](https://github.com/mmalinowski/aws-iot-athena-01).

We can deploy the changes now and using AWS console verify if all the resources were created and working as expected.

## Testing

We do not have simulator yet, but we can use available AWS services to test some part of our solution. We will not be able to test the MQTT connection itself, but we can try to verify the IoT rule we have created. To do that you need to log-in to AWS console and select IoT Core service. It should redirect you the page where you can check if all the resources were created, see their settings etc. There is also one more interesting sub-page, *MQTT test client*. 

We will use it to publish test message and verify if it was stored in the S3 location according to the formula we created. So, let's go to the test client and select 'Publish to topic' tab. Into topic name enter *devices/meteo-station/meteo-xxx*, keep all other setting with default values and press Publish (couple fo times if you wish).

![image](/images/aws-iot-athena-01/mqtt_client.png)

Now we can go to S3 and the bucket you created to store the raw data. There should be created *meteo-station* directory with another one *meteo-xxx* inside. If you go deeper you will find bunch of json files (basically a single file per publish a message). 

![image](/images/aws-iot-athena-01/s3.png)


![image](/images/aws-iot-athena-01/s3_2.png)

## Next steps

As you can see our rule is working as expected. Now, we need to generate example data to be processed later one. We could use the MQTT test client, but it will not be very convenient and we also would like to test MQTT connection. To make that happen we will build a simulator that pushes the data to our MQTT gateway.

Latest posts