FilesExpand file tree

 main

/

lambda_function.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

38 lines (31 loc) · 1.33 KB

 main

/

lambda_function.py

Copy path

Top

File metadata and controls

• Code
• Blame

38 lines (31 loc) · 1.33 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

import awswrangler as wr

import pandas as pd

import urllib.parse

import os

os_input_s3_cleansed_layer = os.environ['s3_cleansed_layer']

os_input_glue_catalog_db_name = os.environ['glue_catalog_db_name']

os_input_glue_catalog_table_name = os.environ['glue_catalog_table_name']

os_input_write_data_operation = os.environ['write_data_operation']

def lambda_handler(event, context):

# Get the object from the event and show its content type

bucket = event['Records'][0]['s3']['bucket']['name']

key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')

try:

# Creating DF from content

df_raw = wr.s3.read_json('s3://{}/{}'.format(bucket, key))

# Extract required columns:

df_step_1 = pd.json_normalize(df_raw['items'])

# Write to S3

wr_response = wr.s3.to_parquet(

df=df_step_1,

path=os_input_s3_cleansed_layer,

dataset=True,

database=os_input_glue_catalog_db_name,

table=os_input_glue_catalog_table_name,

mode=os_input_write_data_operation

)

return wr_response

except Exception as e:

print(e)

print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket))

raise e