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

<Meta title="Understanding Node.js - Event loop" tags="Node.js, EventLoop, 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

Understanding Node.js - Event loop

Partition projection with Amazon Athena

Fraud detection with Kinesis Data Analytics

Athena to read IoT data (part 3)

Athena to read IoT data (part 2)

Athena to read IoT data (part 1)