Forem: Chatchai Komrangded (Bas)

Build your own data quality rules with AWS Glue DataBrew

Chatchai Komrangded (Bas) — Sun, 21 Nov 2021 09:53:31 +0000

In previous post, I have showed How AWS Glue DataBrew can help you to Handling PII data (The post is Thai language) For an English reader, the screenshot speaks for itself, you can easily follow.

There was a recent announcement AWS Glue DataBrew users can now develop data quality rules, which are customized validation tests that set business needs for specific data, according to the company. As a result, Any data person who does not want to invest in a high-cost DQ licensing product or does not want to use an open-source framework that requires coding knowledge can define their own quality rules and populate them in a data quality dashboard and validation report, allowing customers to quickly view rule outcomes and determine whether their data is fit for use. As a Builder, with New two announcement AWS GlueDataBrew with DQ, and PII Handing will help us a lot in term of design DQ process at scale, as it's no compute server to maintenance. In this post I will walk you though How to do validate your data quality with AWS Glue DataBrew.

What is AWS Glue Databrew?
What do you mean by Data Quality rule?

Architecture Diagram
Pre-requisites
Preparing dataset
Run data profile without DQ ruleset
Create DQ ruleset
Re-run data profile with DQ ruleset
View DQ dashboard
Conclusion

Architecture Diagram

Pre-requisites

AWS Account❗
Download Data here
Unzip, and upload only patient.csv upload to Amazon S3.

Preparing dataset

Go to AWS Glue DataBrew console here

On the left you click Datasets to create Dataset for Glue DataBrew

click Create new dataset

Choose patient.csv that you just uploaded, you also have an option to get the data from Glue data catalog, Amazon Redshift, Appflow, or Snowflake as well. Click Create dataset

Run data profile

At Datasets, choose Patient dataset, and Click "Run data profile"

You will wait few minutes for data profile to populate Data profile overview, Column statistics, Data quality rules suggestion, and Data lineage, once it's done click "View data profile"

You expect to see all the data statistics that useful to understand your dataset as follow

Dataset preview

Data profile overview

Potential PII detection

Attribute Correlations

Individual Column level statistic

All Columns level statistic summary such as min, max, distribution, column type, unique value, and etc

You can also drill down to columns that identified as PII as well in All Column level statistic

You can also see Data lineage as well!

Enough for Data exploration. The purpose of this post related to Data quality, let's take a look on the Data quality tab, you will see nothing, as you don't have a rule yet! but after you run Data profile job, it suggest you based on standard DQ such as uniqueness, completeness, and etc on the right hand side

Create DQ ruleset

You can pick what that reasonable to your business criteria, Again there's zero code require! I choose to check uniqueness in my Id, SSN, and length for SSN, and etc

Again this is recommendation from AWS DataBrew. you can create your own DQ rule later, after I have done, I just click "Create ruleset"

I can add, remove, adjust, or review DQ rule I have just added

I just click "Create ruleset", I can see DQ rulesets here

click DQ rulesets name, in my case is "DQ Rulesets for Patient data" I should be able to review all my DQ rules.

Associate DQ rulesets, with Data profile job, Click here to see all your Data profile job

For my case it's "Patients - Data Profile job, choose, and click Actions, and Edit

Search for "Data quality rules", and click "Apply data quality ruleset"

Choose your DQ Rulesets, and click Apply selected rulesets, and click Save

Re-run data profile with DQ ruleset

You choose your Profile jobs, and Click "Run job"

View DQ dashboard

Wait a few minutes, you should be able to see Result from Data profile job, with DQ result, based on your DQ rules!

My dataset has 4 pass, with 6 failed, based on My DQ rules, obviously I have to fix my data, before downstream process will consume!

Conclusion

AWS Glue DataBrew is a visual data preparation tool that makes it easy to clean and normalize data using over 250 pre-built transformations, all without the need to write any code, with New feature for DQ, and PII handing it will help you to add your own data quality rules, and PII tokenization in your automated data pipelines (AWS managed airflow, or AWS step function) to make sure your data are clean, and protected.

You can think about How add to PII obfuscate, and DQ zone like this

Identifying and handling personally identifiable information (PII) ด้วย AWS Glue DataBrew

Chatchai Komrangded (Bas) — Sat, 20 Nov 2021 06:33:26 +0000

โพสต์นี้เราจะทดสอบการใช้ AWS Glue DataBrew ในการ handling sensitive data ที่อาจอยู่ในระบบบน AWS เช่น Amazon S3 ซึ่งกระบวนการนี้หลักการคือให้เนื้อข้อมูลไม่อยู่ในรูปปัจจุบันบางคนเรียกว่า Hasing, Encryption, or Masking/Redact ในโพสน์นี้เราจะทำการ Redact ข้อมูลแบบ One-way (เอาคืนกลับไม่ได้) และ Two-way (เราสามารถเรียกข้อมูลที่เป็น Orginal value ได้ อาจจะใช้กับปลายทางทีหลัง)

สำหรับที่ใครไม่รู้จักที่ข้างบนคืออะไรสามารถอ่านได้ตามลิ้งนี้
AWS Glue Databrew คืออะไร?
Common Technique มีอยู่กี่วิธี?

Architecture Diagram
Pre-requisites
Preparing dataset
Build Transformation Receipt
Obfuscation reversible
Create Transformation job
Query Result with Amazon Athena
Conclusion

Architecture Diagram

Pre-requisites

AWS Account❗
Data ที่จะใช้ ที่ใช้ทั้งหมดโหลดจาก ที่นี่
พอ Download ข้อมูลเสร็จให้ unzip แล้วเอาเฉพาะ file patient.csv upload ขึ้น Amazon S3 ก่อนนะ

Preparing dataset

เราจะเข้าไปที่ AWS Glue Databrew console ที่นี่

ด้านซ้ายมือกด Datasets เพื่อจะสร้าง Dataset สำหรับ Glue Databrew ก่อน

กด Create new dataset

เลือก csv file มะกี้ที่ upload ไปซึ่งตัวอย่างข้อมูลอยู่ใน Amazon S3 แต่อาจเอาข้อมูลมาจาก Glue data catalog, Amazon Redshift, Appflow, or Snowflake ก็ได้ จากนั้นกด Create dataset

Build Transformation Receipt

ในหน้า Glue Databrew console กด Create project

ตั้งชื่อ Project แล้วเลือก Dataset ที่เราสร้างเมื่อกี้แล้วกด Create project

เราสามารถกำหนดจำนวน Row sampling ได้ และเราต้องกำหนด IAM role เพื่อให้ Data brew สำหรับเข้าถึง Datastore นั้นๆได้เช่น Amazon S3

AWS จะ provision Glue databrew node สำหรับในการสร้าง Receipt (Transformation step ที่จะใช้ Handling PII data)

เราจะมาลองทดสอบการใช้ Redaction function (Oneway ในการ masking ข้อมูลโดยที่เราจะไม่สามารถเอา orginal กลับมาได้) เราเลือก Data masking แล้วเลือก Technique Redaction

เราเลือก SSN column มาทดสอบในการทำ Redaction เรากำหนดได้ว่าจะ Redact กี่ค่าเช่น First N, Last N ในที่นี่เราเลือกเป็น full character

เลือกเป็น Full string value, และ Apply กับทุก Rows

Glue DataBrew สามารถให้เราเข้าใจข้อมูลที่จะเปลี่ยนไปเทียบกับ Original value จริงผ่าน Preview feature (Nice!!)

หลังจาก Ok แล้วให้กด Apply สิ! เราจะเห็น receipt อยู่ด้านขวาก็คือชุด Transformation ที่ AWS Glue databrew จะรันให้เรานั่นเอง

ทีนี้เรามาทดสอบการทำ obfuscation reversible ในบางกรณีแทนที่จะ masking. เราอาจต้องการแค่ encrypt column values เมื่อไหร่ก็ต้องที่เราอาจจะใช้ข้อมูลนั้น เราสามารถ decrypt เพื่อนำ orginal data กลับมา เราจะใช้ Deterministic encryption ในการทำการ encrypt columns first_name, last_name

เลือก FIRST, LAST columns สำหรับ encryption

เลือก Secret key สำหรับการทำ encryption โดย default เราจะมี databrew!default ใน secret manager เราสามารถใช้ของเราเองก็ได้ ในที่นี้จะใช้ค่า default

เลือก Apply all rows, และ Full string

เราสามารถ Preview ได้ก่อนเพื่อเปรียบเทียบ

กด Apply เราก็จะได้ผลลัพธ์ตามนี้

หลังจากนั้นให้เรากด Publish Receipt

Obfuscation reversible

ในทางกลับกันถ้าเราต้องการ Decrypt ข้อมูลกลับไปใน Orginal format เราก็ใช้ Deterministic decryption โดยเลือก Secret key ตัวเดิมดังรูป เราจะได้ค่า FIRST กลับมาที่ Original format (Field ที่ highlight สีเหลือง)

Create Transformation job

ถัดมาเราจะสร้าง Transformation job จาก Databrew receipt กดสร้าง Create job

ตั้งชื่อ Job และเลือก Receipt ที่สร้างมะกี้

เลือก Output to Amazon S3 กด Create and Run job

เราจะเห็น Glue DataBrew job running

ไปสร้าง Table ใน Athena รอเลย!

Query Result with Amazon Athena

สร้าง Athena table แล้วทดสอบ Query

CREATE EXTERNAL TABLE patient_masked_encrypted_data (
   `id` string, 
  `birthdate` date, 
  `deathdate` date, 
  `ssn` string, 
  `drivers` string, 
  `passport` string, 
  `prefix` string, 
  `first` string, 
  `last` string, 
  `suffix` string, 
  `maiden` string, 
  `marital` string, 
  `race` string, 
  `ethnicity` string, 
  `gender` string, 
  `birthplace` string, 
  `address` string, 
  `city` string, 
  `state` string, 
  `county` string, 
  `zip` int, 
  `lat` double, 
  `lon` double, 
  `healthcare_expenses` double, 
  `healthcare_coverage` double 
)
STORED AS PARQUET
LOCATION 's3://<YOUR_S3_BUCKET>/cleaned_data_output/'

ข้อมูลถูก masking, or encryption จาก Athena query เรียบร้อย

✅ Conclusion

จบแล้ว! AWS Glue DataBrew สามารถ integrated ร่วมกับ AWS Services ตัวอืื่นในการทำ automation pipelines ได้ในการทำ obfuscate the PII data เช่น AWS Managed Apache Airflow หรือ Step function เยี่ยม! AWS Glue DataBrew เป็น serverless no code data prep เราสามารถ Scale out ตามขนาดข้อมูลที่เราต้องการได้แล้วอีกอย่างคือ Pay per use model จ่ายเท่าที่ใช้ Technical blog สำหรับภาคภาษาอังกฤษอ่านได้ที่นี่

ทดสอบทำ Machine Learning predict customer churn โดยใช้งาน Amazon SageMaker กับ Snowflake!

Chatchai Komrangded (Bas) — Fri, 17 Sep 2021 16:49:50 +0000

โพสต์นี้เราจะทดสอบการใช้ Amazon SageMaker ด้วย AutoGlueon Tabular ในการ train และ inference machine learning โดยเอาข้อมูลจาก Cloud data warehouse อย่าง snowflake ซึ่ง snowflake ถือว่าเป็น AWS partner solution ที่อยู่บน AWS marketplace, หลายๆคนอาจสงสัยว่าเอ๊ะ AWS ก็มี Amazon Redshift นิ? ใช่ครับ เรามีตัวอย่างเยอะเลยที่ใช้ Amazon redshift ร่วมกัย Sage Maker หาแปปเดียวก็เจอ 55 ตัวอย่าง, แต่เราก็จะมีบางลูกค้าในไทยที่มีการใช้ Solution ของ AWS partner เช่น snowflake คำถามที่ลูกค้าถามบ่อยๆคือจะ integrated solution รอบๆเหล่านี้เข้าไปที่ AWS บ้านหลังใหญ่ๆของเค้าได้ยังไงเช่นทำพวก ML บน Sage Maker, ก็เลยคิดว่าคงมีประโยชน์ถ้าเราทำตัวอย่างการ integration ระหว่าง Amazon Sage maker กับ Snowflake นี้กันดู! แน่นอนเหมือนเดิมเราเน้นทฏษดีพอเพียงน้ำไม่ต้องเยอะ, Let's get hand dirty!

Sage Maker คืออะไร?

Snowflake คืออะไร?

Architecture Diagram
Pre-requisites
Preparing to Load Data & Loading Data
Machine Learning Workflow in SageMaker
Update the Churn Scores to Snowflake
Conclusion

Architecture Diagram

Pre-requisites

AWS Account❗ Sage Maker มี free tier นะ
Snowflake Account❗ ทดลองเล่นได้ 30 วัน
Code ที่ใช้ทั้งหมดโหลดจากที่นี่

Preparing to Load Data & Loading Data

เราจะทำการ load ข้อมูลเข้า snowflake ก่อนโดยเพื่อจะเอาไปทำการ predict churn ใน Sage Maker อีกทีนึง หลังจากเรา login เข้ามาแล้วหน้าตา snowflake console ก็จะประมาณนี้

เราจะทำการเปลี่ยน Role เป็น Account Admin ก่อน ใช้คำสั่งด้านล่างใน snowflake worksheet ได้เลย

USE ROLE ACCOUNTADMIN;

ทำการ configure the ตัว Warehouse spec, Role, User และ Database

--Create Warehouse for AI/ML work
CREATE OR REPLACE WAREHOUSE SAGEMAKER_WH
  WITH WAREHOUSE_SIZE = 'XSMALL'
  AUTO_SUSPEND = 120
  AUTO_RESUME = true
  INITIALLY_SUSPENDED = TRUE;

--Create ROLE และ USER สำหรับ Sagemaker ในการ ดึงข้อมูลจาก Snowflake
CREATE OR REPLACE ROLE SAGEMAKER_ROLE COMMENT='SageMaker Role';
GRANT ALL ON WAREHOUSE SAGEMAKER_WH TO ROLE SAGEMAKER_ROLE;
GRANT ROLE SAGEMAKER_ROLE TO ROLE SYSADMIN;

CREATE OR REPLACE USER SAGEMAKER PASSWORD='<YOUR_PASSWORD>' 
    DEFAULT_ROLE=SAGEMAKER_ROLE 
    DEFAULT_WAREHOUSE=SAGEMAKER_WH
    DEFAULT_NAMESPACE=ML_WORKSHOP.PUBLIC
    COMMENT='SageMaker User';

GRANT ROLE SAGEMAKER_ROLE TO USER SAGEMAKER;

USE ROLE SYSADMIN;

ทำการสร้าง Databasse และ Schema และ grant access

CREATE DATABASE IF NOT EXISTS ML_WORKSHOP;
GRANT USAGE ON DATABASE ML_WORKSHOP TO ROLE SAGEMAKER_ROLE;
GRANT ALL ON SCHEMA ML_WORKSHOP.PUBLIC TO ROLE SAGEMAKER_ROLE;

--สร้าง Table แล้ว load Customer data จาก Amazon S3

USE ML_WORKSHOP.PUBLIC;
USE WAREHOUSE SAGEMAKER_WH;

CREATE OR REPLACE TABLE CUSTOMER_CHURN (
    Cust_ID INT,
    State varchar(10),
    Account_Length INT,
    Area_Code INT,
    Phone varchar(10),
    Intl_Plan varchar(10),
    VMail_Plan varchar(10),
    VMail_Message INT,
    Day_Mins FLOAT,
    Day_Calls INT,
    Day_Charge  FLOAT,
    Eve_Mins FLOAT,
    Eve_Calls INT,
    Eve_Charge FLOAT,
    Night_Mins FLOAT,
    Night_Calls INT,
    Night_Charge FLOAT,
    Intl_Mins FLOAT,
    Intl_Calls INT,
    Intl_Charge FLOAT,
    CustServ_Calls INT,
    Churn varchar(10)
);
GRANT ALL ON TABLE CUSTOMER_CHURN TO ROLE SAGEMAKER_ROLE;

เนื่องจากเราจะเอา ML scoring result สุดท้ายมาโชใน snowflake ดังนั้นเราจะสร้าง Table รอไว้ก่อน!!

CREATE OR REPLACE TABLE ML_RESULTS (
    Churn_IN INT,
    Cust_ID INT,
    Churn_Score REAL
);
GRANT ALL ON TABLE ML_RESULTS TO ROLE SAGEMAKER_ROLE;

Configure External Data Stages ใน Snowflake เพื่อใช้เป็น pointer ชี้ไปที่ Amazon S3 ซึ่งเป็นที่เก็บข้อมูลของเรา

CREATE OR REPLACE FILE FORMAT CSVHEADER
    TYPE = 'CSV'
    FIELD_DELIMITER = ','
    SKIP_HEADER = 0;

GRANT USAGE ON FILE FORMAT CSVHEADER TO ROLE SAGEMAKER_ROLE;    

--Snowflake เค้ามี sample dataset ตัวนี้ไว้ให้ละ เราก็โหลดมาเลย
CREATE OR REPLACE STAGE CHURN_DATA
  url='s3://snowflake-corp-se-workshop/sagemaker-snowflake-devdays-v1.5/sourcedata/';

--Grant Stage Object Piviliges ไปที่ SAGEMAKER_ROLE

GRANT USAGE ON STAGE CHURN_DATA TO ROLE SAGEMAKER_ROLE;

--เราสามารถดู files ที่อยู่ใน external stage ได้List and view the files in the external stage
LIST @CHURN_DATA;

--สามารถลอง Query ดูได้
SELECT $1,$2 FROM @CHURN_DATA/ LIMIT 10;

--โหลด Data เข้า Snowflake!

--เปลี่ยนมาใช้ SAGEMAKER_ROLE
USE ROLE SAGEMAKER_ROLE;

--โหลดข้อมูลจาก External Stage ไป Snowflake table
COPY INTO CUSTOMER_CHURN FROM @CHURN_DATA/ FILE_FORMAT = (FORMAT_NAME = CSVHEADER);

--ทดสอบดึงข้อมูลดู!
SELECT * FROM CUSTOMER_CHURN LIMIT 10;

Machine Learning Workflow in SageMaker

เราจะใช้ตัว AutoGluon-Tabular บน AWS marketeplace เพื่อช่วยประหยัดเวลาในการ handling พวก feature engineering ต่างๆเช่น missing data, feature transformation, model selection, algo selection, hyper parameter selection และ หรือ ensembling multiple models

ถ้าจะใช้ code ตามตัวอย่างกด Continue to Subscibe (แต่ถ้าใครจะใช้ Built in sage maker algo อย่าง XGBoost หรือ Open source framework เองก็ไม่ต้องกดนะจ้า)

เราต้อง Provision SagaMaker notebook ก่อนนะ ถ้าก็กดเข้าไป ทีนี้เราก็จะเจอ notebook แล้ว open jupyer lab ก็จะได้ภาพแบบข้างล่าง Notebook ที่คุ้นเคย

เราจะมาไล่ในส่วนแรกก่อน BAU, import lib ที่จะใช้ใข้ boto3 sdk สร้าง sagemaker session, สร้าง ตัวแปร region และ iam role

import boto3
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import io
import os
import sys
import time
import json
from IPython.display import display
from time import strftime, gmtime

import sagemaker
from sagemaker import AlgorithmEstimator, get_execution_role
from sagemaker.predictor import RealTimePredictor, csv_serializer, StringDeserializer

from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay

sess = sagemaker.Session()
role = get_execution_role()
region = boto3.Session().region_name
print("IAM role ARN: {}".format(role))

bucket = '<YOUR_BUCKET_NAME>'
prefix = 'churn-analytics'

Access ข้อมูลใน snowflake ผ่าน snowflake connector

import snowflake.connector
# Connecting to Snowflake using the default authenticator
ctx = snowflake.connector.connect(
  user='sagemaker',
  password='<YOUR_PASSWORD>',
  account='<snowflake_account>.<aws_region>',
  warehouse='SAGEMAKER_WH',
  database='ML_WORKSHOP',
  schema='PUBLIC'
)

ในเชิง practice, security standard เราไม่ควรเอา credentials เป็น clear text ฝังในนี้!!!, best practices เราควรเก็บ secret credential สักที่นึงอย่าง AWS Secrets Manager ช่วยท่านได้

ดึงข้อมูลใน snowflake ดูหน่อย

cs=ctx.cursor()
allrows=cs.execute("""select Cust_ID,STATE,ACCOUNT_LENGTH,AREA_CODE,PHONE,INTL_PLAN,VMAIL_PLAN,VMAIL_MESSAGE,
                   DAY_MINS,DAY_CALLS,DAY_CHARGE,EVE_MINS,EVE_CALLS,EVE_CHARGE,NIGHT_MINS,NIGHT_CALLS,
                   NIGHT_CHARGE,INTL_MINS,INTL_CALLS,INTL_CHARGE,CUSTSERV_CALLS,
                   CHURN from CUSTOMER_CHURN """).fetchall()

churn = pd.DataFrame(allrows)
churn.columns=['Cust_id','State','Account Length','Area Code','Phone','Intl Plan', 'VMail Plan', 'VMail Message','Day Mins',
            'Day Calls', 'Day Charge', 'Eve Mins', 'Eve Calls', 'Eve Charge', 'Night Mins', 'Night Calls','Night Charge',
            'Intl Mins','Intl Calls','Intl Charge','CustServ Calls', 'Churn']

pd.set_option('display.max_columns', 500)     # Make sure we can see all of the columns
pd.set_option('display.max_rows', 10)         # Keep the output on one page
churn

เตรียม Dataset สำหรับเอาไป Train!!

churn = churn.drop('Phone', axis=1)
churn['Area Code'] = churn['Area Code'].astype(object)
churn = churn.drop(['Day Charge', 'Eve Charge', 'Night Charge', 'Intl Charge'], axis=1)
to_split_data = churn.drop(['Cust_id'], axis=1)
train_data, test_data = np.split(to_split_data.sample(frac=1, random_state=1729), [int(0.9 * len(to_split_data))])
train_data.to_csv('train.csv', header=True, index=False)

pd.set_option('display.max_columns', 100)
pd.set_option('display.width', 1000)
display(train_data)

Get AutoGlue on container image ARN

AUTOGLUON_PRODUCT = "autogluon-tabular-v2-a11d018f6028f8192e60553704ee3d97"
def get_algorithm_arn(region, algo_name):
    acct_mapping = {
        "ap-northeast-1" : "977537786026",
        "ap-northeast-2" : "745090734665",
        "ap-southeast-1" : "192199979996",
        "ap-southeast-2" : "666831318237",
        "us-east-1"      : "865070037744",
        "eu-central-1"   : "446921602837",
        "ap-south-1"     : "077584701553",
        "sa-east-1"      : "270155090741",
        "ca-central-1"   : "470592106596",
        "eu-west-1"      : "985815980388",
        "eu-west-2"      : "856760150666",
        "eu-west-3"      : "843114510376",
        "eu-north-1"     : "136758871317",
        "us-west-1"      : "382657785993",
        "us-east-2"      : "057799348421",
        "us-west-2"      : "594846645681"
    }

    return "arn:aws:sagemaker:{}:{}:algorithm/{}".format(region, acct_mapping[region], algo_name)

algorithm_arn = get_algorithm_arn(boto3.Session().region_name, AUTOGLUON_PRODUCT)
print("The Tabular AutoGluon ARN in your region is {}.".format(algorithm_arn))

เมื่อเรา subscribed มาแล้วก็จะให้อารมณ์คล้ายกับ SageMaker built-in algorithm, ซึ่งเรียกได้ว่า deployed กันแบบ "low-to-no-code" กันไปเลย

มี 3 ค่าที่ต้องคำนึง

Hyperparamters: AutoML algorithms like AutoGluon ถูกออกมาให้ automated hpo, คือการ set คือ optional, แต่เราสามารถ override ได้! เราแค่ต้องกำหนด target column แค่นั้นอะ
Infrastructure: ใน SageMaker เราจะใช้ remote training instance ข้างนอกซึ่งเรากำหนด size กับจำนวนตามใจชอบ SageMaker จะ provision, และ de-provision ให้เรา auto (อย่าใช้ notebook นี้ train นะ❗ นั่นคือการใช้ที่ผิด)
ข้อมูลที่จะใช้ train เราควรเอาข้อมูลที่ train เก็บไว้ใน Amazon S3 นะ! สำหรับการรองรับ Scale training ใหญ่ๆในอนาคต

hyperparameters = {
    #"hyperparameters": {
    #    "NN":{"num_epochs": "1"}
    #},
    #"auto_stack": "True",
    "label": "Churn"
}

compatible_training_instance_type='ml.m5.4xlarge' 
s3_input_train = sagemaker.inputs.TrainingInput(s3_data='s3://{}/{}/train'.format(bucket, prefix), content_type='csv')

#กำหนดค่า Configuration
autogluon = AlgorithmEstimator(algorithm_arn=algorithm_arn, 
                                  role=role, 
                                  instance_count=1, 
                                  instance_type=compatible_training_instance_type, 
                                  sagemaker_session=sess, 
                                  base_job_name='autogluon',
                                  hyperparameters=hyperparameters,
                                  train_volume_size=100) 

#สั่ง Train
autogluon.fit({'training': s3_input_train})

ทำ Batch inference คือการทำ offline สร้างตัว churn scores เพื่อดูว่าลูกค้าคนไหนที่มีโอกาสจะหนีจากเรา เตรียม Test data ก่อน

batch_input = churn.iloc[:,:-1]
batch_input.to_csv('batch.csv', header=False, index=False)
boto3.Session().resource('s3').Bucket(bucket).Object(os.path.join(prefix, 'batch/in/batch.csv')).upload_file('batch.csv')

s3uri_batch_input ='s3://{}/{}/batch/in'.format(bucket, prefix)
print('Batch Transform input S3 uri: {}'.format(s3uri_batch_input))

s3uri_batch_output= 's3://{}/{}/batch/out'.format(bucket, prefix)
print('Batch Transform output S3 uri: {}'.format(s3uri_batch_output))

ทำการเรียก transformer() เพื่อสร้างและรัน batch inference process

from sagemaker.transformer import Transformer
BATCH_INSTANCE_TYPE = 'ml.c5.2xlarge'

transformer = autogluon.transformer(instance_count=1,
                                         strategy='SingleRecord',
                                         assemble_with='Line',
                                         instance_type= BATCH_INSTANCE_TYPE,
                                         accept = 'text/csv',
                                         output_path=s3uri_batch_output)

transformer.transform(s3uri_batch_input,
                      split_type= 'Line',
                      content_type= 'text/csv',   
                      input_filter = "$[1:]",
                      join_source = "Input",
                      output_filter = "$[0,-1]",
                      logs = False)

ลอง produce ผลผ่าน confusion matrix ง่ายๆดู

batched_churn_scores = pd.read_csv(s3uri_batch_output+'/batch.csv.out', usecols=[0,1], names=['id','scores'])
batched_churn_scores['scores'] = (batched_churn_scores['scores'] == "True.").astype(int)
#batched_churn_scores['Churn'] = (churn['Churn'] == "True.").astype(int)
gt_df = pd.DataFrame((churn['Churn'] == "True.").astype(int)).reset_index(drop=True)

results_df= pd.concat([gt_df,batched_churn_scores],axis=1)
pd.crosstab(index=results_df['Churn'], columns=np.round(results_df['scores']), rownames=['actual'], colnames=['predictions'])

Update the Churn Scores to Snowflake

สุดท้ายเราจะไปโชผลลัพธ์ของ Predict scoring ผ่าน snowflake กัน! เอาข้อมูลกลับ snowflake ผ่าน write_pandas

from snowflake.connector.pandas_tools import write_pandas

results_df.columns = ['CHURN_IN','CUST_ID','CHURN_SCORE']

# Write the predictions to the table named "ML_RESULTS".
success, nchunks, nrows, _ = write_pandas(ctx, results_df, 'ML_RESULTS')

display(nrows)

กลับมาที่หน้า snowflake เราจะมาลอง Query ดู

--ดู CHURN SCORES output ที่ได้มาจาก Sagemaker
SELECT * FROM ML_RESULTS LIMIT 100;

--ตัวอย่างลอง identify พื้นที่ที่มี rate ที่คาดว่าลูกค้า จะ churn โดยเราจะทำการ join ผลลัพธ์จาก predict score กลับไปที่ข้อมูลลูกค้า โดยเงื่อนไขที่โอกาส churn มากกว่าหรือเท่ากับ 10 คน และ churn scoring confidence มากกว่า 75%
SELECT C.STATE, COUNT(DISTINCT(C.CUST_ID))
FROM ML_RESULTS M INNER JOIN  CUSTOMER_CHURN C on M.CUST_ID = C.CUST_ID
WHERE M.CHURN_SCORE >= 0.75
GROUP BY C.STATE
HAVING COUNT(DISTINCT(C.CUST_ID)) >= 10
ORDER BY COUNT(C.CUST_ID) DESC;

มีประมาน 17 รัฐที่ต้อง focus ว่าลูกค้าเราน่ากำลังจะ churn!

✅ Conclusion

จบแล้ว!

ทดสอบการทำ Anonymize data in your data lake with Amazon Athena

Chatchai Komrangded (Bas) — Tue, 31 Aug 2021 11:23:37 +0000

การแชข้อมูลหรือใช้ต้อง comply ตาม regulations เวลา deal กับข้อมูลลูกค้าซึ่งอาจจะประกอบไปด้วยข้อมูลส่วนตัว (PII) ที่ส่วนมากจะถูกทำ data anonymized (การลบข้อมูลโดยไม่ระบุชื่อ) เช่นพวก PII อย่าง ชื่อ หรือ นามสกุล

ในโพสต์นี้เราจะลองมาทำกระบวนการ annonymize data โดยใช้ Amazon Athena!

Athena คืออะไร?

Architecture Diagram
Pre-requisites
Anonymize with Athena
Conclusion

Architecture Diagram

กระบวนการที่จะทำคือ เราจะ upload ข้อมูลเข้าไปใน Amazon S3 ทำการ annonymize data และเขียนข้อมูลลงไปใน S3 (persistent) แล้วสุดท้ายเราจะมาลองสร้าง annonymize data ให้อยู่ในรูปแบบ view

Pre-requisites

AWS Account - ❗ ดูวิธีการคิดราคาของ Amazon Athena ก่อนนะจ๊ะ..
โหลดข้อมูลแล้วลุยเลย! ลิ้งนี้

Anonymize with Athena

เราจะสร้าง tables ที่มีข้อมูลที่ถูก anonymized แล้ว แต่ก่อนที่จะทำลองคำนึงสิ่งที่เราจะเอาข้อมูลไปใช้ต่อ ตัวอย่างเช่น ถ้า columns นั้นเราคิดว่าไม่มีโอกาสหรือไม่แน่ใจว่าจะใช้หรือใน analytics process บางครั้งการที่เราละเว้นหรือไม่นำข้อมูลนั้นมาอาจจะเป็นวิธีที่ดีกว่า วิธีนึงในการทำ anonymize sensitive information คือการทำ hashing

hash function คือ function ที่เราใช้ในการ map data จากขนาดไหนก็ได้ให้อยู่ในรูปแบบขนาดที่คงที่ หาอ่านเพิ่มเกี่ยวกับ Hash function

เริ่มสร้าง Table โดยใช้ Athena ลิ้งนี้

CREATE EXTERNAL TABLE `cust_data`(
  `customer_id` bigint, 
  `prefix` string, 
  `first_name` string, 
  `middle_name` string, 
  `last_name` string, 
  `suffix` string, 
  `gender` string, 
  `dob` string, 
  `address` string)
ROW FORMAT DELIMITED 
  FIELDS TERMINATED BY ',' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  's3://chatkom-doc-shared/data/'
TBLPROPERTIES (
  'classification'='csv', 
  'delimiter'=',', 
  'skip.header.line.count'='1', 
  'typeOfData'='file')

ลอง Query ผ่าน Athena

SELECT * FROM "default"."customers" limit 10;

สร้าง Table ที่ทำการ Anonymize data โดยใช้ Athena ลิ้งนี้

CREATE table "default"."cust_data_anonym"
WITH (format='parquet')
as
SELECT 
customer_id,
prefix,
sha256(to_utf8(cast(first_name AS varchar))) AS first_name_anonym,
sha256(to_utf8(cast(middle_name AS varchar))) AS middle_name_anonym,
last_name,suffix,
gender,
dob,
address
FROM "default"."cust_data"

ลอง Query ผ่าน Athena

SELECT * FROM "default"."cust_data_anonym" limit 10;

เราจะเห็นข้อมูลใน first_name กับ last_name ถูก Anonymize โดย hashing function

ในกรณีเราไม่ต้องการสร้าง Table สามารถทำเป็น View ครอบได้เช่นกัน โดยใช้ Athena ลิ้งนี้

CREATE OR REPLACE VIEW "default"."v_cust_data_anonym"
as
SELECT 
customer_id,
prefix,
sha256(to_utf8(cast(first_name AS varchar))) AS first_name_anonym,
sha256(to_utf8(cast(middle_name AS varchar))) AS middle_name_anonym,
last_name,suffix,
gender,
dob,
address
FROM "default"."cust_data"

✅ Conclusion

จบแล้ว! Hasing function เป็นทางเลือกนึงในการ Anonymizing dataset ซึ่งเป็นสิ่งเราเจอบ่อยๆก่อนที่จะ analyze ข้อมูล ซึ่งในตัวอย่างนี้เรามีการ Anonymizing dataset ใน S3 กับ Athena hashing function

เริ่มใช้งาน SQL-based INSERTS, DELETES and UPSERTS in S3 โดยใช้ AWS Glue 3.0 และ Delta Lake

Chatchai Komrangded (Bas) — Mon, 30 Aug 2021 17:24:53 +0000

Credit Original English Version written by Kyle Data Science Specialist at Info Alchemy, It's great work!

โพสต์นี้เราจะมาดูวิธีการใช้ Spark SQL Engines โดยใช้ AWS Glue ในการทำ UPSERTS, DELETES, และ INSERTS

AWS Glue คืออะไร?

ในเดือนนี้ AWS ออก Glue เวอร์ชั่น 3.0! AWS Glue 3.0 สิ่งที่เพิ่มมาคือ การทำให้ Performance Apache Spark 3.1 สำหรับงานที่เป็นการประมวลผลแบบแบตช์และสตรีม เพิ่มความเร็วทั้งในส่วนการทำ ingestion, processing, และ integration! ลองมาดู benchmark กัน

ได้ Thoughput เพิ่มขึ้นหลายเท่า!

Architecture Diagram
Pre-requisites
Format to Delta
Upsert
Delete
Insert
Partitioned Data
Conclusion

Architecture Diagram

Delta lake คืออะไร
Storage layer open source project ที่อำนวยความสะดวกให้เราสามารถสร้าง Lakehouse architecture บน data lake โดยคุณสมบัติประกอบไปด้วย รองรับ ACID transactions (คล้าย RDBMS), การจัดการ metadata, และสามารถสร้าง streaming, หรือ batch processing บนพวก data lakes storage ต่างๆได้เช่น Amazon S3

รายละเอียดเพิ่มเติมที่นี่
https://docs.delta.io/latest/delta-intro.html

ทฏษดีพอละ! ลอง Build เล่นกันดู! เราจะสร้าง flow ง่ายๆคือจะเอาข้อมูลใน csv file มาแล้ว upload ไปที่ S3 bucket แล้วจะ process โดยใช้ AWS Glue! หลังจากนั้นจะ Query ข้อมูลผ่าน Amazon Athena

Pre-requisites

โหลด the Delta Lake package ลิ้งนี้ - หา file ในตารางที่ลงท้ายด้วย .jar
AWS Account - ❗ Glue ETL ไม่อยู่ใน free tier นะจ๊ะ..
โหลดข้อมูลก่อนเลย! ลิ้งนี้
Codes สามารถอ้างจาก Github เจ้าของเดิมได้ ที่นี่

Format to Delta Table

อันดับแรกเราจะ convert dataset ของเราก่อน

สร้าง Spark job จาก Glue ลิ้งนี้
เลือก Spark script editor
เลือก Create a new script with boilerplate code



# Import the packages
from delta import *
from pyspark.sql.session import SparkSession

# Initialize Spark Session along with configs for Delta Lake
spark = SparkSession \
    .builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Read Source
inputDF = spark.read.format("csv").option("header", "true").load('s3://delta-lake-aws-glue-demo/raw/')

# Write data as a DELTA TABLE
inputDF.write.format("delta").mode("overwrite").save("s3a://delta-lake-aws-glue-demo/current/")

# Read Source
updatesDF = spark.read.format("csv").option("header", "true").load('s3://delta-lake-aws-glue-demo/updates/')

# Write data as a DELTA TABLE
updatesDF.write.format("delta").mode("overwrite").save("s3a://delta-lake-aws-glue-demo/updates_delta/")

# Generate MANIFEST file for Athena/Catalog
deltaTable = DeltaTable.forPath(spark, "s3a://delta-lake-aws-glue-demo/current/")
deltaTable.generate("symlink_format_manifest")

หลักๆคือการ converts ตัว data เป็น delta format ซึ่งทำ 2 datasets ที่เป็นข้อมูลตั้งต้นกับข้อมูลที่จะใช้ updates. หลังจาก convert เราจะมีการสร้าง SYMLINK MANIFEST file ซึ่งเราจะใช้ในการดูผ่าน Athena

ลองสร้าง Table ใน Athena บน Delta Lake storage



CREATE EXTERNAL TABLE IF NOT EXISTS superstore ( 
    row_id STRING,
    order_id STRING,
    order_date STRING,
    ship_date STRING,
    ship_mode STRING,
    customer_id STRING,
    customer_name STRING,
    segment STRING,
    country STRING,
    city STRING,
    state STRING,
    postal_code STRING,
    region STRING,
    product_id STRING,
    category STRING,
    sub_category STRING,
    product_name STRING,
    sales STRING,
    quantity STRING,
    discount STRING,
    profit STRING
) ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' 
LOCATION 's3://delta-lake-aws-glue-demo/current/_symlink_format_manifest/'

ลอง Query ผ่าน Athena



select * from superstore

Upsert

Upsert คือกระบวนการในการ inserts rows เข้าไปใน db table ถ้าข้อมูลนั้นยังไม่มี หรือ update ข้อมูลถ้าข้อมูลนั้นมีแล้ว ในตัวอย่างนี้เราจะ update ค่าบาง rows ของ ship_mode, customer_name, sales, and profit columns

สร้าง Spark job จาก Glue ลิ้งนี้
เลือก Spark script editor
เลือก Create a new script with boilerplate code



# Import as always
from delta import *
from pyspark.sql.session import SparkSession

# Initialize Spark Session along with configs for Delta Lake
spark = SparkSession \
    .builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()


updateDF = spark.sql("""

MERGE INTO delta.`s3a://delta-lake-aws-glue-demo/current/` as superstore
USING delta.`s3a://delta-lake-aws-glue-demo/updates_delta/` as updates
ON superstore.row_id = updates.row_id
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED
  THEN INSERT *
""")

# Generate MANIFEST file for Athena/Catalog
deltaTable = DeltaTable.forPath(spark, "s3a://delta-lake-aws-glue-demo/current/")
deltaTable.generate("symlink_format_manifest")

คำสั่งด้านบนนั้นเราะทำการ updates ข้อมูลบน table หลัก (superstore) ถ้ามีการพบข้อมูลใน updates table (updates) จากการ matching ค่าใน row_id ถ้า row_id ตรงกันจะทำการ UPDATE ALL. ถ้าไม่มีจะทำการ INSERT ALL

สามารถดูรายละเอียดได้ ที่นี่

Delete

ลองทำการลบข้อมูลกันมั่ง! Delete จะค่อนข้างตรงไปตรงมา..

สร้าง Spark job จาก Glue ลิ้งนี้
เลือก Spark script editor
เลือก Create a new script with boilerplate code



from delta import *
from pyspark.sql.session import SparkSession


spark = SparkSession \
    .builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()


deleteDF = spark.sql("""
DELETE 
FROM delta.`s3a://delta-lake-aws-glue-demo/current/` as superstore 
WHERE CAST(superstore.row_id as integer) <= 20
""")

# Generate MANIFEST file for Athena/Catalog
deltaTable = DeltaTable.forPath(
    spark, "s3a://delta-lake-aws-glue-demo/current/")
deltaTable.generate("symlink_format_manifest")

หลักๆ code ด้านบนคือลบข้อมูลโดย row_id

ลอง Query ผ่าน Athena



SELECT * 
FROM "default"."superstore" 
-- Need to CAST hehe bec it is currently a STRING
ORDER BY CAST(row_id as integer);

เราจะเห็น records หายไปละ Data Lake กลายเป็น Delta จริงๆ!

Insert

คล้ายๆ Deletes, Inserts ค่อนข้างตรงไปตรงมา มาลองดู

สร้าง Spark job จาก Glue ลิ้งนี้
เลือก Spark script editor
เลือก Create a new script with boilerplate code



from delta import *
from pyspark.sql.session import SparkSession


spark = SparkSession \
    .builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()


insertDF = spark.sql("""
INSERT INTO delta.`s3a://delta-lake-aws-glue-demo/current/`
SELECT *
FROM delta.`s3a://delta-lake-aws-glue-demo/updates_delta/`
WHERE CAST(row_id as integer) <= 20
""")

# Generate MANIFEST file for Athena/Catalog
deltaTable = DeltaTable.forPath(
    spark, "s3a://delta-lake-aws-glue-demo/current/")
deltaTable.generate("symlink_format_manifest")

Partitioned Data

ทำไปหมดเกือบทุกกระบวนท่า แต่นี่เป็นแค่ dataset ง่ายๆ ซึ่งชีวิตจริงเราต้องคำนึงถึงการทำ parition ด้วยนะ! ทำไมต้องทำหละ คิดดูข้อมูลเราใหญ่นะ เราต้องหั่นไปชิ้นๆเพื่อเลือกดึงเฉพาะข้อมูลที่ใช้สิ, การทำ parition นี่จะช่วยเรา 2 เรื่องใหญ่ๆเลยคือ Performance กับ Cost กับบาง services ที่ใช้ในการ Scan data อย่าง Athena

เรามีการแก้ไขบางอย่างใน Spark code ในการ write data เป็น parition ทำให้โครงสร้างการเก็บข้อมูลใน S3 เป็นแบบนี้

แนวคิดของ Delta Lake ขึ้นกับ log history โดยตัว Delta lake จะมีการสร้าง delta logs ทุกๆการ committed transactions

โดย Delta logs นั้นจะมี delta files ซึ่งเก็บเป็น JSON ซึ่งในนั้นเก็บข้อมูลของการทำที่เกิดขึ้นและรายละเอียดของ snapshpt file ล่าสุด และมีการเก็บข้อมูลเกี่ยวกับ statistics ของข้อมูลเช่นกัน

จาก Data Floq

เช่นตัวอย่างข้างล่าง

raw date_part=2014-08-27/

current date_part=2014-08-27/ - DELETED ROWS

ถ้าเราลองเปิด parquet files ดู

เราจะเห็นว่า code เราทำให้เกิดการเขียน parquet file ใหม่ระหว่างการ delete โดย row_id ตั้งแต่ 21 ขึ้นไปไม่ได้ถูกลบตามเงื่อนไข หลังจากนั้น JSON file ก็จะทำการ map เข้ากับ file ที่เพิ่งถูกสร้างขึ้นมาใหม่ในการทำ versioning

เราสามารถสร้าง Table ที่มีการทำ partition แล้วใน Athena ตามคำสั่งด้านล่าง



CREATE EXTERNAL TABLE IF NOT EXISTS superstore ( 
    row_id STRING,
    order_id STRING,
    order_date STRING,
    ship_date STRING,
    ship_mode STRING,
    customer_id STRING,
    customer_name STRING,
    segment STRING,
    country STRING,
    city STRING,
    state STRING,
    postal_code STRING,
    region STRING,
    product_id STRING,
    category STRING,
    sub_category STRING,
    product_name STRING,
    sales STRING,
    quantity STRING,
    discount STRING,
    profit STRING,
    date_part STRING

)
-- Add PARTITIONED BY option
PARTITIONED BY (date_part STRING)

ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' 
LOCATION 's3://delta-lake-aws-glue-demo/current/_symlink_format_manifest/'

หลังจากนั้นให้รันคำสั่งด้านล่างเพื่อทำการ update metadata ใน catalog เพื่อให้เราสามารถ query ข้อมูลใน paritions ได้

MSCK REPAIR

to add the partitions.

✅ Conclusion

จบแล้ว!

Forem: Chatchai Komrangded (Bas)

Build your own data quality rules with AWS Glue DataBrew

Table Of Contents

Architecture Diagram

Pre-requisites

Preparing dataset

Run data profile

Create DQ ruleset

Re-run data profile with DQ ruleset

View DQ dashboard

Conclusion

Identifying and handling personally identifiable information (PII) ด้วย AWS Glue DataBrew

Table Of Contents

Architecture Diagram

Pre-requisites

Preparing dataset

Build Transformation Receipt

Obfuscation reversible

Create Transformation job

Query Result with Amazon Athena

✅ Conclusion

ทดสอบทำ Machine Learning predict customer churn โดยใช้งาน Amazon SageMaker กับ Snowflake!

Table Of Contents

Architecture Diagram

Pre-requisites

Preparing to Load Data & Loading Data

Machine Learning Workflow in SageMaker

Update the Churn Scores to Snowflake

✅ Conclusion

ทดสอบการทำ Anonymize data in your data lake with Amazon Athena

Table Of Contents

Architecture Diagram

Pre-requisites

Anonymize with Athena

✅ Conclusion

เริ่มใช้งาน SQL-based INSERTS, DELETES and UPSERTS in S3 โดยใช้ AWS Glue 3.0 และ Delta Lake

Table Of Contents

Architecture Diagram

Pre-requisites

Format to Delta Table

Upsert

Delete

Insert

Partitioned Data

raw date_part=2014-08-27/

current date_part=2014-08-27/ - DELETED ROWS

✅ Conclusion