An AWS Professional Service open source initiative | aws-proserve-opensource@amazon.com

Quick Start

>>> pip install awswrangler

import awswrangler as wr
import pandas as pd
from datetime import datetime

df = pd.DataFrame({"id": [1, 2], "value": ["foo", "boo"]})

# Storing data on Data Lake
wr.s3.to_parquet(
    df=df,
    path="s3://bucket/dataset/",
    dataset=True,
    database="my_db",
    table="my_table"
)

# Retrieving the data directly from Amazon S3
df = wr.s3.read_parquet("s3://bucket/dataset/", dataset=True)

# Retrieving the data from Amazon Athena
df = wr.athena.read_sql_query("SELECT * FROM my_table", database="my_db")

# Get a Redshift connection from Glue Catalog and retrieving data from Redshift Spectrum
con = wr.redshift.connect("my-glue-connection")
df = wr.redshift.read_sql_query("SELECT * FROM external_schema.my_table", con=con)
con.close()

# Amazon Timestream Write
df = pd.DataFrame({
    "time": [datetime.now(), datetime.now()],
    "my_dimension": ["foo", "boo"],
    "measure": [1.0, 1.1],
})
rejected_records = wr.timestream.write(df,
    database="sampleDB",
    table="sampleTable",
    time_col="time",
    measure_col="measure",
    dimensions_cols=["my_dimension"],
)

# Amazon Timestream Query
wr.timestream.query("""
SELECT time, measure_value::double, my_dimension
FROM "sampleDB"."sampleTable" ORDER BY time DESC LIMIT 3
""")

Read The Docs

What is AWS Data Wrangler?

An AWS Professional Service open source python initiative that extends the power of Pandas library to AWS connecting DataFrames and AWS data related services.

Easy integration with Athena, Glue, Redshift, Timestream, QuickSight, Chime, CloudWatchLogs, DynamoDB, EMR, SecretManager, PostgreSQL, MySQL, SQLServer and S3 (Parquet, CSV, JSON and EXCEL).

Built on top of other open-source projects like Pandas, Apache Arrow and Boto3, it offers abstracted functions to execute usual ETL tasks like load/unload data from Data Lakes, Data Warehouses and Databases.

Check our tutorials or the list of functionalities.

Install

AWS Data Wrangler runs with Python 3.6, 3.7, 3.8 and 3.9 and on several platforms (AWS Lambda, AWS Glue Python Shell, EMR, EC2, on-premises, Amazon SageMaker, local, etc).

Some good practices for most of the methods below are:

• Use new and individual Virtual Environments for each project (venv).

• On Notebooks, always restart your kernel after installations.

Note

If you want to use awswrangler for connecting to Microsoft SQL Server, some additional configuration is needed. Please have a look at the corresponding section below.

PyPI (pip)

>>> pip install awswrangler

Conda

>>> conda install -c conda-forge awswrangler

AWS Lambda Layer

Managed Layer

AWS Data Wrangler is available as a Lambda Managed layer in the following regions:

• ap-northeast-1

• ap-southeast-2

• eu-central-1

• eu-west-1

• us-east-1

• us-east-2

• us-west-2

It can be accessed in the AWS Lambda console directly:

Or via its arn: arn:aws:lambda:<region>:336392948345:layer:AWSDataWrangler-Python<version>:<layer-version>. For example: arn:aws:lambda:us-east-1:336392948345:layer:AWSDataWrangler-Python37:1. Both Python 3.7 and 3.8 are supported.

Here is a mapping of layer version to library version:

Lambda layer to Python library versions mapping

Lambda Layer Version	Python Library Version
1	2.12.0

Custom Layer

For AWS regions not in the above list, you can create your own Lambda layer following these instructions:

1 - Go to GitHub’s release section and download the layer zip related to the desired version. Alternatively, you can download the zip from the public artifacts bucket.

2 - Go to the AWS Lambda Panel, open the layer section (left side) and click create layer.

3 - Set name and python version, upload your fresh downloaded zip file and press create to create the layer.

4 - Go to your Lambda and select your new layer!

Serverless Application Repository (SAR)

AWS Data Wrangler layers are also available in the AWS Serverless Application Repository (SAR).

Here is an example of how to create the Lambda layer in your CDK app:

CfnApplication(
    self,
    "wrangler-layer",
    location=CfnApplication.ApplicationLocationProperty(
        application_id="arn:aws:serverlessrepo:us-east-1:336392948345:applications/aws-data-wrangler-layer-py3-8",
        semantic_version="2.12.0",
    ),
)

AWS Glue Python Shell Jobs

1 - Go to GitHub’s release page and download the wheel file (.whl) related to the desired version. Alternatively, you can download the wheel from the public artifacts bucket.

2 - Upload the wheel file to any Amazon S3 location.

3 - Go to your Glue Python Shell job and point to the wheel file on S3 in the Python library path field.

Official Glue Python Shell Reference

AWS Glue PySpark Jobs

Note

AWS Data Wrangler has compiled dependencies (C/C++) so there is only support for Glue PySpark Jobs >= 2.0.

Go to your Glue PySpark job and create a new Job parameters key/value:

• Key: --additional-python-modules

• Value: pyarrow==2,awswrangler

To install a specific version, set the value for above Job parameter as follows:

• Value: cython==0.29.21,pg8000==1.21.0,pyarrow==2,pandas==1.3.0,awswrangler==2.13.0

Note

Pyarrow 3 is not currently supported in Glue PySpark Jobs, which is why a previous installation of pyarrow 2 is required.

Official Glue PySpark Reference

Public Artifacts

Lambda zipped layers and Python wheels are stored in a publicly accessible S3 bucket for all versions.

• Bucket: aws-data-wrangler-public-artifacts

• Prefix: releases/<version>/

  • Lambda layer: awswrangler-layer-<version>-py<py-version>.zip

  • Python wheel: awswrangler-<version>-py3-none-any.whl

For example: s3://aws-data-wrangler-public-artifacts/releases/2.13.0/awswrangler-layer-2.13.0-py3.8.zip

Amazon SageMaker Notebook

Run this command in any Python 3 notebook paragraph and then make sure to restart the kernel before import the awswrangler package.

>>> !pip install awswrangler

Amazon SageMaker Notebook Lifecycle

Open SageMaker console, go to the lifecycle section and use the follow snippet to configure AWS Data Wrangler for all compatible SageMaker kernels (Reference).

#!/bin/bash

set -e

# OVERVIEW
# This script installs a single pip package in all SageMaker conda environments, apart from the JupyterSystemEnv which
# is a system environment reserved for Jupyter.
# Note this may timeout if the package installations in all environments take longer than 5 mins, consider using
# "nohup" to run this as a background process in that case.

sudo -u ec2-user -i <<'EOF'

# PARAMETERS
PACKAGE=awswrangler

# Note that "base" is special environment name, include it there as well.
for env in base /home/ec2-user/anaconda3/envs/*; do
    source /home/ec2-user/anaconda3/bin/activate $(basename "$env")
    if [ $env = 'JupyterSystemEnv' ]; then
        continue
    fi
    nohup pip install --upgrade "$PACKAGE" &
    source /home/ec2-user/anaconda3/bin/deactivate
done
EOF

EMR Cluster

Even not being a distributed library, AWS Data Wrangler could be a good helper to complement Big Data pipelines.

• Configure Python 3 as the default interpreter for PySpark on your cluster configuration [ONLY REQUIRED FOR EMR < 6]

  [
    {
       "Classification": "spark-env",
       "Configurations": [
         {
           "Classification": "export",
           "Properties": {
              "PYSPARK_PYTHON": "/usr/bin/python3"
            }
         }
      ]
    }
  ]
  

• Keep the bootstrap script above on S3 and reference it on your cluster.

  • For EMR Release < 6

    #!/usr/bin/env bash
    set -ex
    
    sudo pip-3.6 install pyarrow==2 awswrangler
    

  • For EMR Release >= 6

    #!/usr/bin/env bash
    set -ex
    
    sudo pip install pyarrow==2 awswrangler
    

Note

Make sure to freeze the Wrangler version in the bootstrap for productive environments (e.g. awswrangler==2.13.0)

Note

Pyarrow 3 is not currently supported in the default EMR image, which is why a previous installation of pyarrow 2 is required.

From Source

>>> git clone https://github.com/awslabs/aws-data-wrangler.git
>>> cd aws-data-wrangler
>>> pip install .

Notes for Microsoft SQL Server

awswrangler is using the pyodbc for interacting with Microsoft SQL Server. For installing this package you need the ODBC header files, which can be installed, for example, with the following commands:

>>> sudo apt install unixodbc-dev
>>> yum install unixODBC-devel

After installing these header files you can either just install pyodbc or awswrangler with the sqlserver extra, which will also install pyodbc:

>>> pip install pyodbc
>>> pip install awswrangler[sqlserver]

Finally you also need the correct ODBC Driver for SQL Server. You can have a look at the documentation from Microsoft to see how they can be installed in your environment.

If you want to connect to Microsoft SQL Server from AWS Lambda, you can build a separate Layer including the needed OBDC drivers and pyobdc.

If you maintain your own environment, you need to take care of the above steps. Because of this limitation usage in combination with Glue jobs is limited and you need to rely on the provided functionality inside Glue itself.

Tutorials

Note

You can also find all Tutorial Notebooks on GitHub.

1 - Introduction

What is AWS Data Wrangler?

An open-source Python package that extends the power of Pandas library to AWS connecting DataFrames and AWS data related services (Amazon Redshift, AWS Glue, Amazon Athena, Amazon Timestream, Amazon EMR, etc).

Built on top of other open-source projects like Pandas, Apache Arrow and Boto3, it offers abstracted functions to execute usual ETL tasks like load/unload data from Data Lakes, Data Warehouses and Databases.

Check our list of functionalities.

How to install?

The Wrangler runs almost anywhere over Python 3.6, 3.7, 3.8 and 3.9, so there are several different ways to install it in the desired enviroment.

• PyPi (pip)

• Conda

• AWS Lambda Layer

• AWS Glue Python Shell Jobs

• AWS Glue PySpark Jobs

• Amazon SageMaker Notebook

• Amazon SageMaker Notebook Lifecycle

• EMR Cluster

• From source

Some good practices for most of the above methods are: - Use new and individual Virtual Environments for each project (venv) - On Notebooks, always restart your kernel after installations.

Let’s Install it!

!pip install awswrangler

Restart your kernel after the installation!

import awswrangler as wr

wr.__version__

'2.0.0'

2 - Sessions

How Wrangler handle Sessions and AWS credentials?

After version 1.0.0 Wrangler absolutely relies on Boto3.Session() to manage AWS credentials and configurations.

Wrangler will not store any kind of state internally. Users are in charge of managing Sessions.

Most Wrangler functions receive the optional boto3_session argument. If None is received, the default boto3 Session will be used.

import awswrangler as wr
import boto3

Using the default Boto3 Session

wr.s3.does_object_exist("s3://noaa-ghcn-pds/fake")

False

Customizing and using the default Boto3 Session

boto3.setup_default_session(region_name="us-east-2")

wr.s3.does_object_exist("s3://noaa-ghcn-pds/fake")

False

Using a new custom Boto3 Session

my_session = boto3.Session(region_name="us-east-2")

wr.s3.does_object_exist("s3://noaa-ghcn-pds/fake", boto3_session=my_session)

False

3 - Amazon S3

Table of Contents

• 1. CSV files

  • 1.1 Writing CSV files

  • 1.2 Reading single CSV file

  • 1.3 Reading multiple CSV files

    • 1.3.1 Reading CSV by list

    • 1.3.2 Reading CSV by prefix

• 2. JSON files

  • 2.1 Writing JSON files

  • 2.2 Reading single JSON file

  • 2.3 Reading multiple JSON files

    • 2.3.1 Reading JSON by list

    • 2.3.2 Reading JSON by prefix

• 3. Parquet files

  • 3.1 Writing Parquet files

  • 3.2 Reading single Parquet file

  • 3.3 Reading multiple Parquet files

    • 3.3.1 Reading Parquet by list

    • 3.3.2 Reading Parquet by prefix

• 4. Fixed-width formatted files (only read)

  • 4.1 Reading single FWF file

  • 4.2 Reading multiple FWF files

    • 4.2.1 Reading FWF by list

    • 4.2.2 Reading FWF by prefix

• 5. Excel files

  • 5.1 Writing Excel file

  • 5.2 Reading Excel file

• 6. Reading with lastModified filter

  • 6.1 Define the Date time with UTC Timezone

  • 6.2 Define the Date time and specify the Timezone

  • 6.3 Read json using the LastModified filters

• 7. Download Objects

  • 7.1 Download object to a file path

  • 7.2 Download object to a file-like object in binary mode

• 8. Upload Objects

  • 8.1 Upload object from a file path

  • 8.2 Upload object from a file-like object in binary mode

• 9. Delete objects

import awswrangler as wr
import pandas as pd
import boto3
import pytz
from datetime import datetime

df1 = pd.DataFrame({
    "id": [1, 2],
    "name": ["foo", "boo"]
})

df2 = pd.DataFrame({
    "id": [3],
    "name": ["bar"]
})

Enter your bucket name:

import getpass
bucket = getpass.getpass()

1. CSV files

1.1 Writing CSV files

path1 = f"s3://{bucket}/csv/file1.csv"
path2 = f"s3://{bucket}/csv/file2.csv"

wr.s3.to_csv(df1, path1, index=False)
wr.s3.to_csv(df2, path2, index=False);

1.2 Reading single CSV file

wr.s3.read_csv([path1])

	id	name
0	1	foo
1	2	boo

1.3 Reading multiple CSV files

1.3.1 Reading CSV by list

wr.s3.read_csv([path1, path2])

	id	name
0	1	foo
1	2	boo
2	3	bar

1.3.2 Reading CSV by prefix

wr.s3.read_csv(f"s3://{bucket}/csv/")

	id	name
0	1	foo
1	2	boo
2	3	bar

2. JSON files

2.1 Writing JSON files

path1 = f"s3://{bucket}/json/file1.json"
path2 = f"s3://{bucket}/json/file2.json"

wr.s3.to_json(df1, path1)
wr.s3.to_json(df2, path2)

['s3://woodadw-test/json/file2.json']

2.2 Reading single JSON file

wr.s3.read_json([path1])

	id	name
0	1	foo
1	2	boo

2.3 Reading multiple JSON files

2.3.1 Reading JSON by list

wr.s3.read_json([path1, path2])

	id	name
0	1	foo
1	2	boo
0	3	bar

2.3.2 Reading JSON by prefix

wr.s3.read_json(f"s3://{bucket}/json/")

	id	name
0	1	foo
1	2	boo
0	3	bar

3. Parquet files

For more complex features releated to Parquet Dataset check the tutorial number 4.

3.1 Writing Parquet files

path1 = f"s3://{bucket}/parquet/file1.parquet"
path2 = f"s3://{bucket}/parquet/file2.parquet"

wr.s3.to_parquet(df1, path1)
wr.s3.to_parquet(df2, path2);

3.2 Reading single Parquet file

wr.s3.read_parquet([path1])

	id	name
0	1	foo
1	2	boo

3.3 Reading multiple Parquet files

3.3.1 Reading Parquet by list

wr.s3.read_parquet([path1, path2])

	id	name
0	1	foo
1	2	boo
2	3	bar

3.3.2 Reading Parquet by prefix

wr.s3.read_parquet(f"s3://{bucket}/parquet/")

	id	name
0	1	foo
1	2	boo
2	3	bar

4. Fixed-width formatted files (only read)

As of today, Pandas doesn’t implement a to_fwf functionality, so let’s manually write two files:

content = "1  Herfelingen 27-12-18\n"\
          "2    Lambusart 14-06-18\n"\
          "3 Spormaggiore 15-04-18"
boto3.client("s3").put_object(Body=content, Bucket=bucket, Key="fwf/file1.txt")

content = "4    Buizingen 05-09-19\n"\
          "5   San Rafael 04-09-19"
boto3.client("s3").put_object(Body=content, Bucket=bucket, Key="fwf/file2.txt")

path1 = f"s3://{bucket}/fwf/file1.txt"
path2 = f"s3://{bucket}/fwf/file2.txt"

4.1 Reading single FWF file

wr.s3.read_fwf([path1], names=["id", "name", "date"])

	id	name	date
0	1	Herfelingen	27-12-18
1	2	Lambusart	14-06-18
2	3	Spormaggiore	15-04-18

4.2 Reading multiple FWF files

4.2.1 Reading FWF by list

wr.s3.read_fwf([path1, path2], names=["id", "name", "date"])

	id	name	date
0	1	Herfelingen	27-12-18
1	2	Lambusart	14-06-18
2	3	Spormaggiore	15-04-18
3	4	Buizingen	05-09-19
4	5	San Rafael	04-09-19

4.2.2 Reading FWF by prefix

wr.s3.read_fwf(f"s3://{bucket}/fwf/", names=["id", "name", "date"])

	id	name	date
0	1	Herfelingen	27-12-18
1	2	Lambusart	14-06-18
2	3	Spormaggiore	15-04-18
3	4	Buizingen	05-09-19
4	5	San Rafael	04-09-19

5. Excel files

5.1 Writing Excel file

path = f"s3://{bucket}/file0.xlsx"

wr.s3.to_excel(df1, path, index=False)

's3://woodadw-test/file0.xlsx'

5.2 Reading Excel file

wr.s3.read_excel(path)

	id	name
0	1	foo
1	2	boo

6. Reading with lastModified filter

Specify the filter by LastModified Date.

The filter needs to be specified as datime with time zone

Internally the path needs to be listed, after that the filter is applied.

The filter compare the s3 content with the variables lastModified_begin and lastModified_end

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html

6.1 Define the Date time with UTC Timezone

begin = datetime.strptime("20-07-31 20:30", "%y-%m-%d %H:%M")
end = datetime.strptime("21-07-31 20:30", "%y-%m-%d %H:%M")

begin_utc = pytz.utc.localize(begin)
end_utc = pytz.utc.localize(end)

6.2 Define the Date time and specify the Timezone

begin = datetime.strptime("20-07-31 20:30", "%y-%m-%d %H:%M")
end = datetime.strptime("21-07-31 20:30", "%y-%m-%d %H:%M")

timezone = pytz.timezone("America/Los_Angeles")

begin_Los_Angeles = timezone.localize(begin)
end_Los_Angeles = timezone.localize(end)

6.3 Read json using the LastModified filters

wr.s3.read_fwf(f"s3://{bucket}/fwf/", names=["id", "name", "date"], last_modified_begin=begin_utc, last_modified_end=end_utc)
wr.s3.read_json(f"s3://{bucket}/json/", last_modified_begin=begin_utc, last_modified_end=end_utc)
wr.s3.read_csv(f"s3://{bucket}/csv/", last_modified_begin=begin_utc, last_modified_end=end_utc)
wr.s3.read_parquet(f"s3://{bucket}/parquet/", last_modified_begin=begin_utc, last_modified_end=end_utc);

7. Download objects

Objects can be downloaded from S3 using either a path to a local file or a file-like object in binary mode.

7.1 Download object to a file path

local_file_dir = getpass.getpass()

import os

path1 = f"s3://{bucket}/csv/file1.csv"
local_file = os.path.join(local_file_dir, "file1.csv")
wr.s3.download(path=path1, local_file=local_file)

pd.read_csv(local_file)

	id	name
0	1	foo
1	2	boo

7.2 Download object to a file-like object in binary mode

path2 = f"s3://{bucket}/csv/file2.csv"
local_file = os.path.join(local_file_dir, "file2.csv")
with open(local_file, mode="wb") as local_f:
    wr.s3.download(path=path2, local_file=local_f)

pd.read_csv(local_file)

	id	name
0	3	bar

8. Upload objects

Objects can be uploaded to S3 using either a path to a local file or a file-like object in binary mode.

8.1 Upload object from a file path

local_file = os.path.join(local_file_dir, "file1.csv")
wr.s3.upload(local_file=local_file, path=path1)

wr.s3.read_csv(path1)

	id	name
0	1	foo
1	2	boo

8.2 Upload object from a file-like object in binary mode

local_file = os.path.join(local_file_dir, "file2.csv")
with open(local_file, "rb") as local_f:
    wr.s3.upload(local_file=local_f, path=path2)

wr.s3.read_csv(path2)

	id	name
0	3	bar

9. Delete objects

wr.s3.delete_objects(f"s3://{bucket}/")

4 - Parquet Datasets

Wrangler has 3 different write modes to store Parquet Datasets on Amazon S3.

• append (Default)

  Only adds new files without any delete.

• overwrite

  Deletes everything in the target directory and then add new files.

• overwrite_partitions (Partition Upsert)

  Only deletes the paths of partitions that should be updated and then writes the new partitions files. It’s like a “partition Upsert”.

from datetime import date
import awswrangler as wr
import pandas as pd

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/dataset/"

 ············

Creating the Dataset

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite"
)

wr.s3.read_parquet(path, dataset=True)

	id	value	date
0	1	foo	2020-01-01
1	2	boo	2020-01-02

Appending

df = pd.DataFrame({
    "id": [3],
    "value": ["bar"],
    "date": [date(2020, 1, 3)]
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="append"
)

wr.s3.read_parquet(path, dataset=True)

	id	value	date
0	3	bar	2020-01-03
1	1	foo	2020-01-01
2	2	boo	2020-01-02

Overwriting

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite"
)

wr.s3.read_parquet(path, dataset=True)

	id	value	date
0	3	bar	2020-01-03

Creating a Partitoned Dataset

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    partition_cols=["date"]
)

wr.s3.read_parquet(path, dataset=True)

	id	value	date
0	1	foo	2020-01-01
1	2	boo	2020-01-02

Upserting partitions (overwrite_partitions)

df = pd.DataFrame({
    "id": [2, 3],
    "value": ["xoo", "bar"],
    "date": [date(2020, 1, 2), date(2020, 1, 3)]
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite_partitions",
    partition_cols=["date"]
)

wr.s3.read_parquet(path, dataset=True)

	id	value	date
0	1	foo	2020-01-01
1	2	xoo	2020-01-02
2	3	bar	2020-01-03

BONUS - Glue/Athena integration

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    database="aws_data_wrangler",
    table="my_table"
)

wr.athena.read_sql_query("SELECT * FROM my_table", database="aws_data_wrangler")

	id	value	date
0	1	foo	2020-01-01
1	2	boo	2020-01-02

5 - Glue Catalog

Wrangler makes heavy use of Glue Catalog to store metadata of tables and connections.

import awswrangler as wr
import pandas as pd

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/data/"

 ············

Creating a Pandas DataFrame

df = pd.DataFrame({
    "id": [1, 2, 3],
    "name": ["shoes", "tshirt", "ball"],
    "price": [50.3, 10.5, 20.0],
    "in_stock": [True, True, False]
})
df

	id	name	price	in_stock
0	1	shoes	50.3	True
1	2	tshirt	10.5	True
2	3	ball	20.0	False

Checking Glue Catalog Databases

databases = wr.catalog.databases()
print(databases)

            Database                                   Description
0  aws_data_wrangler  AWS Data Wrangler Test Arena - Glue Database
1            default                         Default Hive database

Create the database awswrangler_test if not exists

if "awswrangler_test" not in databases.values:
    wr.catalog.create_database("awswrangler_test")
    print(wr.catalog.databases())
else:
    print("Database awswrangler_test already exists")

            Database                                   Description
0  aws_data_wrangler  AWS Data Wrangler Test Arena - Glue Database
1   awswrangler_test
2            default                         Default Hive database

Checking the empty database

wr.catalog.tables(database="awswrangler_test")

	Database	Table	Description	Columns	Partitions

Writing DataFrames to Data Lake (S3 + Parquet + Glue Catalog)

desc = "This is my product table."

param = {
    "source": "Product Web Service",
    "class": "e-commerce"
}

comments = {
    "id": "Unique product ID.",
    "name": "Product name",
    "price": "Product price (dollar)",
    "in_stock": "Is this product availaible in the stock?"
}

res = wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/products/",
    dataset=True,
    database="awswrangler_test",
    table="products",
    mode="overwrite",
    description=desc,
    parameters=param,
    columns_comments=comments
)

Checking Glue Catalog (AWS Console)

Looking Up for the new table!

wr.catalog.tables(name_contains="roduc")

	Database	Table	Description	Columns	Partitions
0	awswrangler_test	products	This is my product table.	id, name, price, in_stock

wr.catalog.tables(name_prefix="pro")

	Database	Table	Description	Columns	Partitions
0	awswrangler_test	products	This is my product table.	id, name, price, in_stock

wr.catalog.tables(name_suffix="ts")

	Database	Table	Description	Columns	Partitions
0	awswrangler_test	products	This is my product table.	id, name, price, in_stock

wr.catalog.tables(search_text="This is my")

	Database	Table	Description	Columns	Partitions
0	awswrangler_test	products	This is my product table.	id, name, price, in_stock

Getting tables details

wr.catalog.table(database="awswrangler_test", table="products")

	Column Name	Type	Partition	Comment
0	id	bigint	False	Unique product ID.
1	name	string	False	Product name
2	price	double	False	Product price (dollar)
3	in_stock	boolean	False	Is this product availaible in the stock?

Cleaning Up the Database

for table in wr.catalog.get_tables(database="awswrangler_test"):
    wr.catalog.delete_table_if_exists(database="awswrangler_test", table=table["Name"])

Delete Database

wr.catalog.delete_database('awswrangler_test')

6 - Amazon Athena

Wrangler has two ways to run queries on Athena and fetch the result as a DataFrame:

• ctas_approach=True (Default)

  Wraps the query with a CTAS and then reads the table data as parquet directly from s3.

  • PROS:

    • Faster for mid and big result sizes.

    • Can handle some level of nested types.

  • CONS:

    • Requires create/delete table permissions on Glue.

    • Does not support timestamp with time zone

    • Does not support columns with repeated names.

    • Does not support columns with undefined data types.

    • A temporary table will be created and then deleted immediately.

    • Does not support custom data_source/catalog_id.

• ctas_approach=False

  Does a regular query on Athena and parse the regular CSV result on s3.

  • PROS:

    • Faster for small result sizes (less latency).

    • Does not require create/delete table permissions on Glue

    • Supports timestamp with time zone.

    • Support custom data_source/catalog_id.

  • CONS:

    • Slower (But stills faster than other libraries that uses the regular Athena API)

    • Does not handle nested types at all.

import awswrangler as wr

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/data/"

 ···········································

Checking/Creating Glue Catalog Databases

if "awswrangler_test" not in wr.catalog.databases().values:
    wr.catalog.create_database("awswrangler_test")

Creating a Parquet Table from the NOAA’s CSV files

Reference

cols = ["id", "dt", "element", "value", "m_flag", "q_flag", "s_flag", "obs_time"]

df = wr.s3.read_csv(
    path="s3://noaa-ghcn-pds/csv/189",
    names=cols,
    parse_dates=["dt", "obs_time"])  # Read 10 files from the 1890 decade (~1GB)

df

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	AGE00135039	1890-01-01	TMAX	160	NaN	NaN	E	NaN
1	AGE00135039	1890-01-01	TMIN	30	NaN	NaN	E	NaN
2	AGE00135039	1890-01-01	PRCP	45	NaN	NaN	E	NaN
3	AGE00147705	1890-01-01	TMAX	140	NaN	NaN	E	NaN
4	AGE00147705	1890-01-01	TMIN	74	NaN	NaN	E	NaN
...	...	...	...	...	...	...	...	...
29240014	UZM00038457	1899-12-31	PRCP	16	NaN	NaN	r	NaN
29240015	UZM00038457	1899-12-31	TAVG	-73	NaN	NaN	r	NaN
29240016	UZM00038618	1899-12-31	TMIN	-76	NaN	NaN	r	NaN
29240017	UZM00038618	1899-12-31	PRCP	0	NaN	NaN	r	NaN
29240018	UZM00038618	1899-12-31	TAVG	-60	NaN	NaN	r	NaN

29240019 rows × 8 columns

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    database="awswrangler_test",
    table="noaa"
);

wr.catalog.table(database="awswrangler_test", table="noaa")

	Column Name	Type	Partition	Comment
0	id	string	False
1	dt	timestamp	False
2	element	string	False
3	value	bigint	False
4	m_flag	string	False
5	q_flag	string	False
6	s_flag	string	False
7	obs_time	string	False

Reading with ctas_approach=False

%%time

wr.athena.read_sql_query("SELECT * FROM noaa", database="awswrangler_test", ctas_approach=False)

CPU times: user 8min 45s, sys: 6.52 s, total: 8min 51s
Wall time: 11min 3s

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	AGE00135039	1890-01-01	TMAX	160	<NA>	<NA>	E	<NA>
1	AGE00135039	1890-01-01	TMIN	30	<NA>	<NA>	E	<NA>
2	AGE00135039	1890-01-01	PRCP	45	<NA>	<NA>	E	<NA>
3	AGE00147705	1890-01-01	TMAX	140	<NA>	<NA>	E	<NA>
4	AGE00147705	1890-01-01	TMIN	74	<NA>	<NA>	E	<NA>
...	...	...	...	...	...	...	...	...
29240014	UZM00038457	1899-12-31	PRCP	16	<NA>	<NA>	r	<NA>
29240015	UZM00038457	1899-12-31	TAVG	-73	<NA>	<NA>	r	<NA>
29240016	UZM00038618	1899-12-31	TMIN	-76	<NA>	<NA>	r	<NA>
29240017	UZM00038618	1899-12-31	PRCP	0	<NA>	<NA>	r	<NA>
29240018	UZM00038618	1899-12-31	TAVG	-60	<NA>	<NA>	r	<NA>

29240019 rows × 8 columns

Default with ctas_approach=True - 13x faster (default)

%%time

wr.athena.read_sql_query("SELECT * FROM noaa", database="awswrangler_test")

CPU times: user 28 s, sys: 6.07 s, total: 34.1 s
Wall time: 50.5 s

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	ASN00017088	1890-06-11	PRCP	0	<NA>	<NA>	a	<NA>
1	ASN00017087	1890-06-11	PRCP	0	<NA>	<NA>	a	<NA>
2	ASN00017089	1890-06-11	PRCP	71	<NA>	<NA>	a	<NA>
3	ASN00017095	1890-06-11	PRCP	0	<NA>	<NA>	a	<NA>
4	ASN00017094	1890-06-11	PRCP	0	<NA>	<NA>	a	<NA>
...	...	...	...	...	...	...	...	...
29240014	USC00461260	1899-12-31	SNOW	0	<NA>	<NA>	6	<NA>
29240015	USC00461515	1899-12-31	TMAX	-89	<NA>	<NA>	6	<NA>
29240016	USC00461515	1899-12-31	TMIN	-189	<NA>	<NA>	6	<NA>
29240017	USC00461515	1899-12-31	PRCP	0	<NA>	<NA>	6	<NA>
29240018	USC00461515	1899-12-31	SNOW	0	<NA>	<NA>	6	<NA>

29240019 rows × 8 columns

Using categories to speed up and save memory - 24x faster

%%time

wr.athena.read_sql_query("SELECT * FROM noaa", database="awswrangler_test", categories=["id", "dt", "element", "value", "m_flag", "q_flag", "s_flag", "obs_time"])

CPU times: user 6.89 s, sys: 2.27 s, total: 9.16 s
Wall time: 27.3 s

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	GME00102348	1890-08-03	TMAX	172	NaN	NaN	E	NaN
1	GME00102348	1890-08-03	TMIN	117	NaN	NaN	E	NaN
2	GME00102348	1890-08-03	PRCP	63	NaN	NaN	E	NaN
3	GME00102348	1890-08-03	SNWD	0	NaN	NaN	E	NaN
4	GME00121126	1890-08-03	PRCP	32	NaN	NaN	E	NaN
...	...	...	...	...	...	...	...	...
29240014	USC00461260	1899-12-31	SNOW	0	NaN	NaN	6	NaN
29240015	USC00461515	1899-12-31	TMAX	-89	NaN	NaN	6	NaN
29240016	USC00461515	1899-12-31	TMIN	-189	NaN	NaN	6	NaN
29240017	USC00461515	1899-12-31	PRCP	0	NaN	NaN	6	NaN
29240018	USC00461515	1899-12-31	SNOW	0	NaN	NaN	6	NaN

29240019 rows × 8 columns

Batching (Good for restricted memory environments)

%%time

dfs = wr.athena.read_sql_query(
    "SELECT * FROM noaa",
    database="awswrangler_test",
    chunksize=True  # Chunksize calculated automatically for ctas_approach.
)

for df in dfs:  # Batching
    print(len(df.index))

1024
8086528
1024
1024
1024
1024
1024
15360
1024
10090496
2153472
8886995
CPU times: user 22.7 s, sys: 5.41 s, total: 28.1 s
Wall time: 48 s

%%time

dfs = wr.athena.read_sql_query(
    "SELECT * FROM noaa",
    database="awswrangler_test",
    chunksize=100_000_000
)

for df in dfs:  # Batching
    print(len(df.index))

29240019
CPU times: user 34.8 s, sys: 8.54 s, total: 43.4 s
Wall time: 1min 1s

Cleaning Up S3

wr.s3.delete_objects(path)

Delete table

wr.catalog.delete_table_if_exists(database="awswrangler_test", table="noaa");

Delete Database

wr.catalog.delete_database('awswrangler_test')

7 - Redshift, MySQL, PostgreSQL and SQL Server

Wrangler’s Redshift, MySQL and PostgreSQL have two basic function in common that tries to follow the Pandas conventions, but add more data type consistency.

• wr.redshift.to_sql()

• wr.redshift.read_sql_query()

• wr.mysql.to_sql()

• wr.mysql.read_sql_query()

• wr.postgresql.to_sql()

• wr.postgresql.read_sql_query()

• wr.sqlserver.to_sql()

• wr.sqlserver.read_sql_query()

import awswrangler as wr
import pandas as pd

df = pd.DataFrame({
    "id": [1, 2],
    "name": ["foo", "boo"]
})

Connect using the Glue Catalog Connections

• wr.redshift.connect()

• wr.mysql.connect()

• wr.postgresql.connect()

• wr.sqlserver.connect()

con_redshift = wr.redshift.connect("aws-data-wrangler-redshift")
con_mysql = wr.mysql.connect("aws-data-wrangler-mysql")
con_postgresql = wr.postgresql.connect("aws-data-wrangler-postgresql")
con_sqlserver = wr.sqlserver.connect("aws-data-wrangler-sqlserver")

Raw SQL queries (No Pandas)

with con_redshift.cursor() as cursor:
    for row in cursor.execute("SELECT 1"):
        print(row)

[1]

Loading data to Database

wr.redshift.to_sql(df, con_redshift, schema="public", table="tutorial", mode="overwrite")
wr.mysql.to_sql(df, con_mysql, schema="test", table="tutorial", mode="overwrite")
wr.postgresql.to_sql(df, con_postgresql, schema="public", table="tutorial", mode="overwrite")
wr.sqlserver.to_sql(df, con_sqlserver, schema="dbo", table="tutorial", mode="overwrite")

Unloading data from Database

wr.redshift.read_sql_query("SELECT * FROM public.tutorial", con=con_redshift)
wr.mysql.read_sql_query("SELECT * FROM test.tutorial", con=con_mysql)
wr.postgresql.read_sql_query("SELECT * FROM public.tutorial", con=con_postgresql)
wr.sqlserver.read_sql_query("SELECT * FROM dbo.tutorial", con=con_sqlserver)

	id	name
0	1	foo
1	2	boo

con_redshift.close()
con_mysql.close()
con_postgresql.close()
con_sqlserver.close()

8 - Redshift - COPY & UNLOAD

Amazon Redshift has two SQL command that help to load and unload large amount of data staging it on Amazon S3:

1 - COPY

2 - UNLOAD

Let’s take a look and how Wrangler can use it.

import awswrangler as wr

con = wr.redshift.connect("aws-data-wrangler-redshift")

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/stage/"

 ···········································

Enter your IAM ROLE ARN:

iam_role = getpass.getpass()

 ····················································································

Creating a Dataframe from the NOAA’s CSV files

Reference

cols = ["id", "dt", "element", "value", "m_flag", "q_flag", "s_flag", "obs_time"]

df = wr.s3.read_csv(
    path="s3://noaa-ghcn-pds/csv/1897.csv",
    names=cols,
    parse_dates=["dt", "obs_time"])  # ~127MB, ~4MM rows

df

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	AG000060590	1897-01-01	TMAX	170	NaN	NaN	E	NaN
1	AG000060590	1897-01-01	TMIN	-14	NaN	NaN	E	NaN
2	AG000060590	1897-01-01	PRCP	0	NaN	NaN	E	NaN
3	AGE00135039	1897-01-01	TMAX	140	NaN	NaN	E	NaN
4	AGE00135039	1897-01-01	TMIN	40	NaN	NaN	E	NaN
...	...	...	...	...	...	...	...	...
3923594	UZM00038457	1897-12-31	TMIN	-145	NaN	NaN	r	NaN
3923595	UZM00038457	1897-12-31	PRCP	4	NaN	NaN	r	NaN
3923596	UZM00038457	1897-12-31	TAVG	-95	NaN	NaN	r	NaN
3923597	UZM00038618	1897-12-31	PRCP	66	NaN	NaN	r	NaN
3923598	UZM00038618	1897-12-31	TAVG	-45	NaN	NaN	r	NaN

3923599 rows × 8 columns

Load and Unload with COPY and UNLOAD commands

Note: Please use a empty S3 path for the COPY command.

%%time

wr.redshift.copy(
    df=df,
    path=path,
    con=con,
    schema="public",
    table="commands",
    mode="overwrite",
    iam_role=iam_role,
)

CPU times: user 2.78 s, sys: 293 ms, total: 3.08 s
Wall time: 20.7 s

%%time

wr.redshift.unload(
    sql="SELECT * FROM public.commands",
    con=con,
    iam_role=iam_role,
    path=path,
    keep_files=True,
)

CPU times: user 10 s, sys: 1.14 s, total: 11.2 s
Wall time: 27.5 s

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	AG000060590	1897-01-01	TMAX	170	<NA>	<NA>	E	<NA>
1	AG000060590	1897-01-01	PRCP	0	<NA>	<NA>	E	<NA>
2	AGE00135039	1897-01-01	TMIN	40	<NA>	<NA>	E	<NA>
3	AGE00147705	1897-01-01	TMAX	164	<NA>	<NA>	E	<NA>
4	AGE00147705	1897-01-01	PRCP	0	<NA>	<NA>	E	<NA>
...	...	...	...	...	...	...	...	...
3923594	USW00094967	1897-12-31	TMAX	-144	<NA>	<NA>	6	<NA>
3923595	USW00094967	1897-12-31	PRCP	0	P	<NA>	6	<NA>
3923596	UZM00038457	1897-12-31	TMAX	-49	<NA>	<NA>	r	<NA>
3923597	UZM00038457	1897-12-31	PRCP	4	<NA>	<NA>	r	<NA>
3923598	UZM00038618	1897-12-31	PRCP	66	<NA>	<NA>	r	<NA>

7847198 rows × 8 columns

con.close()

9 - Redshift - Append, Overwrite and Upsert

Wrangler’s copy/to_sql function has three different mode options for Redshift.

1 - append

2 - overwrite

3 - upsert

import awswrangler as wr
import pandas as pd
from datetime import date

con = wr.redshift.connect("aws-data-wrangler-redshift")

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/stage/"

 ···········································

Enter your IAM ROLE ARN:

iam_role = getpass.getpass()

 ····················································································

Creating the table (Overwriting if it exists)

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.redshift.copy(
    df=df,
    path=path,
    con=con,
    schema="public",
    table="my_table",
    mode="overwrite",
    iam_role=iam_role,
    primary_keys=["id"]
)

wr.redshift.read_sql_table(table="my_table", schema="public", con=con)

	id	value	date
0	2	boo	2020-01-02
1	1	foo	2020-01-01

Appending

df = pd.DataFrame({
    "id": [3],
    "value": ["bar"],
    "date": [date(2020, 1, 3)]
})

wr.redshift.copy(
    df=df,
    path=path,
    con=con,
    schema="public",
    table="my_table",
    mode="append",
    iam_role=iam_role,
    primary_keys=["id"]
)

wr.redshift.read_sql_table(table="my_table", schema="public", con=con)

	id	value	date
0	1	foo	2020-01-01
1	2	boo	2020-01-02
2	3	bar	2020-01-03

Upserting

df = pd.DataFrame({
    "id": [2, 3],
    "value": ["xoo", "bar"],
    "date": [date(2020, 1, 2), date(2020, 1, 3)]
})

wr.redshift.copy(
    df=df,
    path=path,
    con=con,
    schema="public",
    table="my_table",
    mode="upsert",
    iam_role=iam_role,
    primary_keys=["id"]
)

wr.redshift.read_sql_table(table="my_table", schema="public", con=con)

	id	value	date
0	1	foo	2020-01-01
1	2	xoo	2020-01-02
2	3	bar	2020-01-03

Cleaning Up

with con.cursor() as cursor:
    cursor.execute("DROP TABLE public.my_table")
con.close()

10 - Parquet Crawler

Wrangler can extract only the metadata from Parquet files and Partitions and then add it to the Glue Catalog.

import awswrangler as wr

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/data/"

 ············

Creating a Parquet Table from the NOAA’s CSV files

Reference

cols = ["id", "dt", "element", "value", "m_flag", "q_flag", "s_flag", "obs_time"]

df = wr.s3.read_csv(
    path="s3://noaa-ghcn-pds/csv/189",
    names=cols,
    parse_dates=["dt", "obs_time"])  # Read 10 files from the 1890 decade (~1GB)

df

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	AGE00135039	1890-01-01	TMAX	160	NaN	NaN	E	NaN
1	AGE00135039	1890-01-01	TMIN	30	NaN	NaN	E	NaN
2	AGE00135039	1890-01-01	PRCP	45	NaN	NaN	E	NaN
3	AGE00147705	1890-01-01	TMAX	140	NaN	NaN	E	NaN
4	AGE00147705	1890-01-01	TMIN	74	NaN	NaN	E	NaN
...	...	...	...	...	...	...	...	...
29249753	UZM00038457	1899-12-31	PRCP	16	NaN	NaN	r	NaN
29249754	UZM00038457	1899-12-31	TAVG	-73	NaN	NaN	r	NaN
29249755	UZM00038618	1899-12-31	TMIN	-76	NaN	NaN	r	NaN
29249756	UZM00038618	1899-12-31	PRCP	0	NaN	NaN	r	NaN
29249757	UZM00038618	1899-12-31	TAVG	-60	NaN	NaN	r	NaN

29249758 rows × 8 columns

df["year"] = df["dt"].dt.year

df.head(3)

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time	year
0	AGE00135039	1890-01-01	TMAX	160	NaN	NaN	E	NaN	1890
1	AGE00135039	1890-01-01	TMIN	30	NaN	NaN	E	NaN	1890
2	AGE00135039	1890-01-01	PRCP	45	NaN	NaN	E	NaN	1890

res = wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    partition_cols=["year"],
)

[ x.split("data/", 1)[1] for x in wr.s3.list_objects(path)]

['year=1890/06a519afcf8e48c9b08c8908f30adcfe.snappy.parquet',
 'year=1891/5a99c28dbef54008bfc770c946099e02.snappy.parquet',
 'year=1892/9b1ea5d1cfad40f78c920f93540ca8ec.snappy.parquet',
 'year=1893/92259b49c134401eaf772506ee802af6.snappy.parquet',
 'year=1894/c734469ffff944f69dc277c630064a16.snappy.parquet',
 'year=1895/cf7ccde86aaf4d138f86c379c0817aa6.snappy.parquet',
 'year=1896/ce02f4c2c554438786b766b33db451b6.snappy.parquet',
 'year=1897/e04de04ad3c444deadcc9c410ab97ca1.snappy.parquet',
 'year=1898/acb0e02878f04b56a6200f4b5a97be0e.snappy.parquet',
 'year=1899/a269bdbb0f6a48faac55f3bcfef7df7a.snappy.parquet']

Crawling!

%%time

res = wr.s3.store_parquet_metadata(
    path=path,
    database="awswrangler_test",
    table="crawler",
    dataset=True,
    mode="overwrite",
    dtype={"year": "int"}
)

CPU times: user 1.81 s, sys: 528 ms, total: 2.33 s
Wall time: 3.21 s

Checking

wr.catalog.table(database="awswrangler_test", table="crawler")

	Column Name	Type	Partition	Comment
0	id	string	False
1	dt	timestamp	False
2	element	string	False
3	value	bigint	False
4	m_flag	string	False
5	q_flag	string	False
6	s_flag	string	False
7	obs_time	string	False
8	year	int	True

%%time

wr.athena.read_sql_query("SELECT * FROM crawler WHERE year=1890", database="awswrangler_test")

CPU times: user 3.52 s, sys: 811 ms, total: 4.33 s
Wall time: 9.6 s

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time	year
0	USC00195145	1890-01-01	TMIN	-28	<NA>	<NA>	6	<NA>	1890
1	USC00196770	1890-01-01	PRCP	0	P	<NA>	6	<NA>	1890
2	USC00196770	1890-01-01	SNOW	0	<NA>	<NA>	6	<NA>	1890
3	USC00196915	1890-01-01	PRCP	0	P	<NA>	6	<NA>	1890
4	USC00196915	1890-01-01	SNOW	0	<NA>	<NA>	6	<NA>	1890
...	...	...	...	...	...	...	...	...	...
6139	ASN00022006	1890-12-03	PRCP	0	<NA>	<NA>	a	<NA>	1890
6140	ASN00022007	1890-12-03	PRCP	0	<NA>	<NA>	a	<NA>	1890
6141	ASN00022008	1890-12-03	PRCP	0	<NA>	<NA>	a	<NA>	1890
6142	ASN00022009	1890-12-03	PRCP	0	<NA>	<NA>	a	<NA>	1890
6143	ASN00022011	1890-12-03	PRCP	0	<NA>	<NA>	a	<NA>	1890

1276246 rows × 9 columns

Cleaning Up S3

wr.s3.delete_objects(path)

Cleaning Up the Database

for table in wr.catalog.get_tables(database="awswrangler_test"):
    wr.catalog.delete_table_if_exists(database="awswrangler_test", table=table["Name"])

11 - CSV Datasets

Wrangler has 3 different write modes to store CSV Datasets on Amazon S3.

• append (Default)

  Only adds new files without any delete.

• overwrite

  Deletes everything in the target directory and then add new files.

• overwrite_partitions (Partition Upsert)

  Only deletes the paths of partitions that should be updated and then writes the new partitions files. It’s like a “partition Upsert”.

from datetime import date
import awswrangler as wr
import pandas as pd

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/dataset/"

 ············

Checking/Creating Glue Catalog Databases

if "awswrangler_test" not in wr.catalog.databases().values:
    wr.catalog.create_database("awswrangler_test")

Creating the Dataset

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.s3.to_csv(
    df=df,
    path=path,
    index=False,
    dataset=True,
    mode="overwrite",
    database="awswrangler_test",
    table="csv_dataset"
)

wr.athena.read_sql_table(database="awswrangler_test", table="csv_dataset")

	id	value	date
0	1	foo	2020-01-01
1	2	boo	2020-01-02

Appending

df = pd.DataFrame({
    "id": [3],
    "value": ["bar"],
    "date": [date(2020, 1, 3)]
})

wr.s3.to_csv(
    df=df,
    path=path,
    index=False,
    dataset=True,
    mode="append",
    database="awswrangler_test",
    table="csv_dataset"
)

wr.athena.read_sql_table(database="awswrangler_test", table="csv_dataset")

	id	value	date
0	3	bar	2020-01-03
1	1	foo	2020-01-01
2	2	boo	2020-01-02

Overwriting

wr.s3.to_csv(
    df=df,
    path=path,
    index=False,
    dataset=True,
    mode="overwrite",
    database="awswrangler_test",
    table="csv_dataset"
)

wr.athena.read_sql_table(database="awswrangler_test", table="csv_dataset")

	id	value	date
0	3	bar	2020-01-03

Creating a Partitoned Dataset

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.s3.to_csv(
    df=df,
    path=path,
    index=False,
    dataset=True,
    mode="overwrite",
    database="awswrangler_test",
    table="csv_dataset",
    partition_cols=["date"]
)

wr.athena.read_sql_table(database="awswrangler_test", table="csv_dataset")

	id	value	date
0	2	boo	2020-01-02
1	1	foo	2020-01-01

Upserting partitions (overwrite_partitions)

df = pd.DataFrame({
    "id": [2, 3],
    "value": ["xoo", "bar"],
    "date": [date(2020, 1, 2), date(2020, 1, 3)]
})

wr.s3.to_csv(
    df=df,
    path=path,
    index=False,
    dataset=True,
    mode="overwrite_partitions",
    database="awswrangler_test",
    table="csv_dataset",
    partition_cols=["date"]
)

wr.athena.read_sql_table(database="awswrangler_test", table="csv_dataset")

	id	value	date
0	1	foo	2020-01-01
1	2	xoo	2020-01-02
0	3	bar	2020-01-03

BONUS - Glue/Athena integration

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.s3.to_csv(
    df=df,
    path=path,
    dataset=True,
    index=False,
    mode="overwrite",
    database="aws_data_wrangler",
    table="my_table",
    compression="gzip"
)

wr.athena.read_sql_query("SELECT * FROM my_table", database="aws_data_wrangler")

	id	value	date
0	1	foo	2020-01-01
1	2	boo	2020-01-02

12 - CSV Crawler

Wrangler can extract only the metadata from a Pandas DataFrame and then add it can be added to Glue Catalog as a table.

import awswrangler as wr
from datetime import datetime
import pandas as pd

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/csv_crawler/"

 ············

Creating a Pandas DataFrame

ts = lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S.%f")  # noqa
dt = lambda x: datetime.strptime(x, "%Y-%m-%d").date()  # noqa

df = pd.DataFrame(
    {
        "id": [1, 2, 3],
        "string": ["foo", None, "boo"],
        "float": [1.0, None, 2.0],
        "date": [dt("2020-01-01"), None, dt("2020-01-02")],
        "timestamp": [ts("2020-01-01 00:00:00.0"), None, ts("2020-01-02 00:00:01.0")],
        "bool": [True, None, False],
        "par0": [1, 1, 2],
        "par1": ["a", "b", "b"],
    }
)

df

	id	string	float	date	timestamp	bool	par0	par1
0	1	foo	1.0	2020-01-01	2020-01-01 00:00:00	True	1	a
1	2	None	NaN	None	NaT	None	1	b
2	3	boo	2.0	2020-01-02	2020-01-02 00:00:01	False	2	b

Extracting the metadata

columns_types, partitions_types = wr.catalog.extract_athena_types(
    df=df,
    file_format="csv",
    index=False,
    partition_cols=["par0", "par1"]
)

columns_types

{'id': 'bigint',
 'string': 'string',
 'float': 'double',
 'date': 'date',
 'timestamp': 'timestamp',
 'bool': 'boolean'}

partitions_types

{'par0': 'bigint', 'par1': 'string'}

Creating the table

wr.catalog.create_csv_table(
    table="csv_crawler",
    database="awswrangler_test",
    path=path,
    partitions_types=partitions_types,
    columns_types=columns_types,
)

Checking

wr.catalog.table(database="awswrangler_test", table="csv_crawler")

	Column Name	Type	Partition	Comment
0	id	bigint	False
1	string	string	False
2	float	double	False
3	date	date	False
4	timestamp	timestamp	False
5	bool	boolean	False
6	par0	bigint	True
7	par1	string	True

We can still using the extracted metadata to ensure all data types consistence to new data

df = pd.DataFrame(
    {
        "id": [1],
        "string": ["1"],
        "float": [1],
        "date": [ts("2020-01-01 00:00:00.0")],
        "timestamp": [dt("2020-01-02")],
        "bool": [1],
        "par0": [1],
        "par1": ["a"],
    }
)

df

	id	string	float	date	timestamp	bool	par0	par1
0	1	1	1	2020-01-01	2020-01-02	1	1	a

res = wr.s3.to_csv(
    df=df,
    path=path,
    index=False,
    dataset=True,
    database="awswrangler_test",
    table="csv_crawler",
    partition_cols=["par0", "par1"],
    dtype=columns_types
)

You can also extract the metadata directly from the Catalog if you want

dtype = wr.catalog.get_table_types(database="awswrangler_test", table="csv_crawler")

res = wr.s3.to_csv(
    df=df,
    path=path,
    index=False,
    dataset=True,
    database="awswrangler_test",
    table="csv_crawler",
    partition_cols=["par0", "par1"],
    dtype=dtype
)

Checking out

df = wr.athena.read_sql_table(database="awswrangler_test", table="csv_crawler")

df

	id	string	float	date	timestamp	bool	par0	par1
0	1	1	1.0	None	2020-01-02	True	1	a
1	1	1	1.0	None	2020-01-02	True	1	a

df.dtypes

id                    Int64
string               string
float               float64
date                 object
timestamp    datetime64[ns]
bool                boolean
par0                  Int64
par1                 string
dtype: object

Cleaning Up S3

wr.s3.delete_objects(path)

Cleaning Up the Database

wr.catalog.delete_table_if_exists(database="awswrangler_test", table="csv_crawler")

True

13 - Merging Datasets on S3

Wrangler has 3 different copy modes to store Parquet Datasets on Amazon S3.

• append (Default)

  Only adds new files without any delete.

• overwrite

  Deletes everything in the target directory and then add new files.

• overwrite_partitions (Partition Upsert)

  Only deletes the paths of partitions that should be updated and then writes the new partitions files. It’s like a “partition Upsert”.

from datetime import date
import awswrangler as wr
import pandas as pd

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path1 = f"s3://{bucket}/dataset1/"
path2 = f"s3://{bucket}/dataset2/"

 ············

Creating Dataset 1

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
    "date": [date(2020, 1, 1), date(2020, 1, 2)]
})

wr.s3.to_parquet(
    df=df,
    path=path1,
    dataset=True,
    mode="overwrite",
    partition_cols=["date"]
)

wr.s3.read_parquet(path1, dataset=True)

	id	value	date
0	1	foo	2020-01-01
1	2	boo	2020-01-02

Creating Dataset 2

df = pd.DataFrame({
    "id": [2, 3],
    "value": ["xoo", "bar"],
    "date": [date(2020, 1, 2), date(2020, 1, 3)]
})

dataset2_files = wr.s3.to_parquet(
    df=df,
    path=path2,
    dataset=True,
    mode="overwrite",
    partition_cols=["date"]
)["paths"]

wr.s3.read_parquet(path2, dataset=True)

	id	value	date
0	2	xoo	2020-01-02
1	3	bar	2020-01-03

Merging (Dataset 2 -> Dataset 1) (APPEND)

wr.s3.merge_datasets(
    source_path=path2,
    target_path=path1,
    mode="append"
)

wr.s3.read_parquet(path1, dataset=True)

	id	value	date
0	1	foo	2020-01-01
1	2	xoo	2020-01-02
2	2	boo	2020-01-02
3	3	bar	2020-01-03

Merging (Dataset 2 -> Dataset 1) (OVERWRITE_PARTITIONS)

wr.s3.merge_datasets(
    source_path=path2,
    target_path=path1,
    mode="overwrite_partitions"
)

wr.s3.read_parquet(path1, dataset=True)

	id	value	date
0	1	foo	2020-01-01
1	2	xoo	2020-01-02
2	3	bar	2020-01-03

Merging (Dataset 2 -> Dataset 1) (OVERWRITE)

wr.s3.merge_datasets(
    source_path=path2,
    target_path=path1,
    mode="overwrite"
)

wr.s3.read_parquet(path1, dataset=True)

	id	value	date
0	2	xoo	2020-01-02
1	3	bar	2020-01-03

Cleaning Up

wr.s3.delete_objects(path1)
wr.s3.delete_objects(path2)

14 - Schema Evolution

Wrangler support new columns on Parquet and CSV datasets through:

• wr.s3.to_parquet()

• wr.s3.store_parquet_metadata() i.e. “Crawler”

• wr.s3.to_csv()

from datetime import date
import awswrangler as wr
import pandas as pd

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/dataset/"

 ···········································

Creating the Dataset

Parquet

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    database="aws_data_wrangler",
    table="my_table"
)

wr.s3.read_parquet(path, dataset=True)

	id	value
0	1	foo
1	2	boo

CSV

df = pd.DataFrame({
    "id": [1, 2],
    "value": ["foo", "boo"],
})

wr.s3.to_csv(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    database="aws_data_wrangler",
    table="my_table"
)

wr.s3.read_csv(path, dataset=True)

Schema Version 0 on Glue Catalog (AWS Console)

Appending with NEW COLUMNS

Parquet

df = pd.DataFrame({
    "id": [3, 4],
    "value": ["bar", None],
    "date": [date(2020, 1, 3), date(2020, 1, 4)],
    "flag": [True, False]
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="append",
    database="aws_data_wrangler",
    table="my_table",
    catalog_versioning=True  # Optional
)

wr.s3.read_parquet(path, dataset=True, validate_schema=False)

	id	value	date	flag
0	3	bar	2020-01-03	True
1	4	None	2020-01-04	False
2	1	foo	NaN	NaN
3	2	boo	NaN	NaN

CSV

Note: for CSV datasets due to column ordering, by default, schema evolution is disabled. Enable it by passing schema_evolution=True flag

df = pd.DataFrame({
    "id": [3, 4],
    "value": ["bar", None],
    "date": [date(2020, 1, 3), date(2020, 1, 4)],
    "flag": [True, False]
})

wr.s3.to_csv(
    df=df,
    path=path,
    dataset=True,
    mode="append",
    database="aws_data_wrangler",
    table="my_table",
    schema_evolution=True,
    catalog_versioning=True  # Optional
)

wr.s3.read_csv(path, dataset=True, validate_schema=False)

Schema Version 1 on Glue Catalog (AWS Console)

Reading from Athena

wr.athena.read_sql_table(table="my_table", database="aws_data_wrangler")

	id	value	date	flag
0	3	bar	2020-01-03	True
1	4	None	2020-01-04	False
2	1	foo	None	<NA>
3	2	boo	None	<NA>

Cleaning Up

wr.s3.delete_objects(path)
wr.catalog.delete_table_if_exists(table="my_table", database="aws_data_wrangler")

True

15 - EMR

import awswrangler as wr
import boto3

Enter your bucket name:

import getpass
bucket = getpass.getpass()

 ··········································

Enter your Subnet ID:

subnet = getpass.getpass()

 ························

Creating EMR Cluster

cluster_id = wr.emr.create_cluster(subnet)

Uploading our PySpark script to Amazon S3

script = """
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("docker-awswrangler").getOrCreate()
sc = spark.sparkContext

print("Spark Initialized")
"""

_ = boto3.client("s3").put_object(
    Body=script,
    Bucket=bucket,
    Key="test.py"
)

Submit PySpark step

step_id = wr.emr.submit_step(cluster_id, command=f"spark-submit s3://{bucket}/test.py")

Wait Step

while wr.emr.get_step_state(cluster_id, step_id) != "COMPLETED":
    pass

Terminate Cluster

wr.emr.terminate_cluster(cluster_id)

16 - EMR & Docker

import awswrangler as wr
import boto3
import getpass

Enter your bucket name:

bucket = getpass.getpass()

 ··········································

Enter your Subnet ID:

subnet = getpass.getpass()

 ························

Build and Upload Docker Image to ECR repository

Replace the {ACCOUNT_ID} placeholder.

%%writefile Dockerfile

FROM amazoncorretto:8

RUN yum -y update
RUN yum -y install yum-utils
RUN yum -y groupinstall development

RUN yum list python3*
RUN yum -y install python3 python3-dev python3-pip python3-virtualenv

RUN python -V
RUN python3 -V

ENV PYSPARK_DRIVER_PYTHON python3
ENV PYSPARK_PYTHON python3

RUN pip3 install --upgrade pip
RUN pip3 install awswrangler

RUN python3 -c "import awswrangler as wr"

%%bash

docker build -t 'local/emr-wrangler' .
aws ecr create-repository --repository-name emr-wrangler
docker tag local/emr-wrangler {ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/emr-wrangler:emr-wrangler
eval $(aws ecr get-login --region us-east-1 --no-include-email)
docker push {ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/emr-wrangler:emr-wrangler

Creating EMR Cluster

cluster_id = wr.emr.create_cluster(subnet, docker=True)

Refresh ECR credentials in the cluster (expiration time: 12h )

wr.emr.submit_ecr_credentials_refresh(cluster_id, path=f"s3://{bucket}/")

's-1B0O45RWJL8CL'

Uploading application script to Amazon S3 (PySpark)

script = """
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("docker-awswrangler").getOrCreate()
sc = spark.sparkContext

print("Spark Initialized")

import awswrangler as wr

print(f"Wrangler version: {wr.__version__}")
"""

boto3.client("s3").put_object(Body=script, Bucket=bucket, Key="test_docker.py");

Submit PySpark step

DOCKER_IMAGE = f"{wr.get_account_id()}.dkr.ecr.us-east-1.amazonaws.com/emr-wrangler:emr-wrangler"

step_id = wr.emr.submit_spark_step(
    cluster_id,
    f"s3://{bucket}/test_docker.py",
    docker_image=DOCKER_IMAGE
)

Wait Step

while wr.emr.get_step_state(cluster_id, step_id) != "COMPLETED":
    pass

Terminate Cluster

wr.emr.terminate_cluster(cluster_id)

Another example with custom configurations

cluster_id = wr.emr.create_cluster(
    cluster_name="my-demo-cluster-v2",
    logging_s3_path=f"s3://{bucket}/emr-logs/",
    emr_release="emr-6.0.0",
    subnet_id=subnet,
    emr_ec2_role="EMR_EC2_DefaultRole",
    emr_role="EMR_DefaultRole",
    instance_type_master="m5.2xlarge",
    instance_type_core="m5.2xlarge",
    instance_ebs_size_master=50,
    instance_ebs_size_core=50,
    instance_num_on_demand_master=0,
    instance_num_on_demand_core=0,
    instance_num_spot_master=1,
    instance_num_spot_core=2,
    spot_bid_percentage_of_on_demand_master=100,
    spot_bid_percentage_of_on_demand_core=100,
    spot_provisioning_timeout_master=5,
    spot_provisioning_timeout_core=5,
    spot_timeout_to_on_demand_master=False,
    spot_timeout_to_on_demand_core=False,
    python3=True,
    docker=True,
    spark_glue_catalog=True,
    hive_glue_catalog=True,
    presto_glue_catalog=True,
    debugging=True,
    applications=["Hadoop", "Spark", "Hive", "Zeppelin", "Livy"],
    visible_to_all_users=True,
    maximize_resource_allocation=True,
    keep_cluster_alive_when_no_steps=True,
    termination_protected=False,
    spark_pyarrow=True
)

wr.emr.submit_ecr_credentials_refresh(cluster_id, path=f"s3://{bucket}/emr/")

DOCKER_IMAGE = f"{wr.get_account_id()}.dkr.ecr.us-east-1.amazonaws.com/emr-wrangler:emr-wrangler"

step_id = wr.emr.submit_spark_step(
    cluster_id,
    f"s3://{bucket}/test_docker.py",
    docker_image=DOCKER_IMAGE
)

while wr.emr.get_step_state(cluster_id, step_id) != "COMPLETED":
    pass

wr.emr.terminate_cluster(cluster_id)

17 - Partition Projection

https://docs.aws.amazon.com/athena/latest/ug/partition-projection.html

import awswrangler as wr
import pandas as pd
from datetime import datetime
import getpass

Enter your bucket name:

bucket = getpass.getpass()

 ···········································

Integer projection

df = pd.DataFrame({
    "value": [1, 2, 3],
    "year": [2019, 2020, 2021],
    "month": [10, 11, 12],
    "day": [25, 26, 27]
})

df

	value	year	month	day
0	1	2019	10	25
1	2	2020	11	26
2	3	2021	12	27

wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/table_integer/",
    dataset=True,
    partition_cols=["year", "month", "day"],
    database="default",
    table="table_integer",
    projection_enabled=True,
    projection_types={
        "year": "integer",
        "month": "integer",
        "day": "integer"
    },
    projection_ranges={
        "year": "2000,2025",
        "month": "1,12",
        "day": "1,31"
    },
);

wr.athena.read_sql_query(f"SELECT * FROM table_integer", database="default")

	value	year	month	day
0	3	2021	12	27
1	2	2020	11	26
2	1	2019	10	25

Enum projection

df = pd.DataFrame({
    "value": [1, 2, 3],
    "city": ["São Paulo", "Tokio", "Seattle"],
})

df

	value	city
0	1	São Paulo
1	2	Tokio
2	3	Seattle

wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/table_enum/",
    dataset=True,
    partition_cols=["city"],
    database="default",
    table="table_enum",
    projection_enabled=True,
    projection_types={
        "city": "enum",
    },
    projection_values={
        "city": "São Paulo,Tokio,Seattle"
    },
);

wr.athena.read_sql_query(f"SELECT * FROM table_enum", database="default")

	value	city
0	1	São Paulo
1	3	Seattle
2	2	Tokio

Date projection

ts = lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S")
dt = lambda x: datetime.strptime(x, "%Y-%m-%d").date()

df = pd.DataFrame({
    "value": [1, 2, 3],
    "dt": [dt("2020-01-01"), dt("2020-01-02"), dt("2020-01-03")],
    "ts": [ts("2020-01-01 00:00:00"), ts("2020-01-01 00:00:01"), ts("2020-01-01 00:00:02")],
})

df

	value	dt	ts
0	1	2020-01-01	2020-01-01 00:00:00
1	2	2020-01-02	2020-01-01 00:00:01
2	3	2020-01-03	2020-01-01 00:00:02

wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/table_date/",
    dataset=True,
    partition_cols=["dt", "ts"],
    database="default",
    table="table_date",
    projection_enabled=True,
    projection_types={
        "dt": "date",
        "ts": "date",
    },
    projection_ranges={
        "dt": "2020-01-01,2020-01-03",
        "ts": "2020-01-01 00:00:00,2020-01-01 00:00:02"
    },
);

wr.athena.read_sql_query(f"SELECT * FROM table_date", database="default")

	value	dt	ts
0	1	2020-01-01	2020-01-01 00:00:00
1	2	2020-01-02	2020-01-01 00:00:01
2	3	2020-01-03	2020-01-01 00:00:02

Injected projection

df = pd.DataFrame({
    "value": [1, 2, 3],
    "uuid": ["761e2488-a078-11ea-bb37-0242ac130002", "b89ed095-8179-4635-9537-88592c0f6bc3", "87adc586-ce88-4f0a-b1c8-bf8e00d32249"],
})

df

	value	uuid
0	1	761e2488-a078-11ea-bb37-0242ac130002
1	2	b89ed095-8179-4635-9537-88592c0f6bc3
2	3	87adc586-ce88-4f0a-b1c8-bf8e00d32249

wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/table_injected/",
    dataset=True,
    partition_cols=["uuid"],
    database="default",
    table="table_injected",
    projection_enabled=True,
    projection_types={
        "uuid": "injected",
    }
);

wr.athena.read_sql_query(
    sql=f"SELECT * FROM table_injected WHERE uuid='b89ed095-8179-4635-9537-88592c0f6bc3'",
    database="default"
)

	value	uuid
0	2	b89ed095-8179-4635-9537-88592c0f6bc3

Cleaning Up

wr.s3.delete_objects(f"s3://{bucket}/table_integer/")
wr.s3.delete_objects(f"s3://{bucket}/table_enum/")
wr.s3.delete_objects(f"s3://{bucket}/table_date/")
wr.s3.delete_objects(f"s3://{bucket}/table_injected/")

wr.catalog.delete_table_if_exists(table="table_integer", database="default")
wr.catalog.delete_table_if_exists(table="table_enum", database="default")
wr.catalog.delete_table_if_exists(table="table_date", database="default")
wr.catalog.delete_table_if_exists(table="table_injected", database="default");

18 - QuickSight

For this tutorial we will use the public AWS COVID-19 data lake.

References:

• A public data lake for analysis of COVID-19 data

• Exploring the public AWS COVID-19 data lake

• CloudFormation template

Please, install the Cloudformation template above to have access to the public data lake.

P.S. To be able to access the public data lake, you must allow explicitly QuickSight to access the related external bucket.

import awswrangler as wr
from time import sleep

List users of QuickSight account

[{"username": user["UserName"], "role": user["Role"]} for user in wr.quicksight.list_users('default')]

[{'username': 'dev', 'role': 'ADMIN'}]

wr.catalog.databases()

	Database	Description
0	aws_data_wrangler	AWS Data Wrangler Test Arena - Glue Database
1	awswrangler_test
2	covid-19
3	default	Default Hive database

wr.catalog.tables(database="covid-19")

	Database	Table	Description	Columns	Partitions
0	covid-19	alleninstitute_comprehend_medical	Comprehend Medical results run against Allen I...	paper_id, date, dx_name, test_name, procedure_...
1	covid-19	alleninstitute_metadata	Metadata on papers pulled from the Allen Insti...	cord_uid, sha, source_x, title, doi, pmcid, pu...
2	covid-19	country_codes	Lookup table for country codes	country, alpha-2 code, alpha-3 code, numeric c...
3	covid-19	county_populations	Lookup table for population for each county ba...	id, id2, county, state, population estimate 2018
4	covid-19	covid_knowledge_graph_edges	AWS Knowledge Graph for COVID-19 data	id, label, from, to, score
5	covid-19	covid_knowledge_graph_nodes_author	AWS Knowledge Graph for COVID-19 data	id, label, first, last, full_name
6	covid-19	covid_knowledge_graph_nodes_concept	AWS Knowledge Graph for COVID-19 data	id, label, entity, concept
7	covid-19	covid_knowledge_graph_nodes_institution	AWS Knowledge Graph for COVID-19 data	id, label, institution, country, settlement
8	covid-19	covid_knowledge_graph_nodes_paper	AWS Knowledge Graph for COVID-19 data	id, label, doi, sha_code, publish_time, source...
9	covid-19	covid_knowledge_graph_nodes_topic	AWS Knowledge Graph for COVID-19 data	id, label, topic, topic_num
10	covid-19	covid_testing_states_daily	USA total test daily trend by state. Sourced ...	date, state, positive, negative, pending, hosp...
11	covid-19	covid_testing_us_daily	USA total test daily trend. Sourced from covi...	date, states, positive, negative, posneg, pend...
12	covid-19	covid_testing_us_total	USA total tests. Sourced from covidtracking.c...	positive, negative, posneg, hospitalized, deat...
13	covid-19	covidcast_data	CMU Delphi's COVID-19 Surveillance Data	data_source, signal, geo_type, time_value, geo...
14	covid-19	covidcast_metadata	CMU Delphi's COVID-19 Surveillance Metadata	data_source, signal, time_type, geo_type, min_...
15	covid-19	enigma_jhu	Johns Hopkins University Consolidated data on ...	fips, admin2, province_state, country_region, ...
16	covid-19	enigma_jhu_timeseries	Johns Hopkins University data on COVID-19 case...	uid, fips, iso2, iso3, code3, admin2, latitude...
17	covid-19	hospital_beds	Data on hospital beds and their utilization in...	objectid, hospital_name, hospital_type, hq_add...
18	covid-19	nytimes_counties	Data on COVID-19 cases from NY Times at US cou...	date, county, state, fips, cases, deaths
19	covid-19	nytimes_states	Data on COVID-19 cases from NY Times at US sta...	date, state, fips, cases, deaths
20	covid-19	prediction_models_county_predictions	County-level Predictions Data. Sourced from Yu...	countyfips, countyname, statename, severity_co...
21	covid-19	prediction_models_severity_index	Severity Index models. Sourced from Yu Group a...	severity_1-day, severity_2-day, severity_3-day...
22	covid-19	tableau_covid_datahub	COVID-19 data that has been gathered and unifi...	country_short_name, country_alpha_3_code, coun...
23	covid-19	tableau_jhu	Johns Hopkins University data on COVID-19 case...	case_type, cases, difference, date, country_re...
24	covid-19	us_state_abbreviations	Lookup table for US state abbreviations	state, abbreviation
25	covid-19	world_cases_deaths_testing	Data on confirmed cases, deaths, and testing. ...	iso_code, location, date, total_cases, new_cas...

Create data source of QuickSight Note: data source stores the connection information.

wr.quicksight.create_athena_data_source(
    name="covid-19",
    workgroup="primary",
    allowed_to_manage=["dev"]
)

wr.catalog.tables(database="covid-19", name_contains="nyt")

	Database	Table	Description	Columns	Partitions
0	covid-19	nytimes_counties	Data on COVID-19 cases from NY Times at US cou...	date, county, state, fips, cases, deaths
1	covid-19	nytimes_states	Data on COVID-19 cases from NY Times at US sta...	date, state, fips, cases, deaths

wr.athena.read_sql_query("SELECT * FROM nytimes_counties limit 10", database="covid-19", ctas_approach=False)

	date	county	state	fips	cases	deaths
0	2020-01-21	Snohomish	Washington	53061	1	0
1	2020-01-22	Snohomish	Washington	53061	1	0
2	2020-01-23	Snohomish	Washington	53061	1	0
3	2020-01-24	Cook	Illinois	17031	1	0
4	2020-01-24	Snohomish	Washington	53061	1	0
5	2020-01-25	Orange	California	06059	1	0
6	2020-01-25	Cook	Illinois	17031	1	0
7	2020-01-25	Snohomish	Washington	53061	1	0
8	2020-01-26	Maricopa	Arizona	04013	1	0
9	2020-01-26	Los Angeles	California	06037	1	0

sql = """
SELECT
  j.*,
  co.Population,
  co.county AS county2,
  hb.*
FROM
  (
    SELECT
      date,
      county,
      state,
      fips,
      cases as confirmed,
      deaths
    FROM "covid-19".nytimes_counties
  ) j
  LEFT OUTER JOIN (
    SELECT
      DISTINCT county,
      state,
      "population estimate 2018" AS Population
    FROM
      "covid-19".county_populations
    WHERE
      state IN (
        SELECT
          DISTINCT state
        FROM
          "covid-19".nytimes_counties
      )
      AND county IN (
        SELECT
          DISTINCT county as county
        FROM "covid-19".nytimes_counties
      )
  ) co ON co.county = j.county
  AND co.state = j.state
  LEFT OUTER JOIN (
    SELECT
      count(objectid) as Hospital,
      fips as hospital_fips,
      sum(num_licensed_beds) as licensed_beds,
      sum(num_staffed_beds) as staffed_beds,
      sum(num_icu_beds) as icu_beds,
      avg(bed_utilization) as bed_utilization,
      sum(
        potential_increase_in_bed_capac
      ) as potential_increase_bed_capacity
    FROM "covid-19".hospital_beds
    WHERE
      fips in (
        SELECT
          DISTINCT fips
        FROM
          "covid-19".nytimes_counties
      )
    GROUP BY
      2
  ) hb ON hb.hospital_fips = j.fips
"""

wr.athena.read_sql_query(sql, database="covid-19", ctas_approach=False)

	date	county	state	fips	confirmed	deaths	population	county2	Hospital	hospital_fips	licensed_beds	staffed_beds	icu_beds	bed_utilization	potential_increase_bed_capacity
0	2020-04-12	Park	Montana	30067	7	0	16736	Park	0	30067	25	25	4	0.432548	0
1	2020-04-12	Ravalli	Montana	30081	3	0	43172	Ravalli	0	30081	25	25	5	0.567781	0
2	2020-04-12	Silver Bow	Montana	30093	11	0	34993	Silver Bow	0	30093	98	71	11	0.551457	27
3	2020-04-12	Clay	Nebraska	31035	2	0	6214	Clay	<NA>	<NA>	<NA>	<NA>	<NA>	NaN	<NA>
4	2020-04-12	Cuming	Nebraska	31039	2	0	8940	Cuming	0	31039	25	25	4	0.204493	0
...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...
227684	2020-06-11	Hockley	Texas	48219	28	1	22980	Hockley	0	48219	48	48	8	0.120605	0
227685	2020-06-11	Hudspeth	Texas	48229	11	0	4795	Hudspeth	<NA>	<NA>	<NA>	<NA>	<NA>	NaN	<NA>
227686	2020-06-11	Jones	Texas	48253	633	0	19817	Jones	0	48253	45	7	1	0.718591	38
227687	2020-06-11	La Salle	Texas	48283	4	0	7531	La Salle	<NA>	<NA>	<NA>	<NA>	<NA>	NaN	<NA>
227688	2020-06-11	Limestone	Texas	48293	36	1	23519	Limestone	0	48293	78	69	9	0.163940	9

227689 rows × 15 columns

Create Dataset with custom SQL option

wr.quicksight.create_athena_dataset(
    name="covid19-nytimes-usa",
    sql=sql,
    sql_name='CustomSQL',
    data_source_name="covid-19",
    import_mode='SPICE',
    allowed_to_manage=["dev"]
)

ingestion_id = wr.quicksight.create_ingestion("covid19-nytimes-usa")

Wait ingestion

while wr.quicksight.describe_ingestion(ingestion_id=ingestion_id, dataset_name="covid19-nytimes-usa")["IngestionStatus"] not in ["COMPLETED", "FAILED"]:
    sleep(1)

Describe last ingestion

wr.quicksight.describe_ingestion(ingestion_id=ingestion_id, dataset_name="covid19-nytimes-usa")["RowInfo"]

{'RowsIngested': 227689, 'RowsDropped': 0}

List all ingestions

[{"time": user["CreatedTime"], "source": user["RequestSource"]} for user in wr.quicksight.list_ingestions("covid19-nytimes-usa")]

[{'time': datetime.datetime(2020, 6, 12, 15, 13, 46, 996000, tzinfo=tzlocal()),
  'source': 'MANUAL'},
 {'time': datetime.datetime(2020, 6, 12, 15, 13, 42, 344000, tzinfo=tzlocal()),
  'source': 'MANUAL'}]

Create new dataset from a table directly

wr.quicksight.create_athena_dataset(
    name="covid-19-tableau_jhu",
    table="tableau_jhu",
    data_source_name="covid-19",
    database="covid-19",
    import_mode='DIRECT_QUERY',
    rename_columns={
        "cases": "Count_of_Cases",
        "combined_key": "County"
    },
    cast_columns_types={
        "Count_of_Cases": "INTEGER"
    },
    allowed_to_manage=["dev"]
)

Cleaning up

wr.quicksight.delete_data_source("covid-19")
wr.quicksight.delete_dataset("covid19-nytimes-usa")
wr.quicksight.delete_dataset("covid-19-tableau_jhu")

19 - Amazon Athena Cache

Wrangler has a cache strategy that is disabled by default and can be enabled passing max_cache_seconds biggier than 0. This cache strategy for Amazon Athena can help you to decrease query times and costs.

When calling read_sql_query, instead of just running the query, we now can verify if the query has been run before. If so, and this last run was within max_cache_seconds (a new parameter to read_sql_query), we return the same results as last time if they are still available in S3. We have seen this increase performance more than 100x, but the potential is pretty much infinite.

The detailed approach is: - When read_sql_query is called with max_cache_seconds > 0 (it defaults to 0), we check for the last queries run by the same workgroup (the most we can get without pagination). - By default it will check the last 50 queries, but you can customize it throught the max_cache_query_inspections argument. - We then sort those queries based on CompletionDateTime, descending - For each of those queries, we check if their CompletionDateTime is still within the max_cache_seconds window. If so, we check if the query string is the same as now (with some smart heuristics to guarantee coverage over both ctas_approaches). If they are the same, we check if the last one’s results are still on S3, and then return them instead of re-running the query. - During the whole cache resolution phase, if there is anything wrong, the logic falls back to the usual read_sql_query path.

P.S. The ``cache scope is bounded for the current workgroup``, so you will be able to reuse queries results from others colleagues running in the same environment.

import awswrangler as wr

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/data/"

 ···········································

Checking/Creating Glue Catalog Databases

if "awswrangler_test" not in wr.catalog.databases().values:
    wr.catalog.create_database("awswrangler_test")

Creating a Parquet Table from the NOAA’s CSV files

Reference

cols = ["id", "dt", "element", "value", "m_flag", "q_flag", "s_flag", "obs_time"]

df = wr.s3.read_csv(
    path="s3://noaa-ghcn-pds/csv/189",
    names=cols,
    parse_dates=["dt", "obs_time"])  # Read 10 files from the 1890 decade (~1GB)

df

	id	dt	element	value	m_flag	q_flag	s_flag	obs_time
0	AGE00135039	1890-01-01	TMAX	160	NaN	NaN	E	NaN
1	AGE00135039	1890-01-01	TMIN	30	NaN	NaN	E	NaN
2	AGE00135039	1890-01-01	PRCP	45	NaN	NaN	E	NaN
3	AGE00147705	1890-01-01	TMAX	140	NaN	NaN	E	NaN
4	AGE00147705	1890-01-01	TMIN	74	NaN	NaN	E	NaN
...	...	...	...	...	...	...	...	...
29240014	UZM00038457	1899-12-31	PRCP	16	NaN	NaN	r	NaN
29240015	UZM00038457	1899-12-31	TAVG	-73	NaN	NaN	r	NaN
29240016	UZM00038618	1899-12-31	TMIN	-76	NaN	NaN	r	NaN
29240017	UZM00038618	1899-12-31	PRCP	0	NaN	NaN	r	NaN
29240018	UZM00038618	1899-12-31	TAVG	-60	NaN	NaN	r	NaN

29240019 rows × 8 columns

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    database="awswrangler_test",
    table="noaa"
);

wr.catalog.table(database="awswrangler_test", table="noaa")

	Column Name	Type	Partition	Comment
0	id	string	False
1	dt	timestamp	False
2	element	string	False
3	value	bigint	False
4	m_flag	string	False
5	q_flag	string	False
6	s_flag	string	False
7	obs_time	string	False

The test query

The more computational resources the query needs, the more the cache will help you. That’s why we’re doing it using this long running query.

query = """
SELECT
    n1.element,
    count(1) as cnt
FROM
    noaa n1
JOIN
    noaa n2
ON
    n1.id = n2.id
GROUP BY
    n1.element
"""

First execution…

%%time

wr.athena.read_sql_query(query, database="awswrangler_test")

CPU times: user 5.31 s, sys: 232 ms, total: 5.54 s
Wall time: 6min 42s

	element	cnt
0	WDMV	49755046
1	SNWD	5089486328
2	DATN	10817510
3	DAPR	102579666
4	MDTN	10817510
5	WT03	71184687
6	WT09	584412
7	TOBS	146984266
8	DASF	7764526
9	WT04	9648963
10	WT18	92635444
11	WT01	87526136
12	WT16	323354156
13	PRCP	71238907298
14	SNOW	21950890838
15	WT06	307339
16	TAVG	2340863803
17	TMIN	41450979633
18	MDTX	11210687
19	WT07	4486872
20	WT10	137873
21	EVAP	970404
22	WT14	8073701
23	DATX	11210687
24	WT08	33933005
25	WT05	8211491
26	TMAX	39876132467
27	MDPR	114320989
28	WT11	22212890
29	DWPR	69005655
30	MDSF	12004843

Second execution with CACHE (400x faster)

%%time

wr.athena.read_sql_query(query, database="awswrangler_test", max_cache_seconds=900)

CPU times: user 493 ms, sys: 34.9 ms, total: 528 ms
Wall time: 975 ms

	element	cnt
0	WDMV	49755046
1	SNWD	5089486328
2	DATN	10817510
3	DAPR	102579666
4	MDTN	10817510
5	WT03	71184687
6	WT09	584412
7	TOBS	146984266
8	DASF	7764526
9	WT04	9648963
10	WT18	92635444
11	WT01	87526136
12	WT16	323354156
13	PRCP	71238907298
14	SNOW	21950890838
15	WT06	307339
16	TAVG	2340863803
17	TMIN	41450979633
18	MDTX	11210687
19	WT07	4486872
20	WT10	137873
21	EVAP	970404
22	WT14	8073701
23	DATX	11210687
24	WT08	33933005
25	WT05	8211491
26	TMAX	39876132467
27	MDPR	114320989
28	WT11	22212890
29	DWPR	69005655
30	MDSF	12004843

Allowing Wrangler to inspect up to 500 historical queries to find same result to reuse.

%%time

wr.athena.read_sql_query(query, database="awswrangler_test", max_cache_seconds=900, max_cache_query_inspections=500)

CPU times: user 504 ms, sys: 44 ms, total: 548 ms
Wall time: 1.19 s

	element	cnt
0	WDMV	49755046
1	SNWD	5089486328
2	DATN	10817510
3	DAPR	102579666
4	MDTN	10817510
5	WT03	71184687
6	WT09	584412
7	TOBS	146984266
8	DASF	7764526
9	WT04	9648963
10	WT18	92635444
11	WT01	87526136
12	WT16	323354156
13	PRCP	71238907298
14	SNOW	21950890838
15	WT06	307339
16	TAVG	2340863803
17	TMIN	41450979633
18	MDTX	11210687
19	WT07	4486872
20	WT10	137873
21	EVAP	970404
22	WT14	8073701
23	DATX	11210687
24	WT08	33933005
25	WT05	8211491
26	TMAX	39876132467
27	MDPR	114320989
28	WT11	22212890
29	DWPR	69005655
30	MDSF	12004843

Cleaning Up S3

wr.s3.delete_objects(path)

Delete table

wr.catalog.delete_table_if_exists(database="awswrangler_test", table="noaa")

True

Delete Database

wr.catalog.delete_database('awswrangler_test')

20 - Spark Table Interoperability

Wrangler has no difficults to insert, overwrite or do any other kind of interaction with a Table created by Apache Spark.

But if you want to do the oposite (Spark interacting with a table created by Wrangler) you should be aware that Wrangler follows the Hive’s format and you must be explicit when using the Spark’s saveAsTable method:

spark_df.write.format("hive").saveAsTable("database.table")

Or just move forward using the insertInto alternative:

spark_df.write.insertInto("database.table")

21 - Global Configurations

Wrangler has two ways to set global configurations that will override the regular default arguments configured in functions signatures.

• Environment variables

• wr.config

P.S. Check thefunction API docto see if your function has some argument that can be configured through Global configurations.

P.P.S. One exception to the above mentioned rules is the ``botocore_config`` property. It cannot be set through environment variables but only via ``wr.config``. It will be used as the ``botocore.config.Config`` for all underlying ``boto3`` calls. The default config is ``botocore.config.Config(retries={“max_attempts”: 5}, connect_timeout=10, max_pool_connections=10)``. If you only want to change the retry behavior, you can use the environment variables ``AWS_MAX_ATTEMPTS`` and ``AWS_RETRY_MODE``. (seeBoto3 documentation)

Environment Variables

%env WR_DATABASE=default
%env WR_CTAS_APPROACH=False
%env WR_MAX_CACHE_SECONDS=900
%env WR_MAX_CACHE_QUERY_INSPECTIONS=500
%env WR_MAX_REMOTE_CACHE_ENTRIES=50
%env WR_MAX_LOCAL_CACHE_ENTRIES=100

env: WR_DATABASE=default
env: WR_CTAS_APPROACH=False
env: WR_MAX_CACHE_SECONDS=900
env: WR_MAX_CACHE_QUERY_INSPECTIONS=500
env: WR_MAX_REMOTE_CACHE_ENTRIES=50
env: WR_MAX_LOCAL_CACHE_ENTRIES=100

import awswrangler as wr
import botocore

wr.athena.read_sql_query("SELECT 1 AS FOO")

	foo
0	1

Resetting

# Specific
wr.config.reset("database")
# All
wr.config.reset()

wr.config

wr.config.database = "default"
wr.config.ctas_approach = False
wr.config.max_cache_seconds = 900
wr.config.max_cache_query_inspections = 500
wr.config.max_remote_cache_entries = 50
wr.config.max_local_cache_entries = 100
# Set botocore.config.Config that will be used for all boto3 calls
wr.config.botocore_config = botocore.config.Config(
    retries={"max_attempts": 10},
    connect_timeout=20,
    max_pool_connections=20
)

wr.athena.read_sql_query("SELECT 1 AS FOO")

	foo
0	1

Visualizing

wr.config

	name	Env. Variable	type	nullable	enforced	configured	value
0	catalog_id	WR_CATALOG_ID	<class 'str'>	True	False	False	None
1	concurrent_partitioning	WR_CONCURRENT_PARTITIONING	<class 'bool'>	False	False	False	None
2	ctas_approach	WR_CTAS_APPROACH	<class 'bool'>	False	False	True	False
3	database	WR_DATABASE	<class 'str'>	True	False	True	default
4	max_cache_query_inspections	WR_MAX_CACHE_QUERY_INSPECTIONS	<class 'int'>	False	False	True	500
5	max_cache_seconds	WR_MAX_CACHE_SECONDS	<class 'int'>	False	False	True	900
6	max_remote_cache_entries	WR_MAX_REMOTE_CACHE_ENTRIES	<class 'int'>	False	False	True	50
7	max_local_cache_entries	WR_MAX_LOCAL_CACHE_ENTRIES	<class 'int'>	False	False	True	100
8	s3_block_size	WR_S3_BLOCK_SIZE	<class 'int'>	False	True	False	None
9	workgroup	WR_WORKGROUP	<class 'str'>	False	True	False	None
10	s3_endpoint_url	WR_S3_ENDPOINT_URL	<class 'str'>	True	True	True	None
11	athena_endpoint_url	WR_ATHENA_ENDPOINT_URL	<class 'str'>	True	True	True	None
12	sts_endpoint_url	WR_STS_ENDPOINT_URL	<class 'str'>	True	True	True	None
13	glue_endpoint_url	WR_GLUE_ENDPOINT_URL	<class 'str'>	True	True	True	None
14	redshift_endpoint_url	WR_REDSHIFT_ENDPOINT_URL	<class 'str'>	True	True	True	None
15	kms_endpoint_url	WR_KMS_ENDPOINT_URL	<class 'str'>	True	True	True	None
16	emr_endpoint_url	WR_EMR_ENDPOINT_URL	<class 'str'>	True	True	True	None
17	botocore_config	WR_BOTOCORE_CONFIG	<class 'botocore.config.Config'>	True	False	True	<botocore.config.Config object at 0x000002D3A362E760>

22 - Writing Partitions Concurrently

• concurrent_partitioning argument:

  If True will increase the parallelism level during the partitions writing. It will decrease the
  writing time and increase the memory usage.
  

P.S. Check thefunction API docto see it has some argument that can be configured through Global configurations.

%reload_ext memory_profiler

import awswrangler as wr

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/data/"

 ············

Reading 4 GB of CSV from NOAA’s historical data and creating a year column

noaa_path = "s3://noaa-ghcn-pds/csv/193"

cols = ["id", "dt", "element", "value", "m_flag", "q_flag", "s_flag", "obs_time"]
dates = ["dt", "obs_time"]
dtype = {x: "category" for x in ["element", "m_flag", "q_flag", "s_flag"]}

df = wr.s3.read_csv(noaa_path, names=cols, parse_dates=dates, dtype=dtype)

df["year"] = df["dt"].dt.year

print(f"Number of rows: {len(df.index)}")
print(f"Number of columns: {len(df.columns)}")

Number of rows: 125407761
Number of columns: 9

Default Writing

%%time
%%memit

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    partition_cols=["year"],
);

peak memory: 22169.04 MiB, increment: 11119.68 MiB
CPU times: user 49 s, sys: 12.5 s, total: 1min 1s
Wall time: 1min 11s

Concurrent Partitioning (Decreasing writing time, but increasing memory usage)

%%time
%%memit

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    partition_cols=["year"],
    concurrent_partitioning=True  # <-----
);

peak memory: 27819.48 MiB, increment: 15743.30 MiB
CPU times: user 52.3 s, sys: 13.6 s, total: 1min 5s
Wall time: 41.6 s

23 - Flexible Partitions Filter (PUSH-DOWN)

• partition_filter argument:

  - Callback Function filters to apply on PARTITION columns (PUSH-DOWN filter).
  - This function MUST receive a single argument (Dict[str, str]) where keys are partitions names and values are partitions values.
  - This function MUST return a bool, True to read the partition or False to ignore it.
  - Ignored if `dataset=False`.
  

P.S. Check thefunction API docto see it has some argument that can be configured through Global configurations.

import awswrangler as wr
import pandas as pd

Enter your bucket name:

import getpass
bucket = getpass.getpass()
path = f"s3://{bucket}/dataset/"

 ············

Creating the Dataset (PARQUET)

df = pd.DataFrame({
    "id": [1, 2, 3],
    "value": ["foo", "boo", "bar"],
})

wr.s3.to_parquet(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    partition_cols=["value"]
)

wr.s3.read_parquet(path, dataset=True)

	id	value
0	3	bar
1	2	boo
2	1	foo

Example 1

my_filter = lambda x: x["value"].endswith("oo")

wr.s3.read_parquet(path, dataset=True, partition_filter=my_filter)

	id	value
0	2	boo
1	1	foo

Example 2

from Levenshtein import distance


def my_filter(partitions):
    return distance("boo", partitions["value"]) <= 1


wr.s3.read_parquet(path, dataset=True, partition_filter=my_filter)

	id	value
0	2	boo
1	1	foo

Creating the Dataset (CSV)

df = pd.DataFrame({
    "id": [1, 2, 3],
    "value": ["foo", "boo", "bar"],
})

wr.s3.to_csv(
    df=df,
    path=path,
    dataset=True,
    mode="overwrite",
    partition_cols=["value"],
    compression="gzip",
    index=False
)

wr.s3.read_csv(path, dataset=True)

	id	value
0	3	bar
1	2	boo
2	1	foo

Example 1

my_filter = lambda x: x["value"].endswith("oo")

wr.s3.read_csv(path, dataset=True, partition_filter=my_filter)

	id	value
0	2	boo
1	1	foo

Example 2

from Levenshtein import distance


def my_filter(partitions):
    return distance("boo", partitions["value"]) <= 1


wr.s3.read_csv(path, dataset=True, partition_filter=my_filter)

	id	value
0	2	boo
1	1	foo

24 - Athena Query Metadata

For wr.athena.read_sql_query() and wr.athena.read_sql_table() the resulting DataFrame (or every DataFrame in the returned Iterator for chunked queries) have a query_metadata attribute, which brings the query result metadata returned by Boto3/Athena.

The expected query_metadata format is the same returned by:

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/athena.html#Athena.Client.get_query_execution

Environment Variables

%env WR_DATABASE=default

env: WR_DATABASE=default

import awswrangler as wr

df = wr.athena.read_sql_query("SELECT 1 AS foo")

df

	foo
0	1

Getting statistics from query metadata

print(f'DataScannedInBytes:            {df.query_metadata["Statistics"]["DataScannedInBytes"]}')
print(f'TotalExecutionTimeInMillis:    {df.query_metadata["Statistics"]["TotalExecutionTimeInMillis"]}')
print(f'QueryQueueTimeInMillis:        {df.query_metadata["Statistics"]["QueryQueueTimeInMillis"]}')
print(f'QueryPlanningTimeInMillis:     {df.query_metadata["Statistics"]["QueryPlanningTimeInMillis"]}')
print(f'ServiceProcessingTimeInMillis: {df.query_metadata["Statistics"]["ServiceProcessingTimeInMillis"]}')

DataScannedInBytes:            0
TotalExecutionTimeInMillis:    2311
QueryQueueTimeInMillis:        121
QueryPlanningTimeInMillis:     250
ServiceProcessingTimeInMillis: 37

25 - Redshift - Loading Parquet files with Spectrum

Enter your bucket name:

import getpass
bucket = getpass.getpass()
PATH = f"s3://{bucket}/files/"

 ···········································

Mocking some Parquet Files on S3

import awswrangler as wr
import pandas as pd

df = pd.DataFrame({
    "col0": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
    "col1": ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"],
})

df

	col0	col1
0	0	a
1	1	b
2	2	c
3	3	d
4	4	e
5	5	f
6	6	g
7	7	h
8	8	i
9	9	j

wr.s3.to_parquet(df, PATH, max_rows_by_file=2, dataset=True, mode="overwrite");

Crawling the metadata and adding into Glue Catalog

wr.s3.store_parquet_metadata(
    path=PATH,
    database="aws_data_wrangler",
    table="test",
    dataset=True,
    mode="overwrite"
)

({'col0': 'bigint', 'col1': 'string'}, None, None)

Running the CTAS query to load the data into Redshift storage

con = wr.redshift.connect(connection="aws-data-wrangler-redshift")

query = "CREATE TABLE public.test AS (SELECT * FROM aws_data_wrangler_external.test)"

with con.cursor() as cursor:
    cursor.execute(query)

Running an INSERT INTO query to load MORE data into Redshift storage

df = pd.DataFrame({
    "col0": [10, 11],
    "col1": ["k", "l"],
})
wr.s3.to_parquet(df, PATH, dataset=True, mode="overwrite");

query = "INSERT INTO public.test (SELECT * FROM aws_data_wrangler_external.test)"

with con.cursor() as cursor:
    cursor.execute(query)

Checking the result

query = "SELECT * FROM public.test"

wr.redshift.read_sql_table(con=con, schema="public", table="test")

	col0	col1
0	5	f
1	1	b
2	3	d
3	6	g
4	8	i
5	10	k
6	4	e
7	0	a
8	2	c
9	7	h
10	9	j
11	11	l

con.close()

26 - Amazon Timestream

Creating resources

import awswrangler as wr
import pandas as pd
from datetime import datetime

wr.timestream.create_database("sampleDB")
wr.timestream.create_table("sampleDB", "sampleTable", memory_retention_hours=1, magnetic_retention_days=1);

Write

df = pd.DataFrame(
    {
        "time": [datetime.now(), datetime.now(), datetime.now()],
        "dim0": ["foo", "boo", "bar"],
        "dim1": [1, 2, 3],
        "measure": [1.0, 1.1, 1.2],
    }
)

rejected_records = wr.timestream.write(
    df=df,
    database="sampleDB",
    table="sampleTable",
    time_col="time",
    measure_col="measure",
    dimensions_cols=["dim0", "dim1"],
)

print(f"Number of rejected records: {len(rejected_records)}")

Number of rejected records: 0

Query

wr.timestream.query(
    'SELECT time, measure_value::double, dim0, dim1 FROM "sampleDB"."sampleTable" ORDER BY time DESC LIMIT 3'
)

	time	measure_value::double	dim0	dim1
0	2020-12-08 19:15:32.468	1.0	foo	1
1	2020-12-08 19:15:32.468	1.2	bar	3
2	2020-12-08 19:15:32.468	1.1	boo	2

Deleting resources

wr.timestream.delete_table("sampleDB", "sampleTable")
wr.timestream.delete_database("sampleDB")

27 - Amazon Timestream - Example 2

Reading test data

import awswrangler as wr
import pandas as pd
from datetime import datetime

df = pd.read_csv(
    "https://raw.githubusercontent.com/awslabs/amazon-timestream-tools/master/sample_apps/data/sample.csv",
    names=[
        "ignore0",
        "region",
        "ignore1",
        "az",
        "ignore2",
        "hostname",
        "measure_kind",
        "measure",
        "ignore3",
        "ignore4",
        "ignore5",
    ],
    usecols=["region", "az", "hostname", "measure_kind", "measure"],
)
df["time"] = datetime.now()
df.reset_index(inplace=True, drop=False)

df

	index	region	az	hostname	measure_kind	measure	time
0	0	us-east-1	us-east-1a	host-fj2hx	cpu_utilization	21.394363	2020-12-08 16:18:47.599597
1	1	us-east-1	us-east-1a	host-fj2hx	memory_utilization	68.563420	2020-12-08 16:18:47.599597
2	2	us-east-1	us-east-1a	host-6kMPE	cpu_utilization	17.144579	2020-12-08 16:18:47.599597
3	3	us-east-1	us-east-1a	host-6kMPE	memory_utilization	73.507870	2020-12-08 16:18:47.599597
4	4	us-east-1	us-east-1a	host-sxj7X	cpu_utilization	26.584865	2020-12-08 16:18:47.599597
...	...	...	...	...	...	...	...
125995	125995	eu-north-1	eu-north-1c	host-De8RB	memory_utilization	68.063468	2020-12-08 16:18:47.599597
125996	125996	eu-north-1	eu-north-1c	host-2z8tn	memory_utilization	72.203680	2020-12-08 16:18:47.599597
125997	125997	eu-north-1	eu-north-1c	host-2z8tn	cpu_utilization	29.212219	2020-12-08 16:18:47.599597
125998	125998	eu-north-1	eu-north-1c	host-9FczW	memory_utilization	71.746134	2020-12-08 16:18:47.599597
125999	125999	eu-north-1	eu-north-1c	host-9FczW	cpu_utilization	1.677793	2020-12-08 16:18:47.599597

126000 rows × 7 columns

Creating resources

wr.timestream.create_database("sampleDB")
wr.timestream.create_table("sampleDB", "sampleTable", memory_retention_hours=1, magnetic_retention_days=1);

Write CPU_UTILIZATION records

df_cpu = df[df.measure_kind == "cpu_utilization"].copy()
df_cpu.rename(columns={"measure": "cpu_utilization"}, inplace=True)
df_cpu

	index	region	az	hostname	measure_kind	cpu_utilization	time
0	0	us-east-1	us-east-1a	host-fj2hx	cpu_utilization	21.394363	2020-12-08 16:18:47.599597
2	2	us-east-1	us-east-1a	host-6kMPE	cpu_utilization	17.144579	2020-12-08 16:18:47.599597
4	4	us-east-1	us-east-1a	host-sxj7X	cpu_utilization	26.584865	2020-12-08 16:18:47.599597
6	6	us-east-1	us-east-1a	host-ExOui	cpu_utilization	52.930970	2020-12-08 16:18:47.599597
8	8	us-east-1	us-east-1a	host-Bwb3j	cpu_utilization	99.134110	2020-12-08 16:18:47.599597
...	...	...	...	...	...	...	...
125990	125990	eu-north-1	eu-north-1c	host-aPtc6	cpu_utilization	89.566125	2020-12-08 16:18:47.599597
125992	125992	eu-north-1	eu-north-1c	host-7ZF9L	cpu_utilization	75.510598	2020-12-08 16:18:47.599597
125994	125994	eu-north-1	eu-north-1c	host-De8RB	cpu_utilization	2.771261	2020-12-08 16:18:47.599597
125997	125997	eu-north-1	eu-north-1c	host-2z8tn	cpu_utilization	29.212219	2020-12-08 16:18:47.599597
125999	125999	eu-north-1	eu-north-1c	host-9FczW	cpu_utilization	1.677793	2020-12-08 16:18:47.599597

63000 rows × 7 columns

rejected_records = wr.timestream.write(
    df=df_cpu,
    database="sampleDB",
    table="sampleTable",
    time_col="time",
    measure_col="cpu_utilization",
    dimensions_cols=["index", "region", "az", "hostname"],
)

assert len(rejected_records) == 0

Write MEMORY_UTILIZATION records

df_memory = df[df.measure_kind == "memory_utilization"].copy()
df_memory.rename(columns={"measure": "memory_utilization"}, inplace=True)

df_memory

	index	region	az	hostname	measure_kind	memory_utilization	time
1	1	us-east-1	us-east-1a	host-fj2hx	memory_utilization	68.563420	2020-12-08 16:18:47.599597
3	3	us-east-1	us-east-1a	host-6kMPE	memory_utilization	73.507870	2020-12-08 16:18:47.599597
5	5	us-east-1	us-east-1a	host-sxj7X	memory_utilization	22.401424	2020-12-08 16:18:47.599597
7	7	us-east-1	us-east-1a	host-ExOui	memory_utilization	45.440135	2020-12-08 16:18:47.599597
9	9	us-east-1	us-east-1a	host-Bwb3j	memory_utilization	15.042701	2020-12-08 16:18:47.599597
...	...	...	...	...	...	...	...
125991	125991	eu-north-1	eu-north-1c	host-aPtc6	memory_utilization	75.686739	2020-12-08 16:18:47.599597
125993	125993	eu-north-1	eu-north-1c	host-7ZF9L	memory_utilization	18.386152	2020-12-08 16:18:47.599597
125995	125995	eu-north-1	eu-north-1c	host-De8RB	memory_utilization	68.063468	2020-12-08 16:18:47.599597
125996	125996	eu-north-1	eu-north-1c	host-2z8tn	memory_utilization	72.203680	2020-12-08 16:18:47.599597
125998	125998	eu-north-1	eu-north-1c	host-9FczW	memory_utilization	71.746134	2020-12-08 16:18:47.599597

63000 rows × 7 columns

rejected_records = wr.timestream.write(
    df=df_memory,
    database="sampleDB",
    table="sampleTable",
    time_col="time",
    measure_col="memory_utilization",
    dimensions_cols=["index", "region", "az", "hostname"],
)

assert len(rejected_records) == 0

Querying CPU_UTILIZATION

wr.timestream.query("""
    SELECT
        hostname, region, az, measure_name, measure_value::double, time
    FROM "sampleDB"."sampleTable"
    WHERE measure_name = 'cpu_utilization'
    ORDER BY time DESC
    LIMIT 10
""")

	hostname	region	az	measure_name	measure_value::double	time
0	host-OgvFx	us-west-1	us-west-1a	cpu_utilization	39.617911	2020-12-08 19:18:47.600
1	host-rZUNx	eu-north-1	eu-north-1a	cpu_utilization	30.793332	2020-12-08 19:18:47.600
2	host-t1kAB	us-east-2	us-east-2b	cpu_utilization	74.453239	2020-12-08 19:18:47.600
3	host-RdQRf	us-east-1	us-east-1c	cpu_utilization	76.984448	2020-12-08 19:18:47.600
4	host-4Llhu	us-east-1	us-east-1c	cpu_utilization	41.862733	2020-12-08 19:18:47.600
5	host-2plqa	us-west-1	us-west-1a	cpu_utilization	34.864762	2020-12-08 19:18:47.600
6	host-J3Q4z	us-east-1	us-east-1b	cpu_utilization	71.574266	2020-12-08 19:18:47.600
7	host-VIR5T	ap-east-1	ap-east-1a	cpu_utilization	14.017491	2020-12-08 19:18:47.600
8	host-G042D	us-east-1	us-east-1c	cpu_utilization	60.199068	2020-12-08 19:18:47.600
9	host-8EBHm	us-west-2	us-west-2c	cpu_utilization	96.631624	2020-12-08 19:18:47.600

Querying MEMORY_UTILIZATION

wr.timestream.query("""
    SELECT
        hostname, region, az, measure_name, measure_value::double, time
    FROM "sampleDB"."sampleTable"
    WHERE measure_name = 'memory_utilization'
    ORDER BY time DESC
    LIMIT 10
""")

	hostname	region	az	measure_name	measure_value::double	time
0	host-7c897	us-west-2	us-west-2b	memory_utilization	63.427726	2020-12-08 19:18:47.600
1	host-2z8tn	eu-north-1	eu-north-1c	memory_utilization	41.071368	2020-12-08 19:18:47.600
2	host-J3Q4z	us-east-1	us-east-1b	memory_utilization	23.944388	2020-12-08 19:18:47.600
3	host-mjrQb	us-east-1	us-east-1b	memory_utilization	69.173431	2020-12-08 19:18:47.600
4	host-AyWSI	us-east-1	us-east-1c	memory_utilization	75.591467	2020-12-08 19:18:47.600
5	host-Axf0g	us-west-2	us-west-2a	memory_utilization	29.720739	2020-12-08 19:18:47.600
6	host-ilMBa	us-east-2	us-east-2b	memory_utilization	71.544134	2020-12-08 19:18:47.600
7	host-CWdXX	us-west-2	us-west-2c	memory_utilization	79.792799	2020-12-08 19:18:47.600
8	host-8EBHm	us-west-2	us-west-2c	memory_utilization	66.082554	2020-12-08 19:18:47.600
9	host-dRIJj	us-east-1	us-east-1c	memory_utilization	86.748960	2020-12-08 19:18:47.600

Deleting resources

wr.timestream.delete_table("sampleDB", "sampleTable")
wr.timestream.delete_database("sampleDB")

28 - Amazon DynamoDB

Writing Data

import awswrangler as wr
import pandas as pd
from pathlib import Path

Writing DataFrame

df = pd.DataFrame({
    "key": [1, 2],
    "value": ["foo", "boo"]
})
wr.dynamodb.put_df(df=df, table_name="table")

Writing CSV file

filepath = Path("items.csv")
df.to_csv(filepath, index=False)
wr.dynamodb.put_csv(path=filepath, table_name="table")
filepath.unlink()

Writing JSON files

filepath = Path("items.json")
df.to_json(filepath, orient="records")
wr.dynamodb.put_json(path="items.json", table_name="table")
filepath.unlink()

Writing list of items

items = df.to_dict(orient="records")
wr.dynamodb.put_items(items=items, table_name="table")

Deleting items

wr.dynamodb.delete_items(items=items, table_name="table")

29 - S3 Select

AWS Data Wrangler supports Amazon S3 Select, enabling applications to use SQL statements in order to query and filter the contents of a single S3 object. It works on objects stored in CSV, JSON or Apache Parquet, including compressed and large files of several TBs.

With S3 Select, the query workload is delegated to Amazon S3, leading to lower latency and cost, and to higher performance (up to 400% improvement). This is in comparison with other Wrangler operations such as read_parquet where the S3 object is downloaded and filtered on the client-side.

This feature has a number of limitations however, and should be used for specific scenarios only: * It operates on a single S3 object * The maximum length of a record in the input or result is 1 MB * The maximum uncompressed row group size is 256 MB (Parquet only) * It can only emit nested data in JSON format * Certain SQL operations are not supported (e.g. ORDER BY)

Read full CSV file

import awswrangler as wr

df = wr.s3.select_query(
    sql="SELECT * FROM s3object",
    path="s3://nyc-tlc/trip data/fhv_tripdata_2019-09.csv",  # 58 MB
    input_serialization="CSV",
    input_serialization_params={
        "FileHeaderInfo": "Use",
        "RecordDelimiter": "\r\n",
    },
    use_threads=True,
)
df.head()

	dispatching_base_num	pickup_datetime	dropoff_datetime	PULocationID	DOLocationID	SR_Flag
0	B00009	2019-09-01 00:35:00	2019-09-01 00:59:00	264	264
1	B00009	2019-09-01 00:48:00	2019-09-01 01:09:00	264	264
2	B00014	2019-09-01 00:16:18	2019-09-02 00:35:37	264	264
3	B00014	2019-09-01 00:55:03	2019-09-01 01:09:35	264	264
4	B00014	2019-09-01 00:13:08	2019-09-02 01:12:31	264	264

Filter JSON file

wr.s3.select_query(
    sql="SELECT * FROM s3object[*] s where s.\"family_name\" = \'Biden\'",
    path="s3://awsglue-datasets/examples/us-legislators/all/persons.json",
    input_serialization="JSON",
    input_serialization_params={
        "Type": "Document",
    },
)

	family_name	contact_details	name	links	gender	image	identifiers	other_names	sort_name	images	given_name	birth_date	id
0	Biden	[{'type': 'twitter', 'value': 'joebiden'}]	Joseph Biden, Jr.	[{'note': 'Wikipedia (ace)', 'url': 'https://a...	male	https://theunitedstates.io/images/congress/ori...	[{'scheme': 'bioguide', 'identifier': 'B000444...	[{'note': 'alternate', 'name': 'Joe Biden'}, {...	Biden, Joseph	[{'url': 'https://theunitedstates.io/images/co...	Joseph	1942-11-20	64239edf-8e06-4d2d-acc0-33d96bc79774

Read Snappy compressed Parquet

df = wr.s3.select_query(
        sql="SELECT * FROM s3object s where s.\"star_rating\" >= 5",
        path="s3://amazon-reviews-pds/parquet/product_category=Gift_Card/part-00000-495c48e6-96d6-4650-aa65-3c36a3516ddd.c000.snappy.parquet",
        input_serialization="Parquet",
        input_serialization_params={},
        use_threads=True,
)
df.loc[:, df.columns != "product_title"].head()

	marketplace	customer_id	review_id	product_id	product_parent	star_rating	helpful_votes	total_votes	vine	verified_purchase	review_headline	review_body	review_date	year
0	US	52670295	RGPOFKORD8RTU	B0002CZPPG	867256265	5	105	107	N	N	Excellent Gift Idea	I wonder if the other reviewer actually read t...	2005-02-08	2005
1	US	29964102	R2U8X8V5KPB4J3	B00H5BMF00	373287760	5	0	0	N	Y	Five Stars	convenience is the name of the game.	2015-05-03	2015
2	US	25173351	R15XV3LXUMLTXL	B00PG40CO4	137115061	5	0	0	N	Y	Birthday Gift	This gift card was handled with accuracy in de...	2015-05-03	2015
3	US	12516181	R3G6G7H8TX4H0T	B0002CZPPG	867256265	5	6	6	N	N	Love 'em.	Gotta love these iTunes Prepaid Card thingys. ...	2005-10-15	2005
4	US	38355314	R2NJ7WNBU16YTQ	B00B2TFSO6	89375983	5	0	0	N	Y	Five Stars	perfect	2015-05-03	2015

30 - Data Api

The Data Api simplifies access to Amazon Redshift and RDS by removing the need to manage database connections and credentials. Instead, you can execute SQL commands to an Amazon Redshift cluster or Amazon Aurora cluster by simply invoking an HTTPS API endpoint provided by the Data API. It takes care of managing database connections and returning data. Since the Data API leverages IAM user credentials or database credentials stored in AWS Secrets Manager, you don’t need to pass credentials in API calls.

Connect to the cluster

• wr.data_api.redshift.connect()

• wr.data_api.rds.connect()

con_redshift = wr.data_api.redshift.connect(
    cluster_id="aws-data-wrangler-1xn5lqxrdxrv3",
    database="test_redshift",
    secret_arn="arn:aws:secretsmanager:us-east-1:111111111111:secret:aws-data-wrangler/redshift-ewn43d"
)

con_mysql = wr.data_api.rds.connect(
    cluster_id="arn:aws:rds:us-east-1:111111111111:cluster:mysql-serverless-cluster-wrangler",
    database="test_rds",
    secret_arn="arn:aws:secretsmanager:us-east-1:111111111111:secret:aws-data-wrangler/mysql-23df3"
)

Read from database

• wr.data_api.redshift.read_sql_query()

• wr.data_api.rds.read_sql_query()

df = wr.data_api.redshift.read_sql_query(
    sql="SELECT * FROM public.test_table",
    con=con_redshift,
)

df = wr.data_api.rds.read_sql_query(
    sql="SELECT * FROM test.test_table",
    con=con_rds,
)

31 - OpenSearch

Table of Contents

• 1. Initialize

  • Connect to your Amazon OpenSearch domain

  • Enter your bucket name

  • Initialize sample data

• 2. Indexing (load)

  • Index documents (no Pandas)

  • Index json file

  • Index CSV

• 3. Search

  • 3.1 Search by DSL

  • 3.2 Search by SQL

• 4. Delete Indices

• 5. Bonus - Prepare data and index from DataFrame

  • Prepare the data for indexing

  • Create index with mapping

  • Index dataframe

  • Execute geo query

1. Initialize

import awswrangler as wr

Connect to your Amazon OpenSearch domain

client = wr.opensearch.connect(
    host='OPENSEARCH-ENDPOINT',
#     username='FGAC-USERNAME(OPTIONAL)',
#     password='FGAC-PASSWORD(OPTIONAL)'
)
client.info()

Enter your bucket name

bucket = 'BUCKET'

Initialize sample data

sf_restaurants_inspections = [
    {
        "inspection_id": "24936_20160609",
        "business_address": "315 California St",
        "business_city": "San Francisco",
        "business_id": "24936",
        "business_location": {"lon": -122.400152, "lat": 37.793199},
        "business_name": "San Francisco Soup Company",
        "business_postal_code": "94104",
        "business_state": "CA",
        "inspection_date": "2016-06-09T00:00:00.000",
        "inspection_score": 77,
        "inspection_type": "Routine - Unscheduled",
        "risk_category": "Low Risk",
        "violation_description": "Improper food labeling or menu misrepresentation",
        "violation_id": "24936_20160609_103141",
    },
    {
        "inspection_id": "60354_20161123",
        "business_address": "10 Mason St",
        "business_city": "San Francisco",
        "business_id": "60354",
        "business_location": {"lon": -122.409061, "lat": 37.783527},
        "business_name": "Soup Unlimited",
        "business_postal_code": "94102",
        "business_state": "CA",
        "inspection_date": "2016-11-23T00:00:00.000",
        "inspection_type": "Routine",
        "inspection_score": 95,
    },
    {
        "inspection_id": "1797_20160705",
        "business_address": "2872 24th St",
        "business_city": "San Francisco",
        "business_id": "1797",
        "business_location": {"lon": -122.409752, "lat": 37.752807},
        "business_name": "TIO CHILOS GRILL",
        "business_postal_code": "94110",
        "business_state": "CA",
        "inspection_date": "2016-07-05T00:00:00.000",
        "inspection_score": 90,
        "inspection_type": "Routine - Unscheduled",
        "risk_category": "Low Risk",
        "violation_description": "Unclean nonfood contact surfaces",
        "violation_id": "1797_20160705_103142",
    },
    {
        "inspection_id": "66198_20160527",
        "business_address": "1661 Tennessee St Suite 3B",
        "business_city": "San Francisco Whard Restaurant",
        "business_id": "66198",
        "business_location": {"lon": -122.388478, "lat": 37.75072},
        "business_name": "San Francisco Restaurant",
        "business_postal_code": "94107",
        "business_state": "CA",
        "inspection_date": "2016-05-27T00:00:00.000",
        "inspection_type": "Routine",
        "inspection_score": 56,
    },
    {
        "inspection_id": "5794_20160907",
        "business_address": "2162 24th Ave",
        "business_city": "San Francisco",
        "business_id": "5794",
        "business_location": {"lon": -122.481299, "lat": 37.747228},
        "business_name": "Soup House",
        "business_phone_number": "+14155752700",
        "business_postal_code": "94116",
        "business_state": "CA",
        "inspection_date": "2016-09-07T00:00:00.000",
        "inspection_score": 96,
        "inspection_type": "Routine - Unscheduled",
        "risk_category": "Low Risk",
        "violation_description": "Unapproved or unmaintained equipment or utensils",
        "violation_id": "5794_20160907_103144",
    },

    # duplicate record
    {
        "inspection_id": "5794_20160907",
        "business_address": "2162 24th Ave",
        "business_city": "San Francisco",
        "business_id": "5794",
        "business_location": {"lon": -122.481299, "lat": 37.747228},
        "business_name": "Soup-or-Salad",
        "business_phone_number": "+14155752700",
        "business_postal_code": "94116",
        "business_state": "CA",
        "inspection_date": "2016-09-07T00:00:00.000",
        "inspection_score": 96,
        "inspection_type": "Routine - Unscheduled",
        "risk_category": "Low Risk",
        "violation_description": "Unapproved or unmaintained equipment or utensils",
        "violation_id": "5794_20160907_103144",
    },
]

2. Indexing (load)

Index documents (no Pandas)

# index documents w/o providing keys (_id is auto-generated)
wr.opensearch.index_documents(
        client,
        documents=sf_restaurants_inspections,
        index="sf_restaurants_inspections"
)

Indexing: 100% (6/6)|####################################|Elapsed Time: 0:00:01

{'success': 6, 'errors': []}

# read all documents. There are total 6 documents
wr.opensearch.search(
        client,
        index="sf_restaurants_inspections",
        _source=["inspection_id", "business_name", "business_location"]
)

	_id	business_name	inspection_id	business_location.lon	business_location.lat
0	663dd72d-0da4-495b-b0ae-ed000105ae73	TIO CHILOS GRILL	1797_20160705	-122.409752	37.752807
1	ff2f50f6-5415-4706-9bcb-af7c5eb0afa3	Soup House	5794_20160907	-122.481299	37.747228
2	b9e8f6a2-8fd1-4660-b041-2997a1a80984	San Francisco Soup Company	24936_20160609	-122.400152	37.793199
3	56b352e6-102b-4eff-8296-7e1fb2459bab	Soup Unlimited	60354_20161123	-122.409061	37.783527
4	6fec5411-f79a-48e4-be7b-e0e44d5ebbab	San Francisco Restaurant	66198_20160527	-122.388478	37.750720
5	7ba4fb17-f9a9-49da-b90e-8b3553d6d97c	Soup-or-Salad	5794_20160907	-122.481299	37.747228

Index json file

import pandas as pd
df = pd.DataFrame(sf_restaurants_inspections)
path = f"s3://{bucket}/json/sf_restaurants_inspections.json"
wr.s3.to_json(df, path,orient='records',lines=True)

# index json w/ providing keys
wr.opensearch.index_json(
        client,
        path=path, # path can be s3 or local
        index="sf_restaurants_inspections_dedup",
        id_keys=["inspection_id"] # can be multiple fields. arg applicable to all index_* functions
)

Indexing: 100% (6/6)|####################################|Elapsed Time: 0:00:00

{'success': 6, 'errors': []}

# now there are no duplicates. There are total 5 documents
wr.opensearch.search(
        client,
        index="sf_restaurants_inspections_dedup",
        _source=["inspection_id", "business_name", "business_location"]
    )

	_id	business_name	inspection_id	business_location.lon	business_location.lat
0	24936_20160609	San Francisco Soup Company	24936_20160609	-122.400152	37.793199
1	66198_20160527	San Francisco Restaurant	66198_20160527	-122.388478	37.750720
2	5794_20160907	Soup-or-Salad	5794_20160907	-122.481299	37.747228
3	60354_20161123	Soup Unlimited	60354_20161123	-122.409061	37.783527
4	1797_20160705	TIO CHILOS GRILL	1797_20160705	-122.409752	37.752807

Index CSV

wr.opensearch.index_csv(
        client,
        index="nyc_restaurants_inspections_sample",
        path='https://data.cityofnewyork.us/api/views/43nn-pn8j/rows.csv?accessType=DOWNLOAD', # index_csv supports local, s3 and url path
        id_keys=["CAMIS"],
        pandas_kwargs={'na_filter': True, 'nrows': 1000},  # pandas.read_csv() args - https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_csv.html
        bulk_size=500 # modify based on your cluster size
)

Indexing: 100% (1000/1000)|##############################|Elapsed Time: 0:00:00

{'success': 1000, 'errors': []}

wr.opensearch.search(
        client,
        index="nyc_restaurants_inspections_sample",
        size=5
)

	_id	CAMIS	DBA	BORO	BUILDING	STREET	ZIPCODE	PHONE	CUISINE DESCRIPTION	INSPECTION DATE	...	RECORD DATE	INSPECTION TYPE	Latitude	Longitude	Community Board	Council District	Census Tract	BIN	BBL	NTA
0	41610426	41610426	GLOW THAI RESTAURANT	Brooklyn	7107	3 AVENUE	11209.0	7187481920	Thai	02/26/2020	...	10/04/2021	Cycle Inspection / Re-inspection	40.633865	-74.026798	310.0	43.0	6800.0	3146519.0	3.058910e+09	BK31
1	40811162	40811162	CARMINE'S	Manhattan	2450	BROADWAY	10024.0	2123622200	Italian	05/28/2019	...	10/04/2021	Cycle Inspection / Initial Inspection	40.791168	-73.974308	107.0	6.0	17900.0	1033560.0	1.012380e+09	MN12
2	50012113	50012113	TANG	Queens	196-50	NORTHERN BOULEVARD	11358.0	7182797080	Korean	08/16/2018	...	10/04/2021	Cycle Inspection / Initial Inspection	40.757850	-73.784593	411.0	19.0	145101.0	4124565.0	4.055200e+09	QN48
3	50014618	50014618	TOTTO RAMEN	Manhattan	248	EAST 52 STREET	10022.0	2124210052	Japanese	08/20/2018	...	10/04/2021	Cycle Inspection / Re-inspection	40.756596	-73.968749	106.0	4.0	9800.0	1038490.0	1.013250e+09	MN19
4	50045782	50045782	OLLIE'S CHINESE RESTAURANT	Manhattan	2705	BROADWAY	10025.0	2129323300	Chinese	10/21/2019	...	10/04/2021	Cycle Inspection / Re-inspection	40.799318	-73.968440	107.0	6.0	19100.0	1056562.0	1.018750e+09	MN12

5 rows × 27 columns

3. Search

3.1 Search by DSL

# add a search query. search all soup businesses
wr.opensearch.search(
        client,
        index="sf_restaurants_inspections",
        _source=["inspection_id", "business_name", "business_location"],
        filter_path=["hits.hits._id","hits.hits._source"],
        search_body={
            "query": {
                "match": {
                    "business_name": "soup"
                }
            }
        }
    )

	_id	business_name	inspection_id	business_location.lon	business_location.lat
0	ff2f50f6-5415-4706-9bcb-af7c5eb0afa3	Soup House	5794_20160907	-122.481299	37.747228
1	7ba4fb17-f9a9-49da-b90e-8b3553d6d97c	Soup-or-Salad	5794_20160907	-122.481299	37.747228
2	b9e8f6a2-8fd1-4660-b041-2997a1a80984	San Francisco Soup Company	24936_20160609	-122.400152	37.793199
3	56b352e6-102b-4eff-8296-7e1fb2459bab	Soup Unlimited	60354_20161123	-122.409061	37.783527

3.1 Search by SQL

wr.opensearch.search_by_sql(
        client,
        sql_query="""SELECT business_name, inspection_score
                    FROM sf_restaurants_inspections_dedup
                    WHERE business_name LIKE '%soup%'
                    ORDER BY inspection_score DESC LIMIT 5"""
)

	_index	_type	_id	_score	business_name	inspection_score
0	sf_restaurants_inspections_dedup	_doc	5794_20160907	None	Soup-or-Salad	96
1	sf_restaurants_inspections_dedup	_doc	60354_20161123	None	Soup Unlimited	95
2	sf_restaurants_inspections_dedup	_doc	24936_20160609	None	San Francisco Soup Company	77

4. Delete Indices

wr.opensearch.delete_index(
     client=client,
     index="sf_restaurants_inspections"
)

{'acknowledged': True}

5. Bonus - Prepare data and index from DataFrame

For this exercise we’ll use DOHMH New York City Restaurant Inspection Results dataset

import pandas as pd

df = pd.read_csv('https://data.cityofnewyork.us/api/views/43nn-pn8j/rows.csv?accessType=DOWNLOAD')

Prepare the data for indexing

# fields names underscore casing
df.columns = [col.lower().replace(' ', '_') for col in df.columns]

# convert lon/lat to OpenSearch geo_point
df['business_location'] = "POINT (" + df.longitude.fillna('0').astype(str) + " " + df.latitude.fillna('0').astype(str) + ")"

Create index with mapping

# delete index if exists
wr.opensearch.delete_index(
    client=client,
    index="nyc_restaurants"

)

# use dynamic_template to map date fields
# define business_location as geo_point
wr.opensearch.create_index(
    client=client,
    index="nyc_restaurants_inspections",
    mappings={
         "dynamic_templates" : [
            {
                "dates" : {
                   "match" : "*date",
                    "mapping" : {
                        "type" : "date",
                        "format" : 'MM/dd/yyyy'
                    }
                }
            }
        ],
         "properties": {
          "business_location": {
            "type": "geo_point"
          }
        }
    }
)

{'acknowledged': True,
 'shards_acknowledged': True,
 'index': 'nyc_restaurants_inspections'}

Index dataframe

wr.opensearch.index_df(
    client,
    df=df,
    index="nyc_restaurants_inspections",
    id_keys=["camis"],
    bulk_size=1000
)

Indexing: 100% (382655/382655)|##########################|Elapsed Time: 0:04:15

{'success': 382655, 'errors': []}

Execute geo query

Sort restaurants by distance from Times-Square

wr.opensearch.search(
    client,
    index="nyc_restaurants_inspections",
    filter_path=["hits.hits._source"],
    size=100,
    search_body={
        "query": {
            "match_all": {}
        },
          "sort": [
            {
              "_geo_distance": {
                "business_location": { # Times-Square - https://geojson.io/#map=16/40.7563/-73.9862
                  "lat":  40.75613228383523,
                  "lon": -73.9865791797638
                },
                "order": "asc"
              }
            }
        ]
    }
)

	camis	dba	boro	building	street	zipcode	phone	cuisine_description	inspection_date	action	...	inspection_type	latitude	longitude	community_board	council_district	census_tract	bin	bbl	nta	business_location
0	41551304	THE COUNTER	Manhattan	7	TIMES SQUARE	10036.0	2129976801	American	12/22/2016	Violations were cited in the following area(s).	...	Cycle Inspection / Initial Inspection	40.755908	-73.986681	105.0	3.0	11300.0	1086069.0	1.009940e+09	MN17	POINT (-73.986680953809 40.755907817312)
1	50055665	ANN INC CAFE	Manhattan	7	TIMES SQUARE	10036.0	2125413287	American	12/11/2019	Violations were cited in the following area(s).	...	Cycle Inspection / Initial Inspection	40.755908	-73.986681	105.0	3.0	11300.0	1086069.0	1.009940e+09	MN17	POINT (-73.986680953809 40.755907817312)
2	50049552	ERNST AND YOUNG	Manhattan	5	TIMES SQ	10036.0	2127739994	Coffee/Tea	11/30/2018	Violations were cited in the following area(s).	...	Cycle Inspection / Initial Inspection	40.755702	-73.987208	105.0	3.0	11300.0	1024656.0	1.010130e+09	MN17	POINT (-73.987207980138 40.755702020307)
3	50014078	RED LOBSTER	Manhattan	5	TIMES SQ	10036.0	2127306706	Seafood	10/03/2017	Violations were cited in the following area(s).	...	Cycle Inspection / Initial Inspection	40.755702	-73.987208	105.0	3.0	11300.0	1024656.0	1.010130e+09	MN17	POINT (-73.987207980138 40.755702020307)
4	50015171	NEW AMSTERDAM THEATER	Manhattan	214	WEST 42 STREET	10036.0	2125825472	American	06/26/2018	Violations were cited in the following area(s).	...	Cycle Inspection / Re-inspection	40.756317	-73.987652	105.0	3.0	11300.0	1024660.0	1.010130e+09	MN17	POINT (-73.987651832547 40.756316895053)
...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...
95	41552060	PROSKAUER ROSE	Manhattan	11	TIMES SQUARE	10036.0	2129695493	American	08/11/2017	Violations were cited in the following area(s).	...	Administrative Miscellaneous / Initial Inspection	40.756891	-73.990023	105.0	3.0	11300.0	1087978.0	1.010138e+09	MN17	POINT (-73.990023200823 40.756890780426)
96	41242148	GABBY O'HARA'S	Manhattan	123	WEST 39 STREET	10018.0	2122788984	Irish	07/30/2019	Violations were cited in the following area(s).	...	Cycle Inspection / Re-inspection	40.753405	-73.986602	105.0	4.0	11300.0	1080611.0	1.008150e+09	MN17	POINT (-73.986602050292 40.753404587174)
97	50095860	THE TIMES EATERY	Manhattan	680	8 AVENUE	10036.0	6463867787	American	02/28/2020	Violations were cited in the following area(s).	...	Pre-permit (Operational) / Initial Inspection	40.757991	-73.989218	105.0	3.0	11900.0	1024703.0	1.010150e+09	MN17	POINT (-73.989218092096 40.757991356019)
98	50072861	ITSU	Manhattan	530	7 AVENUE	10018.0	9176393645	Asian/Asian Fusion	09/10/2018	Violations were cited in the following area(s).	...	Pre-permit (Operational) / Initial Inspection	40.753844	-73.988551	105.0	3.0	11300.0	1014485.0	1.007880e+09	MN17	POINT (-73.988551029682 40.753843959794)
99	50068109	LUKE'S LOBSTER	Manhattan	1407	BROADWAY	10018.0	9174759192	Seafood	09/06/2017	Violations were cited in the following area(s).	...	Pre-permit (Operational) / Initial Inspection	40.753432	-73.987151	105.0	3.0	11300.0	1015265.0	1.008140e+09	MN17	POINT (-73.98715066791 40.753432097521)

100 rows × 27 columns

32 - AWS Lake Formation - Glue Governed tables

This tutorial assumes that your IAM user/role has the required Lake Formation permissions to create and read AWS Glue Governed tables

Table of Contents

• 1. Read Governed table

  • 1.1 Read PartiQL query

    • 1.1.1 Read within transaction

    • 1.1.2 Read within query as of time

  • 1.2 Read full table

• 2. Write Governed table

  • 2.1 Create new Governed table

    • 2.1.1 CSV table

    • 2.1.2 Parquet table

  • 2.2 Overwrite operations

    • 2.2.1 Overwrite

    • 2.2.2 Append

    • 2.2.3 Create partitioned Governed table

    • 2.2.4 Overwrite partitions

• 3. Multiple read/write operations within a transaction

1. Read Governed table

import awswrangler as wr

database = "gov_db"  # Assumes a Glue database registered with Lake Formation exists in the account
table = "gov_table"  # Assumes a Governed table exists in the account
catalog_id = "111111111111"  # AWS Account Id

# Note 1: If a transaction_id is not specified, a new transaction is started
df = wr.lakeformation.read_sql_query(
    sql=f"SELECT * FROM {table};",
    database=database,
    catalog_id=catalog_id
)

1.1.1 Read within transaction

transaction_id = wr.lakeformation.start_transaction(read_only=True)
df = wr.lakeformation.read_sql_query(
    sql=f"SELECT * FROM {table};",
    database=database,
    transaction_id=transaction_id
)

1.1.2 Read within query as of time

import calendar
import time

query_as_of_time = query_as_of_time = calendar.timegm(time.gmtime())
df = wr.lakeformation.read_sql_query(
    sql=f"SELECT * FROM {table} WHERE id=:id; AND name=:name;",
    database=database,
    query_as_of_time=query_as_of_time,
    params={"id": 1, "name": "Ayoub"}
)

1.2 Read full table

df = wr.lakeformation.read_sql_table(
    table=table,
    database=database
)

2. Write Governed table

import getpass

bucket = getpass.getpass()

If a governed table does not exist, it can be created by passing an S3 path argument. Make sure your IAM user/role has enough permissions in the Lake Formation database

2.1.1 CSV table

import pandas as pd

table = "gov_table_csv"

df=pd.DataFrame({
    "col": [1, 2, 3],
    "col2": ["A", "A", "B"],
    "col3": [None, "test", None]
})
# Note 1: If a transaction_id is not specified, a new transaction is started
# Note 2: When creating a new Governed table, `table_type="GOVERNED"` must be specified. Otherwise the default is to create an EXTERNAL_TABLE
wr.s3.to_csv(
    df=df,
    path=f"s3://{bucket}/{database}/{table}/",  # S3 path
    dataset=True,
    database=database,
    table=table,
    table_type="GOVERNED"
)

2.1.2 Parquet table

table = "gov_table_parquet"

df = pd.DataFrame({"c0": [0, None]}, dtype="Int64")
wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/{database}/{table}/",
    dataset=True,
    database=database,
    table=table,
    table_type="GOVERNED",
    description="c0",
    parameters={"num_cols": str(len(df.columns)), "num_rows": str(len(df.index))},
    columns_comments={"c0": "0"}
)

2.2 Overwrite operations

2.2.1 Overwrite

df = pd.DataFrame({"c1": [None, 1, None]}, dtype="Int16")
wr.s3.to_parquet(
    df=df,
    dataset=True,
    mode="overwrite",
    database=database,
    table=table,
    description="c1",
    parameters={"num_cols": str(len(df.columns)), "num_rows": str(len(df.index))},
    columns_comments={"c1": "1"}
)

2.2.2 Append

df = pd.DataFrame({"c1": [None, 2, None]}, dtype="Int8")
wr.s3.to_parquet(
    df=df,
    dataset=True,
    mode="append",
    database=database,
    table=table,
    description="c1",
    parameters={"num_cols": str(len(df.columns)), "num_rows": str(len(df.index) * 2)},
    columns_comments={"c1": "1"}
)

2.2.3 Create partitioned Governed table

table = "gov_table_parquet_partitioned"

df = pd.DataFrame({"c0": ["foo", None], "c1": [0, 1]})
wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/{database}/{table}/",
    dataset=True,
    database=database,
    table=table,
    table_type="GOVERNED",
    partition_cols=["c1"],
    description="c0+c1",
    parameters={"num_cols": "2", "num_rows": "2"},
    columns_comments={"c0": "zero", "c1": "one"}
)

2.2.4 Overwrite partitions

df = pd.DataFrame({"c0": [None, None], "c1": [0, 2]})
wr.s3.to_parquet(
    df=df,
    dataset=True,
    mode="overwrite_partitions",
    database=database,
    table=table,
    partition_cols=["c1"],
    description="c0+c1",
    parameters={"num_cols": "2", "num_rows": "3"},
    columns_comments={"c0": "zero", "c1": "one"}
)

3. Multiple read/write operations within a transaction

read_table = "gov_table_parquet"
write_table = "gov_table_multi_parquet"

transaction_id = wr.lakeformation.start_transaction(read_only=False)

df = pd.DataFrame({"c0": [0, None]}, dtype="Int64")
wr.s3.to_parquet(
    df=df,
    path=f"s3://{bucket}/{database}/{write_table}_1",
    dataset=True,
    database=database,
    table=f"{write_table}_1",
    table_type="GOVERNED",
    transaction_id=transaction_id,
)

df2 = wr.lakeformation.read_sql_table(
    table=read_table,
    database=database,
    transaction_id=transaction_id,
    use_threads=True
)

df3 = pd.DataFrame({"c1": [None, 1, None]}, dtype="Int16")
wr.s3.to_parquet(
    df=df2,
    path=f"s3://{bucket}/{database}/{write_table}_2",
    dataset=True,
    mode="append",
    database=database,
    table=f"{write_table}_2",
    table_type="GOVERNED",
    transaction_id=transaction_id,
)

wr.lakeformation.commit_transaction(transaction_id=transaction_id)

API Reference

• Amazon S3

• AWS Glue Catalog

• Amazon Athena

• AWS Lake Formation

• Amazon Redshift

• PostgreSQL

• MySQL

• Microsoft SQL Server

• Data API Redshift

• Data API RDS

• OpenSearch

• DynamoDB

• Amazon Timestream

• Amazon EMR

• Amazon CloudWatch Logs

• Amazon QuickSight

• AWS STS

• AWS Secrets Manager

• Global Configurations

Amazon S3

copy_objects(paths, source_path, target_path)	Copy a list of S3 objects to another S3 directory.
delete_objects(path[, use_threads, ...])	Delete Amazon S3 objects from a received S3 prefix or list of S3 objects paths.
describe_objects(path[, version_id, ...])	Describe Amazon S3 objects from a received S3 prefix or list of S3 objects paths.
does_object_exist(path[, ...])	Check if object exists on S3.
download(path, local_file[, version_id, ...])	Download file from from a received S3 path to local file.
get_bucket_region(bucket[, boto3_session])	Get bucket region name.
list_buckets([boto3_session])	List Amazon S3 buckets.
list_directories(path[, chunked, ...])	List Amazon S3 objects from a prefix.
list_objects(path[, suffix, ignore_suffix, ...])	List Amazon S3 objects from a prefix.
merge_datasets(source_path, target_path[, ...])	Merge a source dataset into a target dataset.
merge_upsert_table(delta_df, database, ...)	Perform Upsert (Update else Insert) onto an existing Glue table.
read_csv(path[, path_suffix, ...])	Read CSV file(s) from from a received S3 prefix or list of S3 objects paths.
read_excel(path[, version_id, use_threads, ...])	Read EXCEL file(s) from from a received S3 path.
read_fwf(path[, path_suffix, ...])	Read fixed-width formatted file(s) from from a received S3 prefix or list of S3 objects paths.
read_json(path[, path_suffix, ...])	Read JSON file(s) from from a received S3 prefix or list of S3 objects paths.
read_parquet(path[, path_root, path_suffix, ...])	Read Apache Parquet file(s) from from a received S3 prefix or list of S3 objects paths.
read_parquet_metadata(path[, version_id, ...])	Read Apache Parquet file(s) metadata from from a received S3 prefix or list of S3 objects paths.
read_parquet_table(table, database[, ...])	Read Apache Parquet table registered on AWS Glue Catalog.
select_query(sql, path, input_serialization, ...)	Filter contents of an Amazon S3 object based on SQL statement.
size_objects(path[, version_id, ...])	Get the size (ContentLength) in bytes of Amazon S3 objects from a received S3 prefix or list of S3 objects paths.
store_parquet_metadata(path, database, table)	Infer and store parquet metadata on AWS Glue Catalog.
to_csv(df[, path, sep, index, columns, ...])	Write CSV file or dataset on Amazon S3.
to_excel(df, path[, boto3_session, ...])	Write EXCEL file on Amazon S3.
to_json(df[, path, index, columns, ...])	Write JSON file on Amazon S3.
to_parquet(df[, path, index, compression, ...])	Write Parquet file or dataset on Amazon S3.
upload(local_file, path[, use_threads, ...])	Upload file from a local file to received S3 path.
wait_objects_exist(paths[, delay, ...])	Wait Amazon S3 objects exist.
wait_objects_not_exist(paths[, delay, ...])	Wait Amazon S3 objects not exist.

AWS Glue Catalog

add_column(database, table, column_name[, ...])	Add a column in a AWS Glue Catalog table.
add_csv_partitions(database, table, ...[, ...])	Add partitions (metadata) to a CSV Table in the AWS Glue Catalog.
add_parquet_partitions(database, table, ...)	Add partitions (metadata) to a Parquet Table in the AWS Glue Catalog.
create_csv_table(database, table, path, ...)	Create a CSV Table (Metadata Only) in the AWS Glue Catalog.
create_database(name[, description, ...])	Create a database in AWS Glue Catalog.
create_parquet_table(database, table, path, ...)	Create a Parquet Table (Metadata Only) in the AWS Glue Catalog.
databases([limit, catalog_id, boto3_session])	Get a Pandas DataFrame with all listed databases.
delete_column(database, table, column_name)	Delete a column in a AWS Glue Catalog table.
delete_database(name[, catalog_id, ...])	Create a database in AWS Glue Catalog.
delete_partitions(table, database, ...[, ...])	Delete specified partitions in a AWS Glue Catalog table.
delete_all_partitions(table, database[, ...])	Delete all partitions in a AWS Glue Catalog table.
delete_table_if_exists(database, table[, ...])	Delete Glue table if exists.
does_table_exist(database, table[, ...])	Check if the table exists.
drop_duplicated_columns(df)	Drop all repeated columns (duplicated names).
extract_athena_types(df[, index, ...])	Extract columns and partitions types (Amazon Athena) from Pandas DataFrame.
get_columns_comments(database, table[, ...])	Get all columns comments.
get_csv_partitions(database, table[, ...])	Get all partitions from a Table in the AWS Glue Catalog.
get_databases([catalog_id, boto3_session])	Get an iterator of databases.
get_parquet_partitions(database, table[, ...])	Get all partitions from a Table in the AWS Glue Catalog.
get_partitions(database, table[, ...])	Get all partitions from a Table in the AWS Glue Catalog.
get_table_description(database, table[, ...])	Get table description.
get_table_location(database, table[, ...])	Get table's location on Glue catalog.
get_table_number_of_versions(database, table)	Get tatal number of versions.
get_table_parameters(database, table[, ...])	Get all parameters.
get_table_types(database, table[, ...])	Get all columns and types from a table.
get_table_versions(database, table[, ...])	Get all versions.
get_tables([catalog_id, database, ...])	Get an iterator of tables.
overwrite_table_parameters(parameters, ...)	Overwrite all existing parameters.
sanitize_column_name(column)	Convert the column name to be compatible with Amazon Athena.
sanitize_dataframe_columns_names(df)	Normalize all columns names to be compatible with Amazon Athena.
sanitize_table_name(table)	Convert the table name to be compatible with Amazon Athena.
search_tables(text[, catalog_id, boto3_session])	Get Pandas DataFrame of tables filtered by a search string.
table(database, table[, transaction_id, ...])	Get table details as Pandas DataFrame.
tables([limit, catalog_id, database, ...])	Get a DataFrame with tables filtered by a search term, prefix, suffix.
upsert_table_parameters(parameters, ...[, ...])	Insert or Update the received parameters.

Amazon Athena

create_athena_bucket([boto3_session])	Create the default Athena bucket if it doesn't exist.
get_query_columns_types(query_execution_id)	Get the data type of all columns queried.
get_query_execution(query_execution_id[, ...])	Fetch query execution details.
get_work_group(workgroup[, boto3_session])	Return information about the workgroup with the specified name.
read_sql_query(sql, database[, ...])	Execute any SQL query on AWS Athena and return the results as a Pandas DataFrame.
read_sql_table(table, database[, ...])	Extract the full table AWS Athena and return the results as a Pandas DataFrame.
repair_table(table[, database, s3_output, ...])	Run the Hive's metastore consistency check: 'MSCK REPAIR TABLE table;'.
start_query_execution(sql[, database, ...])	Start a SQL Query against AWS Athena.
stop_query_execution(query_execution_id[, ...])	Stop a query execution.
wait_query(query_execution_id[, boto3_session])	Wait for the query end.

AWS Lake Formation

read_sql_query(sql, database[, ...])	Execute PartiQL query on AWS Glue Table (Transaction ID or time travel timestamp).
read_sql_table(table, database[, ...])	Extract all rows from AWS Glue Table (Transaction ID or time travel timestamp).
cancel_transaction(transaction_id[, ...])	Cancel the specified transaction.
commit_transaction(transaction_id[, ...])	Commit the specified transaction.
describe_transaction(transaction_id[, ...])	Return the status of a single transaction.
extend_transaction(transaction_id[, ...])	Indicate to the service that the specified transaction is still active and should not be canceled.
start_transaction([read_only, time_out, ...])	Start a new transaction and returns its transaction ID.
wait_query(query_id[, boto3_session])	Wait for the query to end.

Amazon Redshift

connect([connection, secret_id, catalog_id, ...])	Return a redshift_connector connection from a Glue Catalog or Secret Manager.
connect_temp(cluster_identifier, user[, ...])	Return a redshift_connector temporary connection (No password required).
copy(df, path, con, table, schema[, ...])	Load Pandas DataFrame as a Table on Amazon Redshift using parquet files on S3 as stage.
copy_from_files(path, con, table, schema[, ...])	Load Parquet files from S3 to a Table on Amazon Redshift (Through COPY command).
read_sql_query(sql, con[, index_col, ...])	Return a DataFrame corresponding to the result set of the query string.
read_sql_table(table, con[, schema, ...])	Return a DataFrame corresponding the table.
to_sql(df, con, table, schema[, mode, ...])	Write records stored in a DataFrame into Redshift.
unload(sql, path, con[, iam_role, ...])	Load Pandas DataFrame from a Amazon Redshift query result using Parquet files on s3 as stage.
unload_to_files(sql, path, con[, iam_role, ...])	Unload Parquet files on s3 from a Redshift query result (Through the UNLOAD command).

PostgreSQL

connect([connection, secret_id, catalog_id, ...])	Return a pg8000 connection from a Glue Catalog Connection.
read_sql_query(sql, con[, index_col, ...])	Return a DataFrame corresponding to the result set of the query string.
read_sql_table(table, con[, schema, ...])	Return a DataFrame corresponding the table.
to_sql(df, con, table, schema[, mode, ...])	Write records stored in a DataFrame into PostgreSQL.

MySQL

connect([connection, secret_id, catalog_id, ...])	Return a pymysql connection from a Glue Catalog Connection or Secrets Manager.
read_sql_query(sql, con[, index_col, ...])	Return a DataFrame corresponding to the result set of the query string.
read_sql_table(table, con[, schema, ...])	Return a DataFrame corresponding the table.
to_sql(df, con, table, schema[, mode, ...])	Write records stored in a DataFrame into MySQL.

Microsoft SQL Server

connect([connection, secret_id, catalog_id, ...])	Return a pyodbc connection from a Glue Catalog Connection.
read_sql_query(sql, con[, index_col, ...])	Return a DataFrame corresponding to the result set of the query string.
read_sql_table(table, con[, schema, ...])	Return a DataFrame corresponding the table.
to_sql(df, con, table, schema[, mode, ...])	Write records stored in a DataFrame into Microsoft SQL Server.

Data API Redshift

RedshiftDataApi(cluster_id, database[, ...])	Provides access to a Redshift cluster via the Data API.
connect(cluster_id, database[, secret_arn, ...])	Create a Redshift Data API connection.
read_sql_query(sql, con[, database])	Run an SQL query on a RedshiftDataApi connection and return the result as a dataframe.

Data API RDS

RdsDataApi(resource_arn, database[, ...])	Provides access to the RDS Data API.
connect(resource_arn, database[, secret_arn])	Create a RDS Data API connection.
read_sql_query(sql, con[, database])	Run an SQL query on an RdsDataApi connection and return the result as a dataframe.

OpenSearch

connect(host[, port, boto3_session, region, ...])	Create a secure connection to the specified Amazon OpenSearch domain.
create_index(client, index[, doc_type, ...])	Create an index.
delete_index(client, index)	Create an index.
index_csv(client, path, index[, doc_type, ...])	Index all documents from a CSV file to OpenSearch index.
index_documents(client, documents, index[, ...])	Index all documents to OpenSearch index.
index_df(client, df, index[, doc_type])	Index all documents from a DataFrame to OpenSearch index.
index_json(client, path, index[, doc_type, ...])	Index all documents from JSON file to OpenSearch index.
search(client[, index, search_body, ...])	Return results matching query DSL as pandas dataframe.
search_by_sql(client, sql_query, **kwargs)	Return results matching SQL query as pandas dataframe.

DynamoDB

delete_items(items, table_name[, boto3_session])	Delete all items in the specified DynamoDB table.
get_table(table_name[, boto3_session])	Get DynamoDB table object for specified table name.
put_csv(path, table_name[, boto3_session])	Write all items from a CSV file to a DynamoDB.
put_df(df, table_name[, boto3_session])	Write all items from a DataFrame to a DynamoDB.
put_items(items, table_name[, boto3_session])	Insert all items to the specified DynamoDB table.
put_json(path, table_name[, boto3_session])	Write all items from JSON file to a DynamoDB.

Amazon Timestream

create_database(database[, kms_key_id, ...])	Create a new Timestream database.
create_table(database, table, ...[, tags, ...])	Create a new Timestream database.
delete_database(database[, boto3_session])	Delete a given Timestream database.
delete_table(database, table[, boto3_session])	Delete a given Timestream table.
query(sql[, chunked, pagination_config, ...])	Run a query and retrieve the result as a Pandas DataFrame.
write(df, database, table, time_col, ...[, ...])	Store a Pandas DataFrame into a Amazon Timestream table.

Amazon EMR

build_spark_step(path[, deploy_mode, ...])	Build the Step structure (dictionary).
build_step(command[, name, ...])	Build the Step structure (dictionary).
create_cluster(subnet_id[, cluster_name, ...])	Create a EMR cluster with instance fleets configuration.
get_cluster_state(cluster_id[, boto3_session])	Get the EMR cluster state.
get_step_state(cluster_id, step_id[, ...])	Get EMR step state.
submit_ecr_credentials_refresh(cluster_id, path)	Update internal ECR credentials.
submit_spark_step(cluster_id, path[, ...])	Submit Spark Step.
submit_step(cluster_id, command[, name, ...])	Submit new job in the EMR Cluster.
submit_steps(cluster_id, steps[, boto3_session])	Submit a list of steps.
terminate_cluster(cluster_id[, boto3_session])	Terminate EMR cluster.

Amazon CloudWatch Logs

read_logs(query, log_group_names[, ...])	Run a query against AWS CloudWatchLogs Insights and convert the results to Pandas DataFrame.
run_query(query, log_group_names[, ...])	Run a query against AWS CloudWatchLogs Insights and wait the results.
start_query(query, log_group_names[, ...])	Run a query against AWS CloudWatchLogs Insights.
wait_query(query_id[, boto3_session])	Wait query ends.

Amazon QuickSight

cancel_ingestion(ingestion_id[, ...])	Cancel an ongoing ingestion of data into SPICE.
create_athena_data_source(name[, workgroup, ...])	Create a QuickSight data source pointing to an Athena/Workgroup.
create_athena_dataset(name[, database, ...])	Create a QuickSight dataset.
create_ingestion([dataset_name, dataset_id, ...])	Create and starts a new SPICE ingestion on a dataset.
delete_all_dashboards([account_id, ...])	Delete all dashboards.
delete_all_data_sources([account_id, ...])	Delete all data sources.
delete_all_datasets([account_id, boto3_session])	Delete all datasets.
delete_all_templates([account_id, boto3_session])	Delete all templates.
delete_dashboard([name, dashboard_id, ...])	Delete a dashboard.
delete_data_source([name, data_source_id, ...])	Delete a data source.
delete_dataset([name, dataset_id, ...])	Delete a dataset.
delete_template([name, template_id, ...])	Delete a tamplate.
describe_dashboard([name, dashboard_id, ...])	Describe a QuickSight dashboard by name or ID.
describe_data_source([name, data_source_id, ...])	Describe a QuickSight data source by name or ID.
describe_data_source_permissions([name, ...])	Describe a QuickSight data source permissions by name or ID.
describe_dataset([name, dataset_id, ...])	Describe a QuickSight dataset by name or ID.
describe_ingestion(ingestion_id[, ...])	Describe a QuickSight ingestion by ID.
get_dashboard_id(name[, account_id, ...])	Get QuickSight dashboard ID given a name and fails if there is more than 1 ID associated with this name.
get_dashboard_ids(name[, account_id, ...])	Get QuickSight dashboard IDs given a name.
get_data_source_arn(name[, account_id, ...])	Get QuickSight data source ARN given a name and fails if there is more than 1 ARN associated with this name.
get_data_source_arns(name[, account_id, ...])	Get QuickSight Data source ARNs given a name.
get_data_source_id(name[, account_id, ...])	Get QuickSight data source ID given a name and fails if there is more than 1 ID associated with this name.
get_data_source_ids(name[, account_id, ...])	Get QuickSight data source IDs given a name.
get_dataset_id(name[, account_id, boto3_session])	Get QuickSight Dataset ID given a name and fails if there is more than 1 ID associated with this name.
get_dataset_ids(name[, account_id, ...])	Get QuickSight dataset IDs given a name.
get_template_id(name[, account_id, ...])	Get QuickSight template ID given a name and fails if there is more than 1 ID associated with this name.
get_template_ids(name[, account_id, ...])	Get QuickSight template IDs given a name.
list_dashboards([account_id, boto3_session])	List dashboards in an AWS account.
list_data_sources([account_id, boto3_session])	List all QuickSight Data sources summaries.
list_datasets([account_id, boto3_session])	List all QuickSight datasets summaries.
list_groups([namespace, account_id, ...])	List all QuickSight Groups.
list_group_memberships(group_name[, ...])	List all QuickSight Group memberships.
list_iam_policy_assignments([status, ...])	List IAM policy assignments in the current Amazon QuickSight account.
list_iam_policy_assignments_for_user(user_name)	List all the IAM policy assignments.
list_ingestions([dataset_name, dataset_id, ...])	List the history of SPICE ingestions for a dataset.
list_templates([account_id, boto3_session])	List all QuickSight templates.
list_users([namespace, account_id, ...])	Return a list of all of the Amazon QuickSight users belonging to this account.
list_user_groups(user_name[, namespace, ...])	List the Amazon QuickSight groups that an Amazon QuickSight user is a member of.

AWS STS

get_account_id([boto3_session])	Get Account ID.
get_current_identity_arn([boto3_session])	Get current user/role ARN.
get_current_identity_name([boto3_session])	Get current user/role name.

AWS Secrets Manager

get_secret(name[, boto3_session])	Get secret value.
get_secret_json(name[, boto3_session])	Get JSON secret value.

Amazon Chime

post_message(webhook, message)

Send message on an existing Chime Chat rooms.

Global Configurations

reset([item])	Reset one or all (if None is received) configuration values.
to_pandas()	Load all configurations on a Pandas DataFrame.