data prep

Author: tommydangerous

mlops/metadata.yaml

@@ -1 +1,18 @@
-project_uuid: 22bfea7de412433badba758788c0fcc7
+project_type: standalone
+variables_dir: ~/.mage_data
+variables_retention_period: "90d"
+features:
+  add_new_block_v2: true
+  command_center: true
+  compute_management: false
+  custom_design: true
+  data_integration_in_batch_pipeline: true
+  dbt_v2: true
+  global_hooks: true
+  interactions: true
+  display_local_timezone: true
+  notebook_block_output_split_view: true
+  operation_history: true
+  polars: true
+project_uuid: 36404d0ffc214b8a89f598f3522c1a20
+help_improve_mage: true

mlops/settings.yaml

@@ -1,5 +1,6 @@
 projects:
-  unit_0: {}
+  unit_0:
+    active: true
   unit_1: {}
   unit_2: {}
   unit_3: {}

mlops/unit_5/pipelines/data_preparation/__init__.py mlops/unit_1/charts/__init__.py

@@ -0,0 +1,61 @@
+import statistics
+from mage_ai.data_cleaner.column_types.column_type_detector import infer_column_types
+from mage_ai.data_preparation.models.constants import DATAFRAME_ANALYSIS_MAX_COLUMNS
+from mage_ai.shared.parsers import convert_matrix_to_dataframe
+
+
+df_1 = convert_matrix_to_dataframe(df_1)
+df_1 = df_1.iloc[:, :DATAFRAME_ANALYSIS_MAX_COLUMNS]
+columns_and_types = infer_column_types(df_1).items()
+columns = [t[0] for t in columns_and_types]
+stats = ['Type', 'Missing values', 'Unique values', 'Min', 'Max', 'Mean', 'Median', 'Mode']
+rows = [[] for _ in stats]
+
+for col, col_type in columns_and_types:
+    series = df_1[col]
+
+    min_value = None
+    max_value = None
+    mean = None
+    median = None
+
+    not_null = series[series.notnull()]
+
+    if len(not_null) == 0:
+        continue
+
+    if col_type.value in ['number', 'number_with_decimals']:
+        if str(series.dtype) == 'object':
+            if col_type.value == 'number_with_decimals':
+                series = series.astype('float64')
+                not_null = not_null.astype('float64')
+            else:
+                series = series.astype('int64')
+                not_null = not_null.astype('int64')
+
+        count = len(not_null.index)
+        if count >= 1:
+            mean = round(not_null.sum() / count, 2)
+            median = sorted(not_null)[int(count / 2)]
+        min_value = round(series.min(), 2)
+        max_value = round(series.max(), 2)
+    else:
+        min_value = not_null.astype(str).min()
+        max_value = not_null.astype(str).max()
+
+    _, mode = sorted(
+      [(v, k) for k, v in not_null.value_counts().items()],
+      reverse=True,
+    )[0]
+
+    for idx, value in enumerate([
+        col_type.value,
+        len(series[series.isna()].index),
+        len(series.unique()),
+        min_value,
+        max_value,
+        mean,
+        median,
+        mode,
+    ]):
+      rows[idx].append(value)

mlops/unit_1/charts/feature_profiles_for_ingest.py

@@ -0,0 +1,8 @@
+number_of_rows = len(df_1.index)
+columns_with_mising_values = []
+percentage_of_missing_values = []
+for col in df_1.columns:
+    missing = df_1[col].isna().sum()
+    if missing > 0:
+        columns_with_mising_values.append(col)
+        percentage_of_missing_values.append(100 * missing / number_of_rows)

mlops/unit_5/utils/__init__.py mlops/unit_1/charts/ingest_time_series_bar_chart_y0.py

@@ -0,0 +1,21 @@
+from mage_ai.data_preparation.models.constants import DATAFRAME_ANALYSIS_MAX_COLUMNS
+from mage_ai.shared.parsers import convert_matrix_to_dataframe
+
+
+df_1 = convert_matrix_to_dataframe(df_1)
+columns = ['mode value', 'frequency', '% of values']
+column_index = []
+rows = []
+for col in df_1.columns[:DATAFRAME_ANALYSIS_MAX_COLUMNS]:
+    value_counts = df_1[col].value_counts()
+    if len(value_counts.index) == 0:
+        continue
+    column_value = value_counts.index[0]
+    value = value_counts[column_value]
+    number_of_rows = df_1[col].count()
+    column_index.append(col)
+    rows.append([
+        column_value,
+        f'{round(100 * value / number_of_rows, 2)}%',
+        value,
+      ])

mlops/unit_1/charts/missing_values_for_ingest.py

@@ -0,0 +1,16 @@
+import pandas as pd
+
+from mage_ai.shared.parsers import convert_matrix_to_dataframe
+
+
+if isinstance(df_1, list) and len(df_1) >= 1:
+    item = df_1[0]
+    if isinstance(item, pd.Series):
+        item = item.to_frame()
+    elif not isinstance(item, pd.DataFrame):
+        item = convert_matrix_to_dataframe(item)
+    df_1 = item
+
+columns = df_1.columns
+col = 'trip_distance'
+x = df_1[df_1[col] <= 20][col]

mlops/unit_1/charts/most_frequent_values_for_ingest.py

@@ -0,0 +1,17 @@
+from mage_ai.data_cleaner.column_types.column_type_detector import infer_column_types
+
+
+headers = ['value']
+stats = ['Columns', 'Rows']
+rows = [[len(df_1.columns)], [len(df_1.index)]]
+
+col_counts = {}
+for col, col_type in infer_column_types(df_1).items():
+    col_type_name = col_type.value
+    if not col_counts.get(col_type_name):
+        col_counts[col_type_name] = 0
+    col_counts[col_type_name] += 1
+
+for col_type, count in sorted(col_counts.items()):
+    stats.append(f'# of {col_type}')
+    rows.append([count])

mlops/unit_1/charts/prepare_histogram_k4.py

@@ -0,0 +1,2 @@
+columns = df_1.columns
+number_of_unique_values = [df_1[col].nunique() for col in columns]

mlops/unit_1/charts/summary_overview_for_ingest.py

@@ -0,0 +1,21 @@
+from typing import List
+
+if 'custom' not in globals():
+    from mage_ai.data_preparation.decorators import custom
+
+BASE_URL = 'https://github.com/mage-ai/datasets/raw/master/taxi/green'
+
+
+@custom
+def fan_out(*args, **kwargs) -> List[str]:
+    urls = []
+    
+    years = [(2024, (1, 3))]
+
+    for year, months in years:
+        for i in range(*months):
+            urls.append(f'{BASE_URL}/{year}/{i:02d}.parquet')
+
+    return [
+        urls,
+    ]

mlops/unit_1/charts/unique_values_for_ingest.py

@@ -0,0 +1,112 @@
+from typing import List, Tuple
+
+from pandas import DataFrame, Series
+from scipy.sparse._csr import csr_matrix
+from sklearn.base import BaseEstimator
+
+from mlops.utils.data_preparation.encoders import vectorize_features
+from mlops.utils.data_preparation.feature_selector import select_features
+
+if 'data_exporter' not in globals():
+    from mage_ai.data_preparation.decorators import data_exporter
+if 'test' not in globals():
+    from mage_ai.data_preparation.decorators import test
+
+
+@data_exporter
+def export(
+    data: Tuple[DataFrame, DataFrame, DataFrame], *args, **kwargs
+) -> Tuple[
+    csr_matrix,
+    csr_matrix,
+    csr_matrix,
+    Series,
+    Series,
+    Series,
+    BaseEstimator,
+]:
+    df, df_train, df_val = data
+    target = kwargs.get('target', 'duration')
+
+    X, _, _ = vectorize_features(select_features(df))
+    y: Series = df[target]
+
+    X_train, X_val, dv = vectorize_features(
+        select_features(df_train),
+        select_features(df_val),
+    )
+    y_train = df_train[target]
+    y_val = df_val[target]
+
+    return (
+        X,
+        X_train,
+        X_val,
+        y,
+        y_train,
+        y_val,
+        dv,
+    )
+
+
+@test
+def test_dataset(
+    X: csr_matrix,
+    X_train: csr_matrix,
+    X_val: csr_matrix,
+    y: Series,
+    y_train: Series,
+    y_val: Series,
+    *args,
+) -> None:
+    assert (
+        X.shape[0] == 105870
+    ), f'Entire dataset should have 105870 examples, but has {X.shape[0]}'
+    assert (
+        X.shape[1] == 7027
+    ), f'Entire dataset should have 7027 features, but has {X.shape[1]}'
+    assert (
+        len(y.index) == X.shape[0]
+    ), f'Entire dataset should have {X.shape[0]} examples, but has {len(y.index)}'
+
+
+@test
+def test_training_set(
+    X: csr_matrix,
+    X_train: csr_matrix,
+    X_val: csr_matrix,
+    y: Series,
+    y_train: Series,
+    y_val: Series,
+    *args,
+) -> None:
+    assert (
+        X_train.shape[0] == 54378
+    ), f'Training set for training model should have 54378 examples, but has {X_train.shape[0]}'
+    assert (
+        X_train.shape[1] == 5094
+    ), f'Training set for training model should have 5094 features, but has {X_train.shape[1]}'
+    assert (
+        len(y_train.index) == X_train.shape[0]
+    ), f'Training set for training model should have {X_train.shape[0]} examples, but has {len(y_train.index)}'
+
+
+@test
+def test_validation_set(
+    X: csr_matrix,
+    X_train: csr_matrix,
+    X_val: csr_matrix,
+    y: Series,
+    y_train: Series,
+    y_val: Series,
+    *args,
+) -> None:
+    assert (
+        X_val.shape[0] == 51492
+    ), f'Training set for validation should have 51492 examples, but has {X_val.shape[0]}'
+    assert (
+        X_val.shape[1] == 5094
+    ), f'Training set for validation should have 5094 features, but has {X_val.shape[1]}'
+    assert (
+        len(y_val.index) == X_val.shape[0]
+    ), f'Training set for training model should have {X_val.shape[0]} examples, but has {len(y_val.index)}'

mlops/unit_5/utils/data_preparation/__init__.py mlops/unit_1/custom/__init__.py

@@ -0,0 +1,30 @@
+from io import BytesIO
+from typing import List
+
+import pandas as pd
+
+from mlops.utils.requests.adapters import open_session
+
+if 'data_loader' not in globals():
+    from mage_ai.data_preparation.decorators import data_loader
+
+
+BASE_URL = 'https://github.com/mage-ai/datasets/raw/master/taxi/green'
+
+
+@data_loader
+def ingest_files(**kwargs) -> pd.DataFrame:
+    dfs: List[pd.DataFrame] = []
+
+    for year, months in [(2024, (1, 3))]:
+        for i in range(*months):
+            with open_session() as session:
+                response = session.get(f'{BASE_URL}/{year}/{i:02d}.parquet')
+                
+                if response.status_code != 200:
+                    raise Exception(response.text)
+
+                df = pd.read_parquet(BytesIO(response.content))
+                dfs.append(df)
+
+    return pd.concat(dfs)

mlops/unit_1/custom/download.py

@@ -0,0 +1,8 @@
+training_set:
+  object_type: pipeline
+  object_uuid: data_preparation
+  outdated_after:
+    seconds: 3599
+  settings:
+    build:
+      partitions: 1