โปรเจค DE ง่ายๆ ผ่าน GCP (2)

หลังจากที่เราเอาข้อมูลจาก kaggle มาลอง clean ใน colab โดยใช้ pandas แล้ว
ใครยังไม่ได้อ่านก็ลองไปอ่านได้นะครับ

https://medium.com/@jaroensakyodkantha_654/

เราลองมาจำลองสถานะการว่า เรามีข้อมูล ขอลูกค้า อยู่ในฐาน Data base
ในส่วนของ DB ที่ผมจะเลือกใช้ ้เป็นเว็บที่ให้เราเก็บ DB แบบฟรี

จัดการสมัครสมาชิก และรับ Database Details ต่างๆที่ส่งเข้ามาในเมล
จากนั้นเราก็เข้าไปตรง phpMyAdmin for database administration

ทางหน้าตาเว็บก็จะเหมือนพวกโปรแกรม DB ส่วนใหญ่ ที่ทำให้เราสามารถ
ดู Structure ทำ SQL หรือ Qurey ข้อมูล import export ได้ด้วย

หลังจากนั้นผม ก็ impoyt dataset เข้าไป ในที่นี้ ผมขอเปลี่ยนชื่อของ dataset จาก bank_term_deposit ให้เป็น bank_term_deposit_old

ข้อมูลของเรามีอยู่ 4,522 แถว ลองมาเชื่อใจ colab ดูว่าสามารถเชื่อมได้ไหม โดย สิ่งแรกคือ

! pip install pymysql

หลังจากนั้นก็ใส่ข้อมูลตามใน e-mail ลงใน class เราสร้าง

class Config:

MYSQL_HOST = ‘ — — — — — — ’

MYSQL_PORT = 3306 # default สำหรับ port MySQL

MYSQL_USER = ‘ — — — — — — -’

MYSQL_PASSWORD = ‘ — — — — — -’

MYSQL_DB = ‘ — — — — — — ’

MYSQL_CHARSET = ‘utf8mb4’

ต่อไปก็เชื่อมพอต

import pymysql

# Connect to the database

connection = pymysql.connect(host=Config.MYSQL_HOST,

port=Config.MYSQL_PORT,

user=Config.MYSQL_USER,

password=Config.MYSQL_PASSWORD,

db=Config.MYSQL_DB,

charset=Config.MYSQL_CHARSET,

cursorclass=pymysql.cursors.DictCursor)

จากนั้นลอง print ชื่อ table ดู

cursor = connection.cursor()

cursor.execute(“show tables;”)

tables = cursor.fetchall()

cursor.close()

print(tables)

เชื่อมต่อเรียบร้อย แสดงว่า database ของเราใช้ได้

ลองนับดูมีแถวเท่ากันไหม

with connection.cursor() as cursor:

cursor.execute(“SELECT * FROM bank_term_deposit_old”)

result = cursor.fetchall()

print(“number of rows: “, len(result))

เรียบร้อย เท่ากัน

มาเริ่ม ETL กันดีกว่า

เครื่องมือที่เราใช้คือ Cloud Composer ของ Google

สามารถดู vdo ประกอบการสร้าง composer ได้ที่นี้

เมื่อเราสร้าง Environments ของ Cloud Composer แล้ว ต้องทำการลง pip ด้วยโดย

จากนั้น กดไปที่ airflow และเข้าไปที่ connection และแก้ไข my_sql ใส่ Connect mysql ของ database ที่เราใช้
Admin >> connections >> mysql_default

ETL

กระบวนการต่อไปเป็นกระบวนการทำ ETL โดย

1. Data Collection & Data Cleansing (โดยใช้ Hook และ นำโค้ดที่เราเขียนใน colab มาใช้)
2. Data Lake (กระบวนการจของข้อที่ถูก hook มาและ clean จะอยู่ใน data lake ในที่นี้ เราใช้ Google Cloud Storage ในการเก็บ)
3. Data Pipeline (กระบวนการต่างๆ จะทำโดย Apache Airflow ที่มี Kubernete เป็น Cloud composer
4. Serverless Data Warehouse (ใน GCP เราจะใช้ Google BigQuery เป็น data warehouse )

พอเสร็จขั้นตอน เราก็มาแปลง code จาก colab ของเรา ให้กลายเป็น codeใน composer ของเราโค้ดจะเป็นตามนี้

เข้าไปใน Cloud Shell มุมบนขวา แล้วก็กด Open Editor

สร้าง ไฟล์ใหม่

(โค้ดทั้งหมดสามารถดูได้ที่นี้ https://github.com/jaroensakyod/projectde1/blob/master/RunP1.py)

แล้วก็นำโค้ดที่เราจะ run มาใส่

รันคำสั่งใน command เพือ copy ไฟล์ไว้ใน dags

กด Authorize(ถ้าในตัวโค้ดของเรามีการรัน bigquery ก็อย่าลิมไป enable api ด้วยนะครับ ไม่งั้นจะรันไม่ได้)

ในโฟล์เดอร์ของ dags เราก็จะปรากฏ ไฟล์ที่เรา copy ไป

จากนั้นไปดูที่ Airflow

แปลว่า task เราได้เชื่อกับ airflow แล้ว

ดูที่ Tree view

ผ่านนนนนนน

ต่อไปมาดูที่ Bigquery

มีการแก้โค้ดตรง pdays นิดหน่อยนะครับ จาก -1 เป็น 1

ข้อมูลใน Bigquery สวยงามเหมือนที่เราต้องการใน colab เลย

ใครอ่านมาถึงตรงนี้ ผมก็กราบขอบพระคุณมากๆนะครับ ผมผิดพลาดประการใด ก็ขออภัยมา ณ ที่นี้ด้วย มีข้อติชมตรงไหน ก็แนะนำกันได้นะครับ :)