<jupyter_start><jupyter_text>Importing spark<jupyter_code># import findspark
# findspark.init()
import pyspark
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Python Spark").getOrCreate()
sc = spark.sparkContext<jupyter_output><empty_output><jupyter_text>Preparing the data<jupyter_code>df_transactions = spark.read.option("header", True)\
    .option("delimiter", "|")\
    .option("delimiter", ",")\
    .option("inferSchema", "true")\
    .csv('data/train.csv')\
    .withColumnRenamed('default_payment_next_month', 'label')
df_transactions.printSchema()
train, test = df_transactions.randomSplit([0.8, 0.2])<jupyter_output><empty_output><jupyter_text>Preparing the model<jupyter_code>from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression
assembler = VectorAssembler(
    inputCols=["MARRIAGE", "EDUCATION", "PAY_0", "PAY_2", "PAY_3"],
    outputCol="features"
)

lr = LogisticRegression(maxIter=10, regParam=10.0, elasticNetParam=0.0)
pipeline = Pipeline(stages=[assembler, lr])<jupyter_output><empty_output><jupyter_text>Fitting the model<jupyter_code>model = pipeline.fit(train)<jupyter_output><empty_output><jupyter_text>Evaluation of the model<jupyter_code>predictions = model.transform(test)
predictions.printSchema()
from pyspark.ml.evaluation import BinaryClassificationEvaluator

evaluator_roc = BinaryClassificationEvaluator(
    labelCol='label', 
    rawPredictionCol='rawPrediction',
    metricName='areaUnderROC'
)
print('Area under ROC = %s' % evaluator_roc.evaluate(predictions))<jupyter_output>Area under ROC = 0.6825956971417326<jupyter_text>Streaming data The stream will be produced by ```generate_transactions.py```<jupyter_code>from pyspark.streaming import StreamingContext

ssc = StreamingContext(sc, 1)
def process(time, rdd):
    print("========= %s =========" % str(time))
    try:

        # Convert RDD[String] to DataFrame
        rdd_proper = rdd.map(lambda line: line.split(','))

        df_stream = spark.createDataFrame(rdd_proper)
        
        # changing schema
        for c, i in zip(df_stream.columns, train.schema):
            df_stream = df_stream.withColumn(i.name, df_stream[c].cast(i.dataType))
            
        if df_stream.count() > 0:
            predictions = model.transform(df_stream)
            print(
                'Area under ROC = %s (with %ld elements)' % (evaluator_roc.evaluate(predictions), df_stream.count())
            )
    except Exception as e:
        print(e)

stream = ssc.textFileStream('data/output/')

stream.foreachRDD(process)

ssc.start()
# ssc.awaitTermination()
# stop stream
ssc.stop(True, True)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12949376375641966, 'fraction_numerical': 0.01210564930300807, 'mean_word_length': 5.006607929515418, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 25, 'https://': 0, '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': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '571183', 'license': 'MIT', 'n_tokens_mistral': 957, 'n_tokens_neox': 915, 'n_words': 156, 'path': 'MServajean/TD4/Spark_-_Stream_of_data_correction.ipynb', 'repo_name': 'mordor-ai/M2-fouille-donn-es', 'size': '2726'}