Lập trình Đà Nẵng: hoc lap trinh danang

Hiển thị các bài đăng có nhãn hoc lap trinh danang. Hiển thị tất cả bài đăng

Thứ Ba, 28 tháng 8, 2018

Làm quen với Pipeline trong Python sklearn – Ứng dụng Pipeline

BY ADMIN · AUGUST 28, 2018

Khi chuẩn bị data cho model trong machine learning, các điểm dữ liệu đôi khi chênh lệch nhau quá lớn, một thành phần có khoảng giá trị từ 0 đến 1000, thành phần kia chỉ có khoảng giá trị từ 0 đến 1 chẳng hạn. Lúc này, chúng ta cần chuẩn hóa dữ liệu trước khi thực hiện các bước tiếp theo. Còn gọi là Feature scaling.

Cũng có trường hợp lượng feature của data lớn nên chúng ta cần chọn ra một số lượng nhỏ hơn các feature phù hợp với bài toán. Còn gọi là Feature selection.

Sau khi sử dụng các phương pháp tiền xử lý xong, bước tiếp theo là đưa dữ liệu vào model. Và tất cả các bước trên có thể gói gọn vào Pipeline để mọi việc được đơn giản hoá. Tại sao nên dùng Pipeline và lợi ích nó đem lại như thế nào, cùng thực hành và bạn sẽ thấy được cái hay của Pipeline đem lại.

Lần này mình sẽ sử dụng bộ data: Breast Cancer Wisconsin. Bộ dataset này bao gồm 569 sample chuẩn đoán ác tính, lành tính với tế bào khối u.
Nói qua 1 chút về bộ data này nhé:

2 cột đầu tiên tương ứng với sample id và kết quả chuẩn đoán. M là ác tính, B là lành tính. Từ cột thứ 3 đến 32 chứa dữ liệu được tính toán từ hình ảnh chụp tế bào khối u. Từ những số liệu này, việc cần làm là xây dựng model dự đoán khối u ác tính hay lành tính.

Với 3 bước đơn giản sau để chuẩn bị data:

import pandas as pd
df = pd.read_csv('https://archive.ics.uci.edu/ml/machine-learning-databases/breast-cancer-wisconsin/wdbc.data', header =None)

from sklearn.preprocessing import LabelEncoder

# X chứa data của feature
X = df.loc[:, 2:].values

# y là target
y = df.loc[:, 1].values
le = LabelEncoder()

# Convert target dạng chữ về số
y = le.fit_transform(y)

le.transform(['M', 'B'])
# array([1, 0])

from sklearn.model_selection import train_test_split

# Chia data thành 2 phần: training data, test data
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=1)

Để có thể sử dụng LogisticRegression chúng ta cần chuẩn hoá dữ liệu.

Ngoài ra còn có công đoạn giảm số chiều của dữ liệu. Vậy tại sao lại cần giảm số chiều của dữ liệu làm gì? Thực ra các feature vectors trong các bài toán thực tế có thể có số chiều rất lớn, tới vài nghìn. Ngoài ra, số lượng các điểm dữ liệu cũng thường rất lớn. Nếu thực hiện lưu trữ và tính toán trực tiếp trên dữ liệu có số chiều cao này thì sẽ gặp khó khăn cả về việc lưu trữ và tốc độ tính toán. Vì vậy, giảm số chiều dữ liệu là một bước quan trọng trong nhiều bài toán

Hiện tại số chiều của dữ liệu đang là 30, mình sẽ sử dụng PCA để giảm số chiều xuống còn 2. Cụ thể về PCA bạn có thể tham khảo tại đây.

Vì vậy thay vì lần lượt làm 3 công đoạn riêng, mình sẽ tạo 1 Pipeline, Pipeline sẽ có nhiệm vụ kết hợp 3 Object: StandardScaler, PCA, LogisticRegression.

from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline

pipe_lr = Pipeline([('scl', StandardScaler()),
                    ('pca', PCA(n_components=2)),
                    ('clf', LogisticRegression(random_state=1))
                   ])

pipe_lr.fit(X_train, y_train)

'Test Accuracy: %.3f' % pipe_lr.score(X_test, y_test) 
#'Test Accuracy: 0.947'

Pipeline là list tuple. Ví dụ như ('clf', LogisticRegression(random_state=1)) thì clf là tên gán cho object LogisticRegression phía sau, chỉ là tên nên bạn muốn đặt sao cũng được, nhưng nên đặt tên có nghĩa. Và chúng ta có thể truy cập các component bên trong Pipeline qua tên định danh này.

pipe_lr.named_steps['scl']
#StandardScaler(copy=True, with_mean=True, with_std=True)

Bước trung gian trong Pipeline là transform và cuối cùng là bước dự đoán. Ở ví dụ trên có 2 bước trung gian: StandardScaler và PCA. Bước cuối là model phân loại hồi quy LogisticRegression.

Khi tiến hành gọi method fit của pipe_lr thì Object StandardScaler sẽ tiến hành gọi method fit và transform với đầu vào là Training Data.

Training Data sau khi được scaling sẽ đi đến bước tiếp theo là PCA. Tại bước này, số chiều dữ liệu hiện tại là 30 giảm xuống còn n_components chiều. Cũng giống như bước trên, method fit và transform sẽ được gọi.

Và data sau khi được transform qua 2 bước trên sẽ được chuyển đến model LogisticRegression.

Lưu ý nhỏ là:

Số bước trung gian không giới hạn. Ví dụ bạn có thể thêm bước lựa chọn đặc trưng sử dụng SelectKBest, RFE, RFECV chẳng bạn. Hay sử dụng thêm OneHotEncoder.

Sơ đồ luồng xử lý minh hoạ cho những gì mình nói ở trên:

Ngoài ra trong trường hợp muốn set lại parameter cho model hoặc các transform object bạn có thể làm như sau:

pipe_lr.set_params(pca__n_components = 15, clf__C=0.5)
pipe_lr.fit(X_train, y_train)

'Test Accuracy: %.3f' % pipe_lr.score(X_test, y_test) 
#'Test Accuracy: 0.982'

Như ở ví dụ tên định danh của object và parameter được nối với nhau bởi dấu __, rất đơn giản phải không.

Như mình đã nói ở trên, khi lượng feature của data lớn nên chúng ta cần chọn ra một số lượng nhỏ hơn các feature phù hợp với bài toán. Tuy nhiên với dữ liệu Breast Cancer Wisconsin thì lượng feature cũng không nhiều. Nhưng cứ thử xem kết quả có cải thiện được không nhé.

from sklearn.feature_selection import SelectKBest
from sklearn.feature_selection import f_regression

k_filter = SelectKBest(f_regression, k=15)
pipe_lr = Pipeline([('scl', StandardScaler()),
                    ('kbest', k_filter),
                    ('clf', LogisticRegression(random_state=1))
                   ])

pipe_lr.fit(X_train, y_train)
'Test Accuracy: %.3f' % pipe_lr.score(X_test, y_test) 
#'Test Accuracy: 0.956'

Mình thử chọn ra 15 feature tốt nhất nhưng rất tiếc là kết quả không được cải thiện. Và đây là kết quả khi chưa tunning hyper parameter.

Mình viết bài viết này muốn giới thiệu với các bạn những kiến thức cơ bản về Pipeline. Làm việc với machine learning thì chắc chắn bạn sẽ gặp và ứng dụng rất nhiều bài toán khác với pipeline. Mong rằng những kiến thức cơ bản trên sẽ giúp các bạn làm việc dễ dàng hơn.

Hẹn gặp lại các bạn ở các bài viết sau ^^

Source code bạn có thể xem tại đây.

Bạn muốn học lập trình hãy liên hệ ngay với Mỹ Vân để được nhận ưu đãi từ học viện nhé

Khóa học Lập trình Python

Khóa học lập trình Java

Khóa học lập trình PHP

Khóa học lập trình Asp net

Khóa học lập trình Angular

Khóa học lập trình Ios

Sử dụng coroutine trong python để cài đặt thuật toán điều phối các request

Một trong các vấn đề của một hệ thống backend là bài toán điều phối request tới các nguồn dữ liệu.

Xét bài toán với một hệ thống blog như Kipalog:
Mỗi một User có nhiều bài Post
Mỗi một Post có nhiều Comment
Mỗi Comment lại thuộc về một User khác nhau.

Một trong những tác vụ chính của Kipalog là render một bài Post. Khi render Post cũng cần hiển thị
luôn danh sách các comment, và tác giả của comment đó. Nếu mọi việc chỉ dừng lại ở dây, thì không có gì để bàn. Bạn đơn giản chỉ cần thực hiện một câu query join 3 bảng Post, Comment, và User
là có thể trả về dữ liệu để render Post.

Nhưng do admin và cộng đồng của Kipalog rất tích cực tham gia vào phong trào #hardcore, nên chất lượng các bài viết của Kipalog ngày càng tốt. Chính vì thế càng ngày càng nhiều user tham gia vào Kipalog hơn. Các bài viết càng ngày càng nhiều hơn. Việc lưu trữ tất cả dữ liệu trên một Database là không còn đủ nữa. Các kỹ sư của Kipalog phải thực hiện nhiều việc thay đổi kiến trúc hệ thống bằng cách cài đặt việc sharding database. Giờ đây một một bảng User, Post, Comment được tổ chức nằm trên một database riêng biệt. Đến lúc này việc render một Post đã không còn đơn giản như xưa, đoạn code để trả về dữ liệu có thể mô tả bằng Python như sau:

def get_post_info(pid):
   post = db.get_post(pid)                               
   cids = db.get_post_comment_ids(pid)
   comments = [get_comment(cid) for cid in cids]
   return post, comments

def get_comment(cid):
   comment = db.get_comment(cid)
   user = db.get_user(comment.user_id)
   comment.user = user
   return comment

Trong đoạn code trên, giả sử rằng các hàm db.get_<something> là những hàm sẽ gọi và Database để query dữ liệu. Với cách implement như trên để render được một post, sẽ cần tới 2 + 2N query vào database (1 cho việc lấy get_post, 1 cho get_post_comment_ids, N cho lấy nội dung comment, và N cho lấy thông tin của người comment). Rõ ràng đây là cách không thể nào chấp nhận được!

Với đoạn code trên, có thể thấy mấy điểm để cải thiện như sau: (1) việc get_post và get_post_comment_ids là có thể thực hiện được song song với nhau. (2) thay vì chạy tuần tự việc lấy ra các comment và user, chúng ta có thể chạy chúng song song với nhau. Hoặc tốt hơn, nếu như database hỗ trợ việc batching (ví dụ data được cached vào memcached), thì có thể gộp các cid và user_id lại với nhau để chỉ cần thực hiện 2 requests (việc gộp các request này lại với nhau được gọi là batching)

Để cài đặt giải pháp nói trên cần phải refactor lại code, đồng thời nó cũng phá vỡ tính module hoá của từng hàm, khiến cho các hàm khó dùng lại hơn.

Việc điều phối các request để lấy dữ liệu từ những nguồn khác nhau như nói ở trên là bài toán rất phổ biến đối với các công ty có dữ liệu lớn. Một giải pháp tốt của bài toán này cần implement một framework để giúp cho các lập trình viên ứng dụng: không cần phải suy nghĩ nhiều về việc dữ liệu được fetch như nào, thứ tự ra sao, đồng thời giúp cho việc viết code trở nên đơn giản và hiệu quả.

Có một vài kết quả nghiên cứu đã được đưa ra và cài đặt thành framework cụ thể. Ví dụ: Facebook sử dụng Applicative Fuctor để cài đặt thư viện Haxl, Twitter sử dụng Free monad để cài đặt framework Stitch (rất tiếc rằng Stitch chưa được opensource), và mới đây nhất là Quora đưa ra thư viện Asynq sử dụng Python Coroutine để batching request (

Bài viết này sẽ mô tả thuật toán được sử dụng trong thư viện Asynq cũng như đề xuất một vài cải tiến cho thư viện này.

Chú ý: Tại thời điểm bài viết được thực hiện, Repo Asynq mới chỉ có một commit (https://github.com/quora/asynq/tree/c25fe0834ca64b35b2cd4a4e64666d0b5423c44f). Bài viết chỉ mô tả thuật toán được sử dụng tại thời điểm này.

Coroutine là một hàm có thể có nhiều entry point. Nói cách khác, hàm này có thể được pause và resume lại trong khi chạy.

Trong Python, Coroutine được implement bằng Generator. Chính vì thế, từ đây về sau, thay vì dùng Coroutine, bài viết sẽ dùng Generator

Sau đây là một ví dụ về Generator như sau:

def gen():
   yield 1
   yield 2
   yield 3

g = gen()
g.next() # 1
g.next() # 2
g.next() # 3
g.next() # StopIterator

Các entry point của gen đó là các dòng có lời gọi lệnh yield. Khi lệnh yield được gọi, con trỏ lệnh đang từ hàm gen sẽ được trả về hàm caller của gen. Lúc này caller có thể thực hiện chạy gen tiếp hoặc là chạy hàm khác. Bất cứ hàm nào mà trong thân hàm có sử dụng yield đều được gọi là Generator trong Python.

Một điểm quan trọng của Generator trong Python đó là caller có thể truyền giá trị vào trong gen, bằng cách gọi hàm gen.send.

Ví dụ

def gen():
   a = yield 1
   yield a + 10


g = gen()
value = g.send(None)   # 1
g.send(value)          # 11

Luôn luôn cần send giá trị None như giá trị khởi đầu vào một Generator. Bằng cách lưu lại kết quả trước đó của hàm g.send, chúng ta có thể truyền lại giá trị đó vào lần chạy tiếp theo của Generator.

Tới đây bạn có thể thắc mắc vậy yield thì có tác dụng gì.

Tôi sẽ giới thiệu cho các bạn một cách tư duy mà tôi gọi là tư duy hardcore – Đây là cách tư duy mà chúng tôi (những thành viên của channel #hardcore trên RubyVietnam) rất hay sử dụng. Tư tưởng của cách tư duy này là: Chúng tôi cho rằng tất cả các bài toán ở tầng ứng dụng phần lớn đều đã có lời giải nhưng ở mức độ low level hơn, nên nếu bạn tìm hiểu đủ về cách CPU điều phối giữa các core, cách compiler traslate code của bạn sang assembly, hay cách Kernel handle resource ra sao, bạn có thể hiểu và giải quyết hầu hết các bài toán ở tầng ứng dụng khác. Nếu bạn thấy hứng thú, hãy tham gia vào channel ngày nhé

Quay trở yield, nếu sử dụng bài toán tương tự từ việc thực hiện các chỉ thị của CPU, chúng ta sẽ thấy bằng cách sử dụng yield, chúng ta đã chia việc Execute một Generator thành rất nhiều steps. Do đó khi cần Execute nhiều Generators cùng lúc, chúng ta có thể chạy các Generators này theo cơ chế PipeLine. Giả sử chúng ta có 2 Generators G1, G2. Mỗi generator có 5 Steps. Thay vì phải chạy G1 xong rồi mới chạy đến G2, chúng ta có thể pipeline bằng cách chạy (G1 Step1) (G2 Step1) (G1 Step 2) (G2 Step2), …Việc phân chia các step là dựa vào việc gọi hàm yield trong Generator.
Hay tốt hơn nữa, với các step của G1 và G2 mà có thể gộp lại, chúng ta có thể sắp xếp chúng theo thứ tự sao cho có thể gộp chung lại được, hoặc với những step liên quan tới IO, chúng ta có thể để chúng chạy song song. Đây chính là nền tảng của thuật toán trong thư viện Asynq

Quay trở lại ví dụ ở phần đầu, bằng Asynq với yield chúng ta có thể viết lại code như sau:

from asynq import async

@async()
def get_post_info(pid):
   post = yield db.get_post.async(pid)                               
   cids = yield db.get_post_comment_ids.async(pid)
   futures = [get_comment(cid) for cid in cids]
   comments = yield futures
   raise StopIterator((user, comments))

@async()
def get_comment(cid):
   comment = yield db.get_comment(cid)
   user = yield db.get_user(comment.user_id)
   comment.user = user
   raise StopIterator(comment)


result = get_post_info.async(pid)()

Đoạn code trên chỉ thay đổi đôi chút so với đoạn code ở phần đầu, nhưng khi chạy với Asynq, việc gộp chung các request đê lấy Comments và lấy User của các comments đó là được thực hiện tự động.

Để làm việc này, Asynq tạo bao lấy các Generator bằng các Future object. Future object là một object mà việc tính toán ra dữ liệu của nó là chưa được thực hiện ngay.

Future object được tạo ra bởi việc gọi là Future object được tạo ra khi gọi decorator asynccủa function, hoặc gọi lệnh yield từ trong Generator. Mỗi khi Future object được tạo ra từ lênh yield, Future object đó sẽ tạo ra một mối quan hệ phụ thuộc với caller của nó. Một Future object chỉ có kết quả khi mà các Future con của nó đều đã chạy và ra kết quả

Trong Asynq có 2 loại Future objects cần chú ý:

AsyncTask: Là các task mà bao lấy các generator. Mỗi một task sẽ chứa một generator.
BatchItemBase: là Future object mà việc thực hiện nó có thể đươc gộp lại với nhau. Mỗi một BatchItem sẽ thuộc về một Batch object. Việc thực hiện batching các item sẽ được thực hiện bởi Batch object.

Việc chạy các Future object sẽ được đẩy cho một scheduler. Scheduler sẽ được kích hoạt khi chúng ta gọi một async task như result = get_post_info.async(pid)(). Ở đây get_post_info.async(pid) là một AsyncTask. Việc gọi một async task sẽ tương đương với việc gọi hàm scheduler để schedule task đó.

Scheduler sẽ chứa 2 queues:

queue _tasks chứa các task cần schedule
queue _batches chứa các batch cần schedule.

Ở trạng thái khởi tạo, cả 2 queue này đều rỗng, không chưa bất cứ một phần tử nào.

Khi Scheduler được yêu cần schedule một list các task, scheduler sẽ đẩy các task vào trong queue _tasks. Sau đó nó sẽ lấy các task ra lần lượt. Mỗi khi một task được lấy khỏi queue _task, task đó sẽ được execute bằng cách gọi hàm generator.send để execute từng step của Generator (như mô tả ở phần 2). Khi AsyncTask nhận được kết quả từ generator, kết quả này có thể là một trong 3 loại:

Một AsyncTask khác. Khi đó AsyncTask mới sẽ là AsyncTask con của AsyncTask kể trên. AsyncTask con này sẽ được đẩy tiếp vào queue _tasks của scheduler
Một BatchItemBase, khi đó batch object của BatchItemBase sẽ được đẩy vào trong queue _batches của scheduler. Chú ý rằng chúng
Một giá trị cụ thể, tức là scheduler đã hoàn thành nhiệm vụ của nó.

Scheduler sau khi schedule các task từ queue _tasks, nó sẽ schedule tiếp các object từ trong queue _batches. Các batch này sẽ được thực hiện tuần tự bằng cách gọi hàm flush của từng batch.

Với thuật toán như trên, và đoạn code như chúng ta mô tả ở phần đầu của mục 3. Nếu các tác vụ db.get_post.async được cài đặt như các BatchItemBase, chúng ta sẽ có những batch sau:

batch 1: db.get_post.async với tham số pid
batch 2: db.get_post_comment_ids.async với tham số pid
batch 3: db.get_comment với N items có các tham số từ mảng cids
batch 4: db.get_user với N items có các tham số từ mảng user_ids

Như thế, chúng tả chỉ mất 4 requests thay vì 2 + 2N như ban đầu.

Một điểm Asynq làm chưa tốt đó là khi schedule các batchs chúng ta có thể schedules chúng song song với nhau. Ví dụ batch 1 và batch 2 có thể thực hiện song song được.

Tất nhiên để làm được việc đó, đoạn code ở trên (phần đầu của 3) cũng phải thay đổi một chút (yield 2 futures get_post và get_post_comment_ids cùng lúc) và đồng thời cần phải implement một cơ chế EventLoop trong Asynq để việc thực hiện các batch sẽ đẩy về EventLoop

Việc sử dụng Generator để scheduler các request không phải là idea mới trong Python. Thư viện NDB của Google App Engine đã implement cơ chế giống như Asynq với EventLoop từ những năm 2012. Tuy nhiên source code của NDB phức tạp và khó đọc hơn rất nhiều so với Asynq.

Bằng mô hình coroutine, Asynq sử dụng một thuật toán khá đơn giản để điều phối các request: cố gắng batching càng nhiều request càng tốt.
Áp dụng phương pháp điều phối này cùng với GraphQL tôi nghĩ chúng ta có thể build được những backend vừa rất mềm dẻo trong việc trả về dữ liệu cho client, lại vừa có performance tốt mà chi phí cho việc viết code là không lớn (không thay đổi nhiều so với cách viết truyền thống).