I had a chance to work with this NoSQL timeseries database. So, I thought of sharing a simple framework. Use Lambda to ingest data from S3 into an AWS TimeStream, then remotely run queries from your local Pycharm.
Architecture Diagram:
What’s happening on the architecture diagram:
This is the more elaborated version of how to ingest both real-time and batch data concurrently using a single Lambda.
Let’s first examine the tech stack, which consist of:
- AWS S3
- AWS Lambda
- AWS CloudWatch
- AWS TimeStream
- Pycharm
We will now unload the source CSV data into the S3 Raw Source bucket. That will trigger the splitting lambda. In the Lambda Console, we can set the trigger to a specific S3 bucket using EventBridge.
Purpose of Splitter Lambda:
Assume that when it comes to batch data, the file size can vary from MB to GB. The Lambda’s total memory is 10240 MB, and during the transformation and ingestion, it can only safely process files up to 200 MB in size. As a result, this lambda will break large files into chunks of processable files based on the user-provided size.
For instance, the lambda specifies a size limit of 200 MB. If the file size surpasses this limit, the file chunks it and loads it into the S3 Structured Source. If not, it loads the original file into the S3 Structured Source Bucket, using date partitioning to create folders.
Below is the Python script that splits large files into chunks based on the provided size.
import boto3 import json import os import csv import io import datetime import re import awswrangler as wr import numpy as np s3 = boto3.client('s3') def lambda_handler(event, context): ## Attributes extraction from parameters # Destination Bucket name dest_bucket = os.environ['bucket_name'] ## Extraction of Source Path event_result = event['Records'] bucket_name = ''.join([sub['s3']['bucket']['name'] for sub in event_result]) object_name = ''.join([sub['s3']['object']['key'] for sub in event_result]) path = "s3://" + bucket_name + "/" + object_name ## Gets the required file size_reader = s3.get_object(Bucket=bucket_name, Key=object_name) ## Providing the required size to chunk chunk_size = 200 * 1024 * 1024 ## Reading data from Source df = wr.s3.read_csv(path) ## partition for upload split DataFrame to S3 today = datetime.datetime.now() datim = today.strftime("%Y%m%dT%H%M%S") year = today.strftime("%Y") month = today.strftime("%m") day = today.strftime("%d") ## Getting the size value of the file file_size = size_reader['ContentLength'] if file_size > chunk_size: ## Logic for splitting the actual size value into chunks based on chunk_size variable num_chunks = file_size // chunk_size + 1 ## Using numpy's array_split function the dataframe is splitted into chunks of df's chunks = np.array_split(df, num_chunks) ## The chunked datframes is uploaded to S3 under the loop for i, chunk in enumerate(chunks): chunk_data = chunk.to_csv(index=False) output = "raw_source/" + year + "/" + month + "/" + day + "/" + desired_name + "/" + "part" + str(i + 1) + "-" + desired_name + ".csv" s3.put_object(Body=chunk_data, Bucket=dest_bucket, Key=output) else: ## When the file size is smaller than the desired value whole file is uploaded to S3 s3_c = boto3.resource('s3') copy_source = { 'Bucket': bucket_name, 'Key': object_name } output = "raw_source/" + year + "/" + month + "/" + day + "/" + desired_name + "/" + desired_name + ".csv" # print(output) s3_c.meta.client.copy(copy_source, dest_bucket, output) # TODO implement return { 'statusCode': 200 }
The Transformation Lambda will activate after loading the chunked files or the small-sized original file into the S3 Structured Source.
Purpose of Transformation Lambda:
The primary objective of this lambda is to process both real-time and batch data in accordance with the trigger types.
This lambda is triggered every time a file is loaded into S3 Structured Source, and then the developed real-time and batch logics are used based on the file size.
(Note: According to the scenario in which I worked, the real-time data was less than 1 MB and the batch data size was between 50 and 200 MB.)
For instance, the real-time logic will execute if the file size is less than 1 MB, and the batch logic will run if the file size is greater than that.
What does real-time logic do?
We apply the for-loop concept to the transformations and mapping process, appending the transformed data into a list. We then use the Boto3 write_record function to ingest that list into the AWS TimeStream.
response = client.write_records(DatabaseName=DB_NAME, TableName=TBL_NAME, Records=record) What does Batch Logic do?
We apply the same For-loop concept to the transformation and mapping process, append it to a list, and store it in a subfolder under the S3 Structured Source bucket.
Now, an in-built function known as Batch Load Task is available for ingesting bulk data into an AWS TimeStream. This will assist in inserting GBs of data from S3 into TimeStream in a single run.
For example, once all the batch loads are completed for the day and all the transformed data is stored under a sub-folder in the S3 Structured Source bucket, a cron job under the AWS CloudWatch Rule is scheduled to trigger the same lambda by the end of the day to run the create_batch_load_task function, which will create a Batch Load Task in the AWS TimeStream that’ll ingest all the data from the pointed source (AWS S3) to the required Destination TimeStream Table.
The following is a template for creating a Batch Load Task using the Python SDK:
response = client.create_batch_load_task(TargetDatabaseName=DB_NAME, TargetTableName=TBL_NAME, DataModelConfiguration={ 'DataModel': { "TimeColumn": "Timestamp", "TimeUnit": "MILLISECONDS", "DimensionMappings": [{ "SourceColumn": "Dimension_Columnname_from_S3", "DestinationColumn": "Dimension_Columnname_to_TimeStream" } ], "MultiMeasureMappings": { "MultiMeasureAttributeMappings": [ { "SourceColumn": "MeasureValue_Column_from_S3", "TargetMultiMeasureAttributeName": "MeasureValue_Column_to_TimeStream", "MeasureValueType": "DOUBLE" }, { "SourceColumn": "source_time_from_S3", "TargetMultiMeasureAttributeName": "source_time_to_TimeStream", "MeasureValueType": "TIMESTAMP" } ] }, "MeasureNameColumn": "measure_name" } }, DataSourceConfiguration={ "DataSourceS3Configuration": { "BucketName": 'bucket_name', "ObjectKeyPrefix": output }, "DataFormat": "CSV" }, ReportConfiguration={ "ReportS3Configuration": { "BucketName": 'bucket_name', "ObjectKeyPrefix": 'sub_folder', "EncryptionOption": 'SSE_S3' } } ) print(response)
The complete Python script that can handle both real-time and batch data simultaneously based on S3 and CloudWatch Rule triggers is below:
import boto3 import json import awswrangler as wr import os import csv import io import random import string def lambda_handler(event, context): ## Capture Transformation Start time from datetime import datetime time_var = datetime.now() cur = time_var.strftime("%d/%m/%Y %H:%M:%S") print("Process started at", cur) ## Attributes extraction from parameters # TimeStream Details DB_NAME = os.environ['DB_NAME'] TBL_NAME = os.environ['TBL_NAME'] ## Reading data from Source client = boto3.client('timestream-write') ## Identify the lambda purpose try: event_trigger = event['Records'][0]['eventSource'] except: event_trigger = event['source'] if event_trigger == 'aws:s3': print("Entering record transformation and mapping Process") ## Extraction of Source Path event_result = event['Records'] print(event_result) bucket_name = ''.join([sub['s3']['bucket']['name'] for sub in event_result]) object_name = ''.join([sub['s3']['object']['key'] for sub in event_result]) path = "s3://" + bucket_name + "/" + object_name ## Gets the required file size_reader = s3.get_object(Bucket=bucket_name, Key=object_name) ## Providing the required size to chunk desired_size = 10 * 1024 * 1024 ## Getting the size value of the file file_size = size_reader['ContentLength'] df = wr.s3.read_csv(path) ## Records conversion from csv to dictionary json_df = df.to_json(orient="records") source = json.loads(json_df) ## Extraction, transformation and mapping of data from the source import datetime from time import time record = [] for i in range(0, len(source)): Product_list = {key.strip(): value for key, value in source[i].items()} #---Time transformation from String to MilliSeconds for source data---# time_val = Product_list['Timestamp'] timestamp_epoch = datetime.datetime.strptime(time_val, "%m/%d/%Y %H:%M:%S") timestamp_millisecond = timestamp_epoch.timestamp() * 1000 millisecond_varchar = str(timestamp_millisecond) valid_time = millisecond_varchar.replace(".0", "") #---Current Time transformation from String to MilliSeconds for source data---# milliseconds = int(time() * 1000) # print(milliseconds) if file_size < desired_size: ## Process to read and write the source data for near-real_time use-case. key_list = list(Product_list.keys()) dims = key_list[1] block = dims[:2] measure_name = str(block) time_column = { 'Name': 'source_time', 'Value': str(valid_time), 'Type': 'TIMESTAMP' } temp_val = { 'Name': 'measure_value', 'Value': str(Product_list[sens]), 'Type': 'DOUBLE' } dimension = [{'Name': 'DimensionName', 'Value': dims}] overall_value = {'Time': str(milliseconds), 'Dimensions': dimension, 'MeasureName': measure_name, 'MeasureValues': [time_column, temp_val], 'MeasureValueType': 'MULTI'} record = [overall_value] print(record) try: response = client.write_records(DatabaseName=DB_NAME, TableName=TBL_NAME, Records=record) print("WriteRecords Status: [%s]" % response['ResponseMetadata']['HTTPStatusCode']) except client.exceptions.RejectedRecordsException as err: print("RejectedRecords: ", err) for rr in err.response["RejectedRecords"]: print("Rejected Index " + str(rr["RecordIndex"]) + ": " + rr["Reason"]) print("Other records were written successfully. ") except Exception as err: print("Error:", err) if file_size > desired_size: key_list = list(Product_list.keys()) dims = key_list[1] overall_value = {'DimensionName ': dims, 'Timestamp': str(milliseconds), 'measure_value': str(Product_list[sens]), 'source_time': valid_time, 'measure_name': measure_name} record.append(overall_value) if file_size >desired_size: ## Curation Folder Partition Process for transformed values to be inserted, since batch_load_task only accepts files like csv, json etc... today = datetime.datetime.now() datim = today.strftime("%Y%m%dT%H%M%S") year = today.strftime("%Y") month = today.strftime("%m") day = today.strftime("%d") output = "curated_source/" + year + "/" + month + "/" + day + "/" + dims + "/" + dims + "-" + datim + ".csv" stream = io.StringIO() headers = list(record[0].keys()) writer = csv.DictWriter(stream, fieldnames=headers) writer.writeheader() writer.writerows(record) csv_string_object = stream.getvalue() s3 = boto3.resource('s3') s3.Object('bucket_name', output).put(Body=csv_string_object) if event_trigger == 'aws.events': print("Entering record uploading to TimeStream Process") # Process to read and write the source data using batch_load_task function for batch load use-case. today = datetime.datetime.now() datim = today.strftime("%Y%m%dT%H%M%S") year = today.strftime("%Y") month = today.strftime("%m") day = today.strftime("%d") output = "curated_source/" + year + "/" + month + "/" + day + "/" response = client.create_batch_load_task(TargetDatabaseName=DB_NAME, TargetTableName=TBL_NAME, DataModelConfiguration={ 'DataModel': { "TimeColumn": "Timestamp", "TimeUnit": "MILLISECONDS", "DimensionMappings": [{ "SourceColumn": "Dimension_Columnname_from_S3", "DestinationColumn": "Dimension_Columnname_to_TimeStream" } ], "MultiMeasureMappings": { "MultiMeasureAttributeMappings": [ { "SourceColumn": "MeasureValue_Column_from_S3", "TargetMultiMeasureAttributeName": "MeasureValue_Column_to_TimeStream", "MeasureValueType": "DOUBLE" }, { "SourceColumn": "source_time_from_S3", "TargetMultiMeasureAttributeName": "source_time_to_TimeStream", "MeasureValueType": "TIMESTAMP" } ] }, "MeasureNameColumn": "measure_name" } }, DataSourceConfiguration={ "DataSourceS3Configuration": { "BucketName": 'bucket_name', "ObjectKeyPrefix": output }, "DataFormat": "CSV" }, ReportConfiguration={ "ReportS3Configuration": { "BucketName": 'bucket_name', "ObjectKeyPrefix": 'sub_folder', "EncryptionOption": 'SSE_S3' } } ) print(response) ## Capture Transformation End time from datetime import datetime var_time = datetime.now() cur_end = var_time.strftime("%d/%m/%Y %H:%M:%S") print("Process ended at", cur_end) # TODO implement return { 'statusCode': 200, # 'body': json.dumps('Hello from Lambda!') }
Interacting with the AWS TimeStream from Pycharm:
First, integrate the AWS environment with Pycharm to run queries on the TimeStream table. Get the AWS Access_Key_ID and Secret_Access_Key, and if it’s an organization account, get the Session_Token as well.
Download the AWS Tool Kit in PyCharm, install AWS Configure locally, and then open Terminal on a Mac or open cmd on a Windows.
#Use the Following Command to configure the AWS Credentials in it. #In Mac: naveenchandarb@cmd ~ % vi ~/.aws/credentials [default] Access_Key_ID = ************************ Secret_Access_Key = ************************ Session_Token = ************************ #In Windows: naveenchandarb@cmd ~ % aws configure Access_Key_ID = ************************ Secret_Access_Key = ************************ Session_Token = ************************ region = ************************
After configuring AWS credentials on the local machine, go to PyCharm, create a Python file, and use the below script to interact with the AWS TimeStream.
import boto3 timestream_client = boto3.client('timestream-query') paginator = self.timestream_client.get_paginator('query') def execute_query(query): response = paginator.paginate(QueryString=query) # print(response) column_names = [column['Name'] for column in response['ColumnInfo']] query_id = response['QueryId'] next_token = None rows = [] while True: if 'Rows' in response: rows.extend(response['Rows']) if 'NextToken' in response: next_token = response['NextToken'] response = timestream_client.query(QueryString=query_id, NextToken=next_token) else: break return rows, column_names def parse_rows(rows): parsed_rows = [] for row in rows: parsed_row = [] for data in row['Data']: parsed_row.append(data.get('ScalarValue')) parsed_rows.append(parsed_row) return parsed_rows def run_query(query): rows, column_names = execute_query(query) parsed_rows = parse_rows(rows) return parsed_rows, column_names query = """ unload( SELECT Dimension_Name, AVG(measure_value) AS avg_value, bin(time, 15m) as avg_time FROM "timestream_db"."timestream_table" WHERE time >= '2018-01-01 00:00:00' AND time <= '2018-01-10 00:00:00' and measure_name = 'Medium' group by Dimension_Name, bin(time, 15m)) to S3://bucket_name/sub_folders/ with (format = 'CSV', compression = 'NONE') """ result, column_names = run_query(query) table = [column_names] + result # Print the result for row in table: print(row)
I hope this blog will help in performing TimeStream operations using Lambda without any complex coding to achieve remote query executions and the ingestion of both real-time and batch-load data using the Python SDK.
