New Features in Amazon DynamoDB - PartiQL, Export to S3, Integration with Kinesis Data Streams
Tags: aws, dynamodb, database, dataengineering
Published: 2020-12-15 06:45:34

<!-- wp:paragraph -->
<p>Every time with AWS re:Invent around, AWS releases many new features over a period of month. In this blog post I will touch on 3 new features which were introduced for Amazon DynamoDB. DynamoDB&nbsp;is a non-relational managed database with single digit millisecond performance at any scale.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>New Features in Amazon DynamoDB - </p>
<!-- /wp:paragraph -->

<!-- wp:list {"ordered":true} -->
<ol><li>PartiQL - SQL-compatible query language for Amazon DynamoDB.</li><li>Export to S3 - Export Amazon DynamoDB table to S3. In this blog I have added a use-case of deserializing the DynamoDB items, writing it to S3 and query using Athena.</li><li>Direct integration of DynamoDB with Kinesis Streams - Stream item-level images of Amazon DynamoDB as a Kinesis Data Stream.</li></ol>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p>To start with, lets look at the new Amazon DynamoDB console. I have 2 DDB tables to play around with for this blog. The Books table is partitioned by Author and sorted by Title. The Movies table is partitioned by year and sorted by title.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4460,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/11/screen-shot-2020-11-23-at-9.18.58-pm.png"><img src="https://aprakash.files.wordpress.com/2020/11/screen-shot-2020-11-23-at-9.18.58-pm.png?w=1024" alt="" class="wp-image-4460"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>Lets jump on to the features:</p>
<!-- /wp:paragraph -->

<!-- wp:list {"ordered":true} -->
<ol><li><strong><span style="text-decoration:underline;">PartiQL</span></strong> - You can use SQL to select, insert, update and delete items from Amazon DynamoDB.  Currently you can use PartiQL for DynamoDB from the Amazon DynamoDB console, the AWS Command Line Interface (AWS CLI), and DynamoDB APIs. For this blog, I am using the AWS console. </li></ol>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p>DynamoDB &gt; PartiQL editor</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;">SELECT SQLs -</span> </p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Simple select SQL</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>SELECT * FROM Books where Author='William Shakespeare'</code></pre>
<!-- /wp:code -->

<!-- wp:group -->
<div class="wp-block-group"><div class="wp-block-group__inner-container"></div></div>
<!-- /wp:group -->

<!-- wp:table -->
<figure class="wp-block-table"><table><tbody><tr><td>Title</td><td>Formats</td><td>Author</td><td>Category</td></tr><tr><td>Hamlet</td><td>{ "Hardcover" : { "S" : "GVJZQ7JK" }, "Paperback" : { "S" : "A4TFUR98" }, "Audiobook" : { "S" : "XWMGHW96" } }</td><td>William Shakespeare</td><td>Drama</td></tr></tbody></table></figure>
<!-- /wp:table -->

<!-- wp:paragraph -->
<p>The following SQL returns the title, hardcover and category using key path -</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>SELECT Title, Formats&#91;'Hardcover'], category FROM Books where Author='John Grisham'</code></pre>
<!-- /wp:code -->

<!-- wp:table -->
<figure class="wp-block-table"><table><tbody><tr><td>Category</td><td>Title</td><td>Hardcover</td></tr><tr><td>Suspense</td><td>The Firm</td><td>Q7QWE3U2</td></tr><tr><td>Suspense</td><td>The Rainmaker</td><td>J4SUKVGU</td></tr><tr><td>Thriller</td><td>The Reckoning</td><td>null</td></tr></tbody></table></figure>
<!-- /wp:table -->

<!-- wp:paragraph -->
<p>The following SQL uses "contains" function which returns TRUE if attribute category has string 'Suspense' - </p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>SELECT Title, Formats&#91;'Audiobook'], Category FROM Books where Author='John Grisham' and contains(Category, 'Suspense')</code></pre>
<!-- /wp:code -->

<!-- wp:table -->
<figure class="wp-block-table"><table><thead><tr><th><strong>year</strong></th><th><strong>title</strong></th><th><strong>release_date</strong></th><th><strong>rank</strong></th></tr></thead><tbody><tr><td>2011</td><td>Sherlock Holmes: A Game of Shadows</td><td>2011-12-10T00:00:00Z</td><td>570</td></tr></tbody></table></figure>
<!-- /wp:table -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;">INSERT SQL - </span></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Insert a single item -</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code><strong>INSERT</strong> INTO Books value {'Title' : 'A time to kill', 'Author' : 'John Grisham', 'Category' : 'Suspense' }

SELECT * FROM Books WHERE Title='A time to kill'</code></pre>
<!-- /wp:code -->

<!-- wp:table -->
<figure class="wp-block-table"><table><thead><tr><th><strong>Author</strong></th><th><strong>Title</strong></th><th><strong>Category</strong></th></tr></thead><tbody><tr><td>John Grisham</td><td>A time to kill</td><td>Suspense</td></tr></tbody></table></figure>
<!-- /wp:table -->

<!-- wp:paragraph -->
<p id="block-fd0bc930-3997-46f2-a705-2af32d6f4501">"INSERT INTO SELECT" SQL fails with ValidationException: Unsupported operation: Inserting  multiple items in a single statement is not supported, use "INSERT INTO  tableName VALUE item" instead</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;">UPDATE SQL -</span> </p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p id="block-7de45d7f-d34a-4bc7-ae12-5051e2a7b9d5">In the previous insert sql, Formats column was null. So lets update the Formats column for the book.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code><strong>UPDATE</strong> Books <strong>SET</strong> Formats={'Hardcover':'J4SUKVGU' ,'Paperback': 'D7YF4FCX'} WHERE Author='John Grisham' and Title='A time to kill'</code></pre>
<!-- /wp:code -->

<!-- wp:table -->
<figure class="wp-block-table"><table><thead><tr><th><strong>Title</strong></th><th><strong>Formats</strong></th><th><strong>Author</strong></th><th><strong>Category</strong></th></tr></thead><tbody><tr><td>A time to kill</td><td>{"Hardcover":{"S":"J4SUKVGU"},"Paperback":{"S":"D7YF4FCX"}}</td><td>John Grisham</td><td>Suspense</td></tr></tbody></table></figure>
<!-- /wp:table -->

<!-- wp:paragraph -->
<p>You can use update sql to remove key from map - </p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>UPDATE Books REMOVE Formats.Paperback WHERE Author='John Grisham' and Title='A time to kill'</code></pre>
<!-- /wp:code -->

<!-- wp:table -->
<figure class="wp-block-table"><table><thead><tr><th><strong>Title</strong></th><th><strong>Formats</strong></th><th><strong>Author</strong></th><th><strong>Category</strong></th></tr></thead><tbody><tr><td>A time to kill</td><td>{"Hardcover":{"S":"J4SUKVGU"}}</td><td>John Grisham</td><td>Suspense</td></tr></tbody></table></figure>
<!-- /wp:table -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;">DELETE SQL - </span></p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>DELETE FROM Books WHERE Author='John Grisham' and Title='A time to kill'</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>For more references - <a href="https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/ql-reference.html" target="_blank" rel="noreferrer noopener">https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/ql-reference.html</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>2. <strong><span style="text-decoration:underline;">EXPORT TO S3</span></strong> - Export full Amazon DynamoDB table to Amazon S3 bucket.  Using DynamoDB table export, you can export data from an Amazon DynamoDB table from any time within your point-in-time recovery window. To do so the table must have point in time recovery (PITR) enabled. If PITR is not enabled for the table, the Export to S3 will report error asking it to be enabled.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>DynamoDB &gt; Exports to S3</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4463,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/11/screen-shot-2020-11-23-at-11.03.16-pm.png"><img src="https://aprakash.files.wordpress.com/2020/11/screen-shot-2020-11-23-at-11.03.16-pm.png?w=792" alt="" class="wp-image-4463"/></a></figure>
<!-- /wp:image -->

<!-- wp:image {"id":4504,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-02-at-11.46.36-am-1.png"><img src="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-02-at-11.46.36-am-1.png?w=819" alt="" class="wp-image-4504"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>After the export is complete it generates a manifest-summary.json file summarizing the export details and a manifest-file.json containing the details of S3 file locations.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Further to extend on Export to S3 feature to&nbsp;perform analytics and complex queries on your data you can create a data pipeline to deserialize the DynamoDB JSON format and query data from AWS Athena.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4571,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/ddb_pipeline_1-6.png"><img src="https://aprakash.files.wordpress.com/2020/12/ddb_pipeline_1-6.png?w=1024" alt="" class="wp-image-4571"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>The workflow contains the following steps:</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Time based CloudWatch event is triggered.</li><li>This event triggers AWS Lambda function.</li><li>DynamoDB Export to S3 is initiated.</li><li>It writes the data in DynamoDB JSON format to S3 RAW bucket. The S3 objects are zipped json files.</li><li>For each .json.gz file in S3 RAW bucket an event notification is set to trigger AWS Lambda.  The detail is show below in S3 Event to trigger AWS Lambda section.</li><li>AWS Lambda reads the S3 object and deserializes the DynamoDB JSON format data using DynamoDB TypeDeserializer class. This class deserializes DynamoDB types to Python types. The deserialized data is written to S3 Content bucket in Parquet format. Code is in Lambda function code section.</li><li>AWS Lambda function updates the table location in AWS Glue catalog.</li><li>Query the data using AWS Athena.</li></ul>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;">S3 Event to trigger AWS Lambda </span></p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4477,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-02-at-11.48.08-am.png"><img src="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-02-at-11.48.08-am.png?w=1024" alt="" class="wp-image-4477"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;">Lambda function code</span> </p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code -->
<pre class="wp-block-syntaxhighlighter-code">import io
import gzip
import json
import boto3
import uuid
import pandas as pd
import awswrangler as wr
from datetime import datetime
from urllib.parse import unquote_plus


def update_glue_table(*, database, table_name, new_location, region_name):
    """ Update AWS Glue non-partitioned table location
    """

    glue = boto3.client("glue", region_name=region_name)

    response = glue.get_table(
        DatabaseName=database, Name=table_name)

    table_input = response["Table"]
    current_location = table_input["StorageDescriptor"]["Location"]

    table_input.pop("UpdateTime", None)
    table_input.pop("CreatedBy", None)
    table_input.pop("CreateTime", None)
    table_input.pop("DatabaseName", None)
    table_input.pop("IsRegisteredWithLakeFormation", None)
    table_input.pop("CatalogId", None)

    table_input["StorageDescriptor"]["Location"] = new_location

    response = glue.update_table(
        DatabaseName=database,
        TableInput=table_input
    )

    return response
    

def lambda_handler(event, context): 
    
    """
    Uses class TypeDeserializer which deserializes DynamoDB types to Python types 
    
    Example - 

    raw data format :
        [{'ACTIVE': {'BOOL': True}, 'params': {'M': {'customer': {'S': 'TEST'}, 'index': {'N': '1'}}}}, ]
    deserialized data format:
        [{'ACTIVE': True, 'params': {'customer': 'TEST', 'index': Decimal('1')}}]

    """
        
    
    s3client = boto3.client('s3')
    athena_db = "default"
    athena_table = "movies"
    
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = unquote_plus(record['s3']['object']['key'])
 
        response = s3client.get_object(Bucket=bucket, Key=key)
        content = response['Body'].read()
        with gzip.GzipFile(fileobj=io.BytesIO(content), mode='rb') as fh:
            data = [json.loads(line) for line in fh]
                    
    all_data = []
    
    boto3.resource('dynamodb')
    deserializer = boto3.dynamodb.types.TypeDeserializer()
    for row in data:
        all_data.append({k: deserializer.deserialize(v) for k,v in row['Item'].items()})

    
    data_df = pd.DataFrame(all_data)
    
    dt = datetime.utcnow().strftime("%Y-%m-%d-%H-%M")
    s3_path="s3://%s/dynamodb/%s/content/dt=%s/" % (bucket, athena_table, dt)
    
    wr.s3.to_parquet(
        df=data_df,
        path=s3_path,
        dataset = True,
    )
    

    update_response = update_glue_table(
        database=athena_db, 
        table_name=athena_table, 
        new_location=s3_path,
        region_name="us-west-2")

    if update_response["ResponseMetadata"]["HTTPStatusCode"] == 200:
        return (f"Successfully updated glue table location - {athena_db}.{athena_table}")
    else:
        return (f"Failed updating glue table location - {athena_db}.{athena_table}")

</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;">Query data from Athena</span></p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted"><code>SELECT title, info.actors, info.rating, info.release_date, year FROM movies where title='Christmas Vacation'</code></pre>
<!-- /wp:preformatted -->

<!-- wp:table -->
<figure id="resultTable" class="wp-block-table"><table><thead><tr><th></th><th>title</th><th>actors</th><th>rating</th><th>release_date</th><th>year</th></tr></thead><tbody><tr><td>1</td><td>Christmas Vacation</td><td>[Chevy Chase, Beverly D'Angelo, Juliette Lewis]</td><td>0.73</td><td>1989-11-30T00:00:00Z</td><td>1989</td></tr></tbody></table></figure>
<!-- /wp:table -->

<!-- wp:paragraph -->
<p>For more references - <a href="https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DataExport.html" target="_blank" rel="noreferrer noopener">https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DataExport.html</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><span style="text-decoration:underline;"><strong>3. DynamoDB integration with Kinesis Stream.</strong></span></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>DynamoDB Streams captures a time-ordered sequence of item-level modifications in a DynamoDB table. Earlier to publish DynamoDB data to S3 in near real time, one of the ways was to enable DynamoDB streams and use AWS Lambda function to forward data to Kinesis Firehose which published the data to S3. To do so you can use a handy package provided by aws labs - <a rel="noreferrer noopener" href="https://github.com/awslabs/lambda-streams-to-firehose" target="_blank">https://github.com/awslabs/lambda-streams-to-firehose</a> .</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4524,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/ddb_pipeline_1-1.png"><img src="https://aprakash.files.wordpress.com/2020/12/ddb_pipeline_1-1.png?w=572" alt="" class="wp-image-4524"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>With several such use cases, AWS has now integrated AWS DynamoDB directly with Amazon Kinesis Stream. Now you can capture item-level changes in your DynamoDB tables as a Kinesis data stream. This can enable you to publish the data to S3 as shown in the below pipeline.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4526,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/ddb_pipeline_2.png"><img src="https://aprakash.files.wordpress.com/2020/12/ddb_pipeline_2.png?w=502" alt="" class="wp-image-4526"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>To setup the pipeline, create a Amazon Kinesis Data Stream. </p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4534,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-12-at-12.51.12-pm.png"><img src="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-12-at-12.51.12-pm.png?w=1024" alt="" class="wp-image-4534"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>After setting up the Amazon Kinesis Data Stream, create Amazon Kinesis Firehose. The source to Kinesis Firehose will be Amazon Kinesis Data Stream and the destination will be S3.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4536,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-12-at-12.54.37-pm.png"><img src="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-12-12-at-12.54.37-pm.png?w=1024" alt="" class="wp-image-4536"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>To proceed move on to enable streaming to Kinesis.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>DynamoDB &gt; Table &gt; Kinesis data stream details &gt; Manage streaming to Kinesis</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4528,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-11-29-at-12.16.49-am.png"><img src="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-11-29-at-12.16.49-am.png?w=1024" alt="" class="wp-image-4528"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p></p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4529,"sizeSlug":"large","linkDestination":"media","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><a href="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-11-29-at-12.43.04-am.png"><img src="https://aprakash.files.wordpress.com/2020/12/screen-shot-2020-11-29-at-12.43.04-am.png?w=1024" alt="" class="wp-image-4529"/></a></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>Once the stream is enabled any item-level change in the table will be captured and written to Amazon S3 bucket. Below is an example of the record which was updated in DynamoDB using PartiQL. The record contains approximate creation date time of the record in DynamoDB streams, along with New and Old image of the record. These records can be parsed using AWS Lambda or AWS Glue and stored in Data Lake for analytical use-cases. </p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">{
   "awsRegion": "us-west-2",
   "dynamodb": {
     "ApproximateCreationDateTime": 1606714671542,
     "Keys": {
       "Author": {
         "S": "James Patterson"
       },
       "Title": {
         "S": "The President Is Missing"
       }
     },
     "NewImage": {
       "Title": {
         "S": "The President Is Missing"
       },
       "Formats": {
         "M": {
           "Hardcover": {
             "S": "JSU4KGVU"
           }
         }
       },
       "Author": {
         "S": "James Patterson"
       },
       "Category": {
         "S": "Mystery"
       }
     },
     "OldImage": {
       "Title": {
         "S": "The President Is Missing"
       },
       "Formats": {
         "M": {
           "Hardcover": {
             "S": "JSU4KGVU"
           },
           "Paperback": {
             "S": "DY7F4CFX"
           }
         }
       },
       "Author": {
         "S": "James Patterson"
       },
       "Category": {
         "S": "Mystery"
       }
     },
     "SizeBytes": 254
   },
   "eventID": "bcaaf073-7e0d-49c2-818e-fe3cf7e5f18a",
   "eventName": "MODIFY",
   "userIdentity": null,
   "recordFormat": "application/json",
   "tableName": "Books",
   "eventSource": "aws:dynamodb"
 }</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>To conclude, in this post I have introduced you to PartiQL which provides SQL-compatible query for Amazon DynamoDB.  We also looked at the Export to S3 feature and how we can create an end to end pipeline to query the Amazon DynamoDB data in Amazon S3 bucket using AWS Athena. And finally we looked at real time analytics use-case where you can enable streams to capture item-level changes in Amazon DynamoDB table as Kinesis data streams.</p>
<!-- /wp:paragraph -->