def METHOD_NAME(fold_id, df_target, df_trx, conf_pp):
    spark = SparkSession.builder.getOrCreate()
    preproc = PysparkDataPreprocessor(
        col_id='customer_id',
        col_event_time='event_time',
        event_time_transformation='none',
        cols_category=['mcc_code', 'tr_type'],
        cols_numerical=['amount'],
        cols_identity=['term_id'],
        cols_last_item=['target_gender'],
    )
    df_train_trx = df_trx.join(df_target, how='left', on='customer_id').where(
        F.coalesce(F.col('fold_id'), F.lit(-1)) != fold_id).drop('fold_id')
    df_test_trx = df_trx.join(
        df_target.where(F.col('fold_id') == fold_id).drop('fold_id'),
        how='inner', on='customer_id')
    df_train_data = preproc.fit_transform(df_train_trx)
    df_test_data = preproc.transform(df_test_trx)
    file_name_train = get_file_name_train(fold_id)
    file_name_test = get_file_name_test(fold_id)
    df_train_data.write.parquet(os.path.join(conf_pp.folds_path, file_name_train), mode='overwrite')
    df_test_data.write.parquet(os.path.join(conf_pp.folds_path, file_name_test), mode='overwrite')
    with open(os.path.join(conf_pp.folds_path, f'preproc_{fold_id}.p'), 'wb') as f:
        pickle.dump(preproc, f)
    logger.info(f'Preprocessor[{fold_id}].category_dictionary_sizes={preproc.get_category_dictionary_sizes()}')
    for df_name in [file_name_train, file_name_test]:
        df = spark.read.parquet(os.path.join(conf_pp.folds_path, df_name))
        r_counts = df.groupby().agg(
            F.sum(F.when(F.col('target_gender').isNull(), F.lit(1)).otherwise(F.lit(0))).alias('cnt_none'),
            F.sum(F.when(F.col('target_gender') == 0, F.lit(1))).alias('cnt_0'),
            F.sum(F.when(F.col('target_gender') == 1, F.lit(1))).alias('cnt_1'),
        ).collect()[0]
        cnt_str = ', '.join([
            f'{r_counts.cnt_none:5d} unlabeled'
            f'{r_counts.cnt_0:5d} - 0 class'
            f'{r_counts.cnt_1:5d} - 1 class'
        ])
        logger.info(f'{df_name:30} {cnt_str}')
    for df_name in [file_name_train, file_name_test]:
        df = spark.read.parquet(os.path.join(conf_pp.folds_path, df_name))
        logger.info(f'{df_name:30} {df}')