Querying DynamoDB export data using Athena (with LIST attribute)

Written by mannem on November 10, 2017. Posted in Athena, AWS BIG DATA

Presto supports multiple Array and JSON functions using which you can write queries to get the required results. There is no single way to define a CREATE TABLE and later your QUERIES. You can have CREATE TABLE predefine the schema structure and later your queries can refer the elements you need in the schema. Or you can define your schema as a string and later use functions on your queries to parse this string to get the required results.

https://prestodb.io/docs/current/functions/array.html

https://prestodb.io/docs/current/functions/json.html

In addition, Athena has some examples to use most of the presto functions :

http://docs.aws.amazon.com/athena/latest/ug/querying-arrays.html

http://docs.aws.amazon.com/athena/latest/ug/rows-and-structs.html

http://docs.aws.amazon.com/athena/latest/ug/querying-JSON.html

Now, I was able to use the following CREATE TABLE syntax on the DynamoDB items having List of strings. I was able to flatten out the list of strings using some of the functions like CAST. Please note that this is not the only way to define table and query. My query might be over complexing what your are trying to get. There might be a simpler way as well. So, its really important that you understand the data types that you define and what each query returns and use the correct functions as each function takes a datatype and returns another datatype.

-------------------------------------------------
Data on S3(Exported from DynamoDB table) : 

{"color":{"s":"yellow"},"damageShapes":{"l":[{"s":"991"},{"s":"vvv"},{"s":"qqq"},{"s":"mmmfgh"}]},"time":{"n":"77"},"age":{"s":"22"}}
{"color":{"s":"green"},"damageShapes":{"l":[{"s":"8091"},{"s":"2728"},{"s":"ddf"},{"s":"dfdf"}]},"time":{"n":"34"},"age":{"s":"22"}}

Same data with JSON Pretty print on one Item :

{
  "age": {
    "S": "22"
  },
  "color": {
    "S": "yellow"
  },
  "damageShapes": {
    "L": [
      {
        "S": "991"
      },
      {
        "S": "vvv"
      },
      {
        "S": "qqq"
      },
      {
        "S": "mmmfgh"
      }
    ]
  },
  "time": {
    "N": "77"
  }
}

-------------------------------------------------

# Create table with required data types as per the structure of data. 
# Supported data types of Athena : http://docs.aws.amazon.com/athena/latest/ug/ddl/create-table.html
#  damageShapes can take different structures in schema ex :  damageShapes struct<L:array<struct<S:string>>> or  damageShapes map<string,array<map<string,string>>> or  struct<L:array<map<string,string>>> . 
# Or if may be a simple string so that you need to rely on queries to parse this string. 


CREATE EXTERNAL TABLE IF NOT EXISTS ddb (
  color struct<s:string>,
  time struct<n:string>,
  age struct<s:string>,
  damageShapes struct<L:array<struct<S:string>>>
) 
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
WITH SERDEPROPERTIES ('ignore.malformed.json' = 'true')
LOCATION 's3://bucket.ne/json-ddb/'


-------------------------------------------------
select * from ddb;

color		time	age		damageshapes
{s=yellow}	{n=77}	{s=22}	{l=[{s=991}, {s=vvv}, {s=qqq}, {s=mmmfgh}]}
{s=green}	{n=34}	{s=22}	{l=[{s=8091}, {s=2728}, {s=ddf}, {s=dfdf}]}


# Get the list of strings of damageshapes[list] DynamoDB attribute. 
# From presto perspective damageshapes.l is array<row<K, V>>) data type. i.e array(row(s varchar)) 
select damageshapes.l from ddb;
[{s=991}, {s=vvv}, {s=qqq}, {s=mmmfgh}]
[{s=8091}, {s=2728}, {s=ddf}, {s=dfdf}]


# Get first row of a damageshapes.l array 
select damageshapes.l[1] from ddb;
{s=991}
{s=8091}

# Get value(with key s) of first row of a damageshapes.l array.
select damageshapes.l[1].s from ddb;
991
8091

# Since damageshapes.l is array(row(s varchar))  , you can find any presto functions which can flatten this to your required format which is array<varchar>. If there is not direct function, you might need to do 2 conversions. 

# casting the damageshapes.l [array(row(s varchar))] to JSON automatically strips all keys(S)

select cast(damageshapes.l as json) as values_of_damageshapes , color.s as color from ddb;
values_of_damageshapes						color
[["991"],["vvv"],["qqq"],["mmmfgh"]]	yellow
[["8091"],["2728"],["ddf"],["dfdf"]]	green

# values_of_damageshapes is in JSON format and you can use JSON functions to get the necessary output. 
# https://prestodb.io/docs/current/functions/json.html 

-------------------------------------------------

# Here I am converting values_of_damageshapes(json) to ARRAY(ARRAY(varchar)))) so that I use an array function called flatten which     Flattens an array(array(T)) to an array(T) by concatenating the contained arrays.
# https://prestodb.io/docs/current/functions/array.html

select flatten(cast(values_of_damageshapes as ARRAY(ARRAY(varchar)))) , Actualtime from (select  cast(damageshapes.l as json) as values_of_damageshapes , color.s as Actualcolor , time.n as Actualtime from ddb);
_col0					Actualtime
[991, vvv, qqq, mmmfgh]	77
[8091, 2728, ddf, dfdf]	34

-------------------------------------------------

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

-------------------------------------------------

Data on S3(Exported from DynamoDB table) :

{"color":{"s":"yellow"},"damageShapes":{"l":[{"s":"991"},{"s":"vvv"},{"s":"qqq"},{"s":"mmmfgh"}]},"time":{"n":"77"},"age":{"s":"22"}}

{"color":{"s":"green"},"damageShapes":{"l":[{"s":"8091"},{"s":"2728"},{"s":"ddf"},{"s":"dfdf"}]},"time":{"n":"34"},"age":{"s":"22"}}

Same data with JSON Pretty print on one Item :

{

"age": {

"S": "22"

},

"color": {

"S": "yellow"

},

"damageShapes": {

"L": [

{

"S": "991"

},

{

"S": "vvv"

},

{

"S": "qqq"

},

{

"S": "mmmfgh"

}

]

},

"time": {

"N": "77"

}

-------------------------------------------------

# Create table with required data types as per the structure of data.

# Supported data types of Athena : http://docs.aws.amazon.com/athena/latest/ug/ddl/create-table.html

# damageShapes can take different structures in schema ex : damageShapes struct<L:array<struct<S:string>>> or damageShapes map<string,array<map<string,string>>> or struct<L:array<map<string,string>>> .

# Or if may be a simple string so that you need to rely on queries to parse this string.

CREATE EXTERNAL TABLE IF NOT EXISTS ddb (

color struct<s:string>,

time struct<n:string>,

age struct<s:string>,

damageShapes struct<L:array<struct<S:string>>>

)

ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'

WITH SERDEPROPERTIES ('ignore.malformed.json' = 'true')

LOCATION 's3://bucket.ne/json-ddb/'

-------------------------------------------------

select * from ddb;

color time age damageshapes

{s=yellow} {n=77} {s=22} {l=[{s=991}, {s=vvv}, {s=qqq}, {s=mmmfgh}]}

{s=green} {n=34} {s=22} {l=[{s=8091}, {s=2728}, {s=ddf}, {s=dfdf}]}

# Get the list of strings of damageshapes[list] DynamoDB attribute.

# From presto perspective damageshapes.l is array<row<K, V>>) data type. i.e array(row(s varchar))

select damageshapes.l from ddb;

[{s=991}, {s=vvv}, {s=qqq}, {s=mmmfgh}]

[{s=8091}, {s=2728}, {s=ddf}, {s=dfdf}]

# Get first row of a damageshapes.l array

select damageshapes.l[1] from ddb;

{s=991}

{s=8091}

# Get value(with key s) of first row of a damageshapes.l array.

select damageshapes.l[1].s from ddb;

991

8091

# Since damageshapes.l is array(row(s varchar)) , you can find any presto functions which can flatten this to your required format which is array<varchar>. If there is not direct function, you might need to do 2 conversions.

# casting the damageshapes.l [array(row(s varchar))] to JSON automatically strips all keys(S)

select cast(damageshapes.l as json) as values_of_damageshapes , color.s as color from ddb;

values_of_damageshapes color

[["991"],["vvv"],["qqq"],["mmmfgh"]] yellow

[["8091"],["2728"],["ddf"],["dfdf"]] green

# values_of_damageshapes is in JSON format and you can use JSON functions to get the necessary output.

# https://prestodb.io/docs/current/functions/json.html

-------------------------------------------------

# Here I am converting values_of_damageshapes(json) to ARRAY(ARRAY(varchar)))) so that I use an array function called flatten which Flattens an array(array(T)) to an array(T) by concatenating the contained arrays.

# https://prestodb.io/docs/current/functions/array.html

select flatten(cast(values_of_damageshapes as ARRAY(ARRAY(varchar)))) , Actualtime from (select cast(damageshapes.l as json) as values_of_damageshapes , color.s as Actualcolor , time.n as Actualtime from ddb);

_col0 Actualtime

[991, vvv, qqq, mmmfgh] 77

[8091, 2728, ddf, dfdf] 34

-------------------------------------------------

Continue Reading No Comments

Generate your own CSV/TSV data quickly with urandom + hexdump

Written by mannem on February 16, 2016. Posted in Data-sets

In this article, we will use hexdump + urandom which are included in most linux distributions to quickly generate random data. The values of the first row should be unique and can also be used as hash or index key.

This can be useful, if you wanna upload this data to NoSQL databases like DynamoDB with Primary key as col1 values (or with sort key as second column values)

Here are some one liners to generate data :

# Generate random header-less data in CSV format with million lines 
hexdump -v -e '5/1 "%02x""\n"' /dev/urandom | awk -v OFS=',' '{ print substr($0, 1, 8), substr($0, 9, 2), int(NR * 32768 * rand()) }' | head -n 1000000 > my_csv.csv

cat my_csv.csv | head -3
816b3f48,13,7791
4e0bc8c5,86,19075
54f1623b,5b,83146


# Generate million rows with header in TSV format : 
hexdump -v -e '5/1 "%02x""\n"' /dev/urandom | awk -v OFS='\t' 'NR == 1 { print "COL1", "COL2", "COL3" } { print substr($0, 1, 8), substr($0, 9, 2), int(NR * 32768 * rand()) }' | head -n 1000000 > my_csv.csv

cat my_csv.csv | head -3
foo	bar	baz
7b2c5691	1c	7791
a315f212	bb	19075

# Generate million rows with header in TSV format : 
hexdump -v -e '5/1 "%02x""\n"' /dev/urandom | awk -v OFS=‘\t' '{ print substr($0, 1, 8), substr($0, 9, 2), int(NR * 32768 * rand()) }' | head -n 1000000 > my_csv.csv

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

# Generate random header-less data in CSV format with million lines

hexdump -v -e '5/1 "%02x""\n"' /dev/urandom | awk -v OFS=',' '{ print substr($0, 1, 8), substr($0, 9, 2), int(NR * 32768 * rand()) }' | head -n 1000000 > my_csv.csv

cat my_csv.csv | head -3

816b3f48,13,7791

4e0bc8c5,86,19075

54f1623b,5b,83146

# Generate million rows with header in TSV format :

hexdump -v -e '5/1 "%02x""\n"' /dev/urandom | awk -v OFS='\t' 'NR == 1 { print "COL1", "COL2", "COL3" } { print substr($0, 1, 8), substr($0, 9, 2), int(NR * 32768 * rand()) }' | head -n 1000000 > my_csv.csv

cat my_csv.csv | head -3

foo bar baz

7b2c5691 1c 7791

a315f212 bb 19075

# Generate million rows with header in TSV format :

hexdump -v -e '5/1 "%02x""\n"' /dev/urandom | awk -v OFS=‘\t' '{ print substr($0, 1, 8), substr($0, 9, 2), int(NR * 32768 * rand()) }' | head -n 1000000 > my_csv.csv

Continue Reading No Comments

Search DynamoDB tables using Elasticsearch/Kibana via Logstash plugin

Written by mannem on December 7, 2015. Posted in Dynamo DB, Elasticsearch

The Logstash plugin for Amazon DynamoDB gives you a nearly real-time view of the data in your DynamoDB table. The Logstash plugin for DynamoDB uses DynamoDB Streams to parse and output data as it is added to a DynamoDB table. After you install and activate the Logstash plugin for DynamoDB, it scans the data in the specified table, and then it starts consuming your updates using Streams and then outputs them to Elasticsearch, or a Logstash output of your choice.

Logstash is a data pipeline service that processes data, parses data, and then outputs it to a selected location in a selected format. Elasticsearch is a distributed, full-text search server. For more information about Logstash and Elasticsearch, go to https://www.elastic.co/products/elasticsearch.

Amazon Elasticsearch Service is a managed service that makes it easy to deploy, operate, and scale Elasticsearch in the AWS Cloud. aws.amazon.com/elasticsearch-service/

This article includes an installation guide that is tested on EC2 instance where all the per-requsites are installed and Logstash is configured so that it connects to Amazon ElasticSearch using the input/Output plugins to start indexing records from DynamoDB. Click here to get all the instructions :
https://github.com/mannem/logstash-input-dynamodb

Logstash configuration:

bin/logstash -e 'input { 
    dynamodb{endpoint => "dynamodb.us-west-2.amazonaws.com" 
    streams_endpoint => "streams.dynamodb.us-west-2.amazonaws.com" 
    view_type => "new_and_old_images" 
    aws_access_key_id => "<access_key_id>" 
    aws_secret_access_key => "<secret_key>" 
    table_name => "DynamoDBTableName"} } 
output { 
    elasticsearch 
        { host => "search-imdb-pgjs6t52gf3m2c5xkxpragmaer.us-west-2.es.amazonaws.com" port => "443" protocol => "http" ssl => "true"} 
        stdout { } }'

1

2

3

4

5

6

7

8

9

10

11

bin/logstash -e 'input {

dynamodb{endpoint => "dynamodb.us-west-2.amazonaws.com"

streams_endpoint => "streams.dynamodb.us-west-2.amazonaws.com"

view_type => "new_and_old_images"

aws_access_key_id => "<access_key_id>"

aws_secret_access_key => "<secret_key>"

table_name => "DynamoDBTableName"} }

output {

elasticsearch

{ host => "search-imdb-pgjs6t52gf3m2c5xkxpragmaer.us-west-2.es.amazonaws.com" port => "443" protocol => "http" ssl => "true"}

stdout { } }'

After running a similar command on the shell, Logstash should successfully start and begin indexing the records from your DynamoDB table.

Throughput considerations:

Kibana:

References:
http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Tools.DynamoDBLogstash.html

https://aws.amazon.com/blogs/aws/new-logstash-plugin-search-dynamodb-content-using-elasticsearch/

https://github.com/awslabs/logstash-input-dynamodb

Continue Reading No Comments

Using DynamoDB as session provider with AWS SDK V3

Written by mannem on November 18, 2015. Posted in Dynamo DB

The DynamoDB Session Handler is a custom session handler for PHP that allows developers to use Amazon DynamoDB as a session store. Using DynamoDB for session storage alleviates issues that occur with session handling in a distributed web application by moving sessions off of the local file system and into a shared location. DynamoDB is fast, scalable, easy to setup, and handles replication of your data automatically.

Setting up:

1. Make sure you have PHP >= 5.5.0
2. install AWS PHP SDK(v3) from here http://docs.aws.amazon.com/aws-sdk-php/v3/guide/getting-started/installation.html
3. Configure PHP SDK to use any of the credentials options as mentioned here: http://docs.aws.amazon.com/aws-sdk-php/v3/guide/guide/credentials.html
4. See more details about DyanmoDB provided session handler here: https://docs.aws.amazon.com/aws-sdk-php/v3/guide/service/dynamodb-session-handler.html
5. A DynamoDB table to store session info, with ‘id’ (String) as Hash key.

End to End PHP code with debug turned on:

<?php

require 'vendor/autoload.php';
use Aws\DynamoDb\DynamoDbClient;
use Aws\DynamoDb\SessionHandler;

$sdk = new Aws\Sdk([
    'region'   => 'us-west-2',
    'version'  => 'latest',
    'http'    => [
        'debug' => true
    ]
]);

$dynamodb = $sdk->createDynamoDb();

//var_dump($dynamodb);

$sessionHandler = SessionHandler::fromClient($dynamodb, [
    'table_name' => 'usersession'
]);

$sessionHandler->register();

// Start the session
session_start();

// Alter the session data
$_SESSION['user.name'] = 'jeremy';
$_SESSION['user.role'] = 'admin';

// Close the session (optional, but recommended)
session_write_close();

echo "successfully connected";

?>

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

<?php

require 'vendor/autoload.php';

use Aws\DynamoDb\DynamoDbClient;

use Aws\DynamoDb\SessionHandler;

$sdk = new Aws\Sdk([

'region' => 'us-west-2',

'version' => 'latest',

'http' => [

'debug' => true

]

]);

$dynamodb = $sdk->createDynamoDb();

//var_dump($dynamodb);

$sessionHandler = SessionHandler::fromClient($dynamodb, [

'table_name' => 'usersession'

]);

$sessionHandler->register();

// Start the session

session_start();

// Alter the session data

$_SESSION['user.name'] = 'jeremy';

$_SESSION['user.role'] = 'admin';

// Close the session (optional, but recommended)

session_write_close();

echo "successfully connected";

?>

> php sessionProvider.php
successfully connected

Now, check the DynamoDB table if the session information is stored successfully.

Here is the example structure(DynamoDB JSON format):

{
  "data": {
    "S": "user.name|s:6:\"jeremy\";user.role|s:5:\"admin\";"
  },
  "expires": {
    "N": "1447886783"
  },
  "id": {
    "S": "PHPSESSID_5elf6mc7kmbjec96kn9q07q9i5"
  }
}

1

2

3

4

5

6

7

8

9

10

11

{

"data": {

"S": "user.name|s:6:\"jeremy\";user.role|s:5:\"admin\";"

},

"expires": {

"N": "1447886783"

},

"id": {

"S": "PHPSESSID_5elf6mc7kmbjec96kn9q07q9i5"

}

References :

Continue Reading No Comments

TIME SERIES PROCESSING

Written by mannem on June 27, 2015. Posted in Achitectures

When data arrives as a succession of regular measurements, it is known as time series information. Processing of time series information poses systems scaling challenges that the elasticity of AWS services is uniquely positioned to address.
This elasticity is achieved by using Auto Scaling groups for ingest processing, AWS Data Pipeline for scheduled Amazon Elastic MapReduce jobs, AWS Data Pipeline for intersystem data orchestration, and Amazon Redshift for potentially massive-scale analysis. Key architectural throttle points involving Amazon SQS for sensor message buffering and less frequent AWS Data Pipeline scheduling keep the overall solution costs predictable and controlled.

Continue Reading No Comments

Online Game

Written by mannem on June 27, 2015. Posted in Achitectures

A cost-effective online game architecture featuring automatic capacity adjustment, a highly available and high-speed database, and a data processing cluster for player behavior analysis.