NZFunGroupedApply

This section corresponds to use-case 3. Building ML models for each partition. (where the user is interested in running a complex function).

Applying functions on each and every partition that gets computed at runtime based on user’s selection

In this scenario, your function is applied on each and every partition that gets computed at runtime by using the input column (index parameter) that you send to NZFunGroupedApply. Netezza data slices are regenerated at run time, so that each slice contains one or more groups of the specified column only. As this scenario doesn’t require data to be aggregated in one place before applying the function, this is an optimal scenario.

A real-world ML setting needs control in defining groups. This is the most recommended option to harness Netezza parallelism for complex ML functions. Each and every group and or partition is treated as an independent dataset, and functions are run against these partitions in parallel. While NZFunTApply also applies the function to the data slices in parallel (in parallel=True), there are some differences:

1. NZFunTApply uses static slices, which means that those slices might not be the required data arrangement for your scenario.
2. The function is run against the entire slice and not on the groups in the slices.

Sample scenario: transform the data, build an ML model and score the model.

Let's define a function to do all these steps. The user function that you want to run gets two parameters by default, self (which represents the AE context) and df (dataframe for the incoming slice data). The function first imputes the columns (assigning some default values for null values), builds a decision tree classifier, and then scores the model. And the result (id, size of the dataset, location column value of the first record, prediction value) is printed with self.output.

from nzpyida import IdaDataBase, IdaDataFrame
from nzpyida.ae import  NZFunGroupedApply

idadb = IdaDataBase('weather', 'admin', 'password', verbose=True)

idadf = IdaDataFrame(idadb, 'WEATHER')

code_str_host_spus="""def decision_tree_ml(self, df):
            from sklearn.model_selection import cross_val_score
            from sklearn.impute import SimpleImputer
            from sklearn.tree import DecisionTreeClassifier
            from sklearn.model_selection import train_test_split

            from sklearn.preprocessing import LabelEncoder
            import numpy as np



            # data preparation
            imputed_df = df.copy()
            ds_size = len(imputed_df)
            temp_dict = dict()


            columns = imputed_df.columns

            for column in columns:
                if column=='ID':
                    continue

                if (imputed_df[column].dtype == 'float64' or imputed_df[column].dtype == 'int64'):
                  if imputed_df[column].isnull().sum()==len(imputed_df):
                     imputed_df[column] = imputed_df[column].fillna(0)

                  else :

                     imp = SimpleImputer(missing_values=np.nan, strategy='mean')
                     transformed_column = imp.fit_transform(imputed_df[column].values.reshape(-1, 1))
                     imputed_df[column] = transformed_column

                if (imputed_df[column].dtype == 'object'):
                    # impute missing values for categorical variables
                    imp = SimpleImputer(missing_values=None, strategy='constant', fill_value='missing')
                    imputed_df[column] = imp.fit_transform(imputed_df[column].values.reshape(-1, 1))
                    imputed_df[column] = imputed_df[column].astype('str')
                    le = LabelEncoder()

                    le.fit(imputed_df[column])
                    # print(le.classes_)
                    imputed_df[column] = le.transform(imputed_df[column])
                    temp_dict[column] = le



            # Create a decision tree
            dt = DecisionTreeClassifier(max_depth=5)
            X = imputed_df.drop(['RISK_MM', 'RAINTOMORROW'], axis=1)
            y = imputed_df['RAINTOMORROW']
            X_train, X_test, y_train, y_test = train_test_split(X,y, test_size = 0.25, random_state=42, stratify=y)


            dt.fit(X_train, y_train)

            accuracy = dt.score(X_test, y_test)

            pred_df = X_test.copy()

            y_pred= dt.predict(X_test)

            pred_df['RAINTOMORROW'] = y_pred
            pred_df['DATASET_SIZE'] = ds_size
            pred_df['CLASSIFIER_ACCURACY']=round(accuracy,2)

            original_columns = pred_df.columns

            for column in original_columns:

             if column in temp_dict:
               pred_df[column] = temp_dict[column].inverse_transform(pred_df[column])
               #print(pred_df)

            def print_output(x):
                row = [x['ID'], x['RAINTOMORROW'], x['DATASET_SIZE'], x['CLASSIFIER_ACCURACY']]
                self.output(row)


            pred_df.apply(print_output, axis=1)


"""


output_signature = {'ID':'int', 'RAINTOMORROW_PRED' :'str',  'DATASET_SIZE':'int', 'CLASSIFIER_ACCURACY':'float'}

nz_groupapply = NZFunGroupedApply(df=idadf,  code_str=code_str_host_spus, index='LOCATION', fun_name="decision_tree_ml", output_signature=output_signature, merge_output_with_df=True)

result = nz_groupapply.get_result()
print(result)

You can see a result similar to:

Notice that the result columns were merged with the original df columns with the merge_output_with_df=True option.