<jupyter_start><jupyter_text>Применение MLlib  Операции с файлами в системе<jupyter_code># Просмотр содержимого папки
# dbutils.fs.ls('dbfs:/FileStore/tables/')
# Удаление одиночного файла
# dbutils.fs.rm('dbfs:/FileStore/tables/...csv')
# Рекурсивное удаление самой папки и его содержимого
# %fs rm -r /FileStore/tables/<jupyter_output><empty_output><jupyter_text>Загрузка и сохранение датасета<jupyter_code># DATASET_URL = 'https://raw.githubusercontent.com/sberbank-ai-lab/LightAutoML/master/example_data/test_data_files/sampled_app_train.csv'
# from pyspark import SparkFiles
# spark.sparkContext.addFile(DATASET_URL)
# df = spark.read.csv('file://'+SparkFiles.get('sampled_app_train.csv'), header=True, inferSchema=True)
# display(df)
# df.printSchema()
# Сохраняем файл в хранилище
# df.write.format('parquet').mode('overwrite').save('dbfs:/tmp/dataset/data.parquet')<jupyter_output><empty_output><jupyter_text>Подготовка данных<jupyter_code>df = spark.read.format('parquet').load('/tmp/dataset/data.parquet')
df.limit(10).toPandas().head()
import pyspark.sql.functions as F
from pyspark.sql.functions import date_add

# Переводим даты в формат понятный человеку
df = df.withColumn('DATE_START', F.lit('2018-01-01'))
df = df.withColumn('BIRTH_DATE', F.expr('date_add(DATE_START, DAYS_BIRTH)'))
df = df.withColumn('EMP_DATE', F.expr('date_add(DATE_START, DAYS_EMPLOYED)'))
columns_to_drop = ['DAYS_BIRTH', 'DAYS_EMPLOYED','DATE_START']
df = df.drop(*columns_to_drop)
# Переводим даты в unix формат для машинного обучения
df = df.withColumn('BIRTH_DATE_UNIX', F.unix_timestamp(df.BIRTH_DATE, 'yyyy-MM-dd'))
df = df.withColumn('EMP_DATE_UNIX', F.unix_timestamp(df.EMP_DATE, 'yyyy-MM-dd'))
df = df.drop('BIRTH_DATE').drop('EMP_DATE')
# Заполняем пусты в датасете
df = df.na.fill(0)
df = df.na.fill('-')
df.limit(10).toPandas().head()
df.printSchema()<jupyter_output><empty_output><jupyter_text>Разбивка на тренировочный и тестовый датасет<jupyter_code>RANDOM_STATE = 42
TEST_SIZE = 0.4
TRAIN_SIZE = 0.6

trainDF, testDF = df.randomSplit([TRAIN_SIZE, TEST_SIZE], seed=RANDOM_STATE)
print(f'''Размер первоначального датасета - {df.count()}. Тренирочный массив данных -  {trainDF.count()}, проверочный -  {testDF.count()}''')<jupyter_output><empty_output><jupyter_text>Кодировка<jupyter_code>from pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler


categoricalCols = [field for (field, dataType) in trainDF.dtypes if dataType == 'string']
indexOutputCols = [x + 'Index' for x in categoricalCols]
oheOutputCols = [x + 'OHE' for x in categoricalCols]

stringIndexer = StringIndexer(inputCols=categoricalCols, outputCols=indexOutputCols, handleInvalid='skip')
oheEncoder = OneHotEncoder(inputCols=indexOutputCols, outputCols=oheOutputCols)

numericCols = [field for (field, dataType) in trainDF.dtypes if ((dataType in ['double', 'integer', 'long']) & (field != 'TARGET'))]
assemblerInputs = oheOutputCols + numericCols
vecAssembler = VectorAssembler(inputCols=assemblerInputs, outputCol='features')<jupyter_output><empty_output><jupyter_text>Построение пайплайна с обучением и прогнозом<jupyter_code>from pyspark.ml import Pipeline
from pyspark.ml.classification import RandomForestClassifier


rf = RandomForestClassifier(labelCol='TARGET', numTrees=20, maxDepth=5, maxBins=32, impurity='gini')
pipeline = Pipeline(stages = [stringIndexer, oheEncoder, vecAssembler, rf])

pipelineModel = pipeline.fit (trainDF)
predDF = pipelineModel.transform(testDF)
predDF.select('TARGET','prediction').show()<jupyter_output><empty_output><jupyter_text>Оценка качества работы алгоритма<jupyter_code>from pyspark.ml.evaluation import MulticlassClassificationEvaluator

evaluator_accuracy = MulticlassClassificationEvaluator(labelCol='TARGET', predictionCol='prediction', metricName='accuracy')
accuracy = evaluator_accuracy.evaluate(predDF)
print('Accuracy = ' + str(accuracy))
from pyspark.ml.evaluation import BinaryClassificationEvaluator

evaluator_areaunderroc = BinaryClassificationEvaluator(labelCol='TARGET', metricName='areaUnderROC') 
areaunderroc = evaluator_areaunderroc.evaluate(predDF)
print('AUC = ' + str(areaunderroc))
from sklearn import metrics

t = predDF.select('TARGET','prediction').toPandas()
print(metrics.classification_report(t.TARGET, t.prediction))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13242115027829313, 'fraction_numerical': 0.0055658627087198514, 'mean_word_length': 8.437636761487965, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '434681', 'license': 'MIT', 'n_tokens_mistral': 1647, 'n_tokens_neox': 1666, 'n_words': 253, 'path': 'ml/databricks_mllib_sberbank_dataset.ipynb', 'repo_name': 'grishenkovp/databricks', 'size': '4312'}