create process data

Author: khuyentran1401

data_science_tools/prefect_2/config/data/data.yaml

@@ -1,4 +1,4 @@
 raw: 
   name: pet 
   path: data/raw/pet.csv
-processed: data/processed/pet.csv
+processed: data/processed

data_science_tools/prefect_2/config/process.yaml

@@ -2,4 +2,39 @@ defaults:
   - data: data
   - _self_
 
-n_estimators: 20
+n_estimators: 20
+use_cols:
+  - Type
+  - Age
+  - Breed1
+  - Breed2
+  - Gender 
+  - Color1
+  - Color2
+  - Color3
+  - MaturitySize
+  - FurLength
+  - Vaccinated
+  - Dewormed
+  - Sterilized
+  - Health
+  - Quantity
+  - Fee
+  - desc_length
+  - desc_words
+  - average_word_length
+  - AdoptionSpeed
+cat_cols:
+  - Type
+  - Breed1
+  - Breed2
+  - Gender
+  - Color1
+  - Color2
+  - Color3
+  - MaturitySize
+  - FurLength
+  - Vaccinated
+  - Dewormed
+  - Sterilized
+  - Health

data_science_tools/prefect_2/data/processed/X_test

@@ -1,68 +1,96 @@
+import pickle
+
 import hydra
 import pandas as pd
 from hydra.utils import to_absolute_path as abspath
-from nltk.tokenize import TweetTokenizer
 from prefect import flow, task
-from sklearn.ensemble import RandomForestClassifier
-from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.model_selection import train_test_split
 
+pd.options.mode.chained_assignment = None
 # ---------------------------------------------------------------------------- #
 #                                 Create tasks                                 #
 # ---------------------------------------------------------------------------- #
 
 
 @task
 def get_data(data_path: str):
-    train = pd.read_csv(abspath(data_path.train))
-    test = pd.read_csv(abspath(data_path.test))
-    return {"train": train, "test": test}
+    return pd.read_csv(abspath(data_path))
 
 
-@task
-def get_all_data(data: dict):
-    return pd.concat([data["train"], data["test"]])
+def fill_na_description(data: pd.DataFrame):
+    data["Description"] = data["Description"].fillna("")
+    return data
+
+
+def get_desc_length(data: pd.DataFrame):
+    data["desc_length"] = data.apply(lambda x: len(x))
+    return data
+
+
+def get_desc_words(data: pd.DataFrame):
+    data["desc_words"] = data["Description"].apply(lambda x: len(x.split()))
+    return data
+
+
+def get_average_word_length(data: pd.DataFrame):
+    data["average_word_length"] = data["desc_length"] / data["desc_words"]
+    return data
 
 
 @task
-def get_vectorizer(data: pd.DataFrame):
-    tokenizer = TweetTokenizer()
-    vectorizer = TfidfVectorizer(ngram_range=(1, 2), tokenizer=tokenizer.tokenize)
-    vectorizer.fit(data["Description"].fillna("").values)
-    return vectorizer
+def get_description_features(data: pd.DataFrame):
+    return (
+        data.pipe(fill_na_description)
+        .pipe(get_desc_length)
+        .pipe(get_desc_words)
+        .pipe(get_average_word_length)
+    )
 
 
 @task
-def encode_description(vectorizer: TfidfVectorizer, data: pd.DataFrame):
-    X_train = vectorizer.transform(data["Description"].fillna(""))
-    print(X_train)
-    print(type(X_train))
-    return X_train
+def filter_cols(use_cols: list, data: pd.DataFrame):
+    return data[use_cols]
 
 
 @task
-def get_adoption_speed(data: pd.DataFrame):
-    return data["AdoptionSpeed"]
+def encode_cat_cols(cat_cols: list, data: pd.DataFrame):
+    cat_cols = list(cat_cols)
+    data[cat_cols] = data[cat_cols].astype(str)
+    for col in cat_cols:
+        _, indexer = pd.factorize(data[col])
+        data[col] = indexer.get_indexer(data[col])
+    return data
 
 
 @task
-def get_classifier(data: pd.DataFrame, adoption_speed: pd.Series, n_estimators: int):
-    clf = RandomForestClassifier(n_estimators=n_estimators)
-    clf.fit(data, adoption_speed)
+def split_data(data: pd.DataFrame):
+    X = data.drop(columns=["AdoptionSpeed"])
+    y = data["AdoptionSpeed"]
+    X_train, X_test, y_train, y_test = train_test_split(
+        X,
+        y,
+        test_size=0.3,
+        random_state=0,
+    )
+    return {"X_train": X_train, "X_test": X_test, "y_train": y_train, "y_test": y_test}
 
 
-@flow
-def get_description_features(config, all_data, data: dict):
-    vectorizer = get_vectorizer(all_data)
-    X_train = encode_description(vectorizer, data["train"])
-    y_train = get_adoption_speed
+@task
+def save_data(data: dict, save_dir: str):
+    for name, value in data.items():
+        save_path = abspath(f"{save_dir}/{name}")
+        pickle.dump(value, open(save_path, "wb"))
 
 
 @hydra.main(config_path="../config", config_name="process", version_base=None)
 @flow
 def process_data(config):
-    data = get_data(config.data.raw)
-    all_data = get_all_data(data)
-    get_description_features(config, all_data, data)
+    data = get_data(config.data.raw.path)
+    processed = get_description_features(data)
+    filtered = filter_cols(config.use_cols, processed)
+    encoded = encode_cat_cols(config.cat_cols, filtered)
+    split = split_data(encoded)
+    save_data(split, config.data.processed)
 
 
 # ---------------------------------------------------------------------------- #