Tutoriel Apache Spark : ML avec PySpark

Télécharger pyspark avec pip

Vous pouvez télécharger et installer PySpark avec pip. C'est très simple, comme pour n'importe quel autre package. Exécutez la commande habituelle et laissez l'outil faire le travail :

$ pip install pysparkPropulsé par

Sinon, rendez-vous sur la page de téléchargement de Spark. Conservez les options par défaut pour les trois premières étapes et vous obtiendrez un lien téléchargeable à l'étape 4. Cliquez sur ce lien pour télécharger. Pour ce tutoriel, téléchargez Spark 2.2.0 et le package « Pre-built for Apache Hadoop 2.7 and later ».

Remarque : le téléchargement peut prendre du temps !

Télécharger Spark avec Homebrew

Vous pouvez également installer Spark avec Homebrew, un gestionnaire de paquets libre et open source. C'est particulièrement pratique sous macOS.

Exécutez simplement les commandes suivantes pour chercher Spark, obtenir des informations, puis l'installer sur votre ordinateur :

# Rechercher spark
$ brew search spark

# Obtenir des infos sur apache-spark
$ brew info apache-spark

# Installer apache-spark
$ brew install apache-sparkPropulsé par

Télécharger et configurer Spark

Décompressez ensuite l'archive apparue dans votre dossier Downloads. Cela peut se faire automatiquement en double-cliquant sur spark-2.2.0-bin-hadoop2.7.tgz ou via le Terminal avec la commande :

$ tar xvf spark-2.2.0-bin-hadoop2.7.tgzPropulsé par

Puis, déplacez le dossier extrait vers /usr/local/spark avec :

$ mv spark-2.1.0-bin-hadoop2.7 /usr/local/sparkPropulsé par

Remarque : si vous obtenez une erreur de permission lors du déplacement, préfixez la commande par sudo. La ligne devient alors $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark. Il vous sera demandé votre mot de passe, généralement celui que vous utilisez pour ouvrir votre session.

Vous êtes prêt : ouvrez le fichier README dans /usr/local/spark. Pour cela, exécutez

$ cd /usr/local/sparkPropulsé par

Vous arrivez ainsi dans le bon répertoire. Vous pouvez ensuite lister son contenu et lire le README inclus.

Utilisez d'abord $ ls pour afficher les fichiers et dossiers du dossier spark. Vous verrez un fichier README.md. Ouvrez-le avec l'une des commandes suivantes :

# Ouvrir et éditer le fichier
$ nano README.md

# Simplement lire le fichier 
$ cat README.mdPropulsé par

Astuce : utilisez la touche Tab pour l'autocomplétion des noms de fichiers ; cela vous fera gagner du temps.

Ce README fournit des informations générales sur Spark, la documentation en ligne, la construction de Spark, les shells interactifs Scala et Python, des exemples, etc.

La partie la plus utile ici peut être la section sur la construction de Spark, mais elle ne concerne vraiment que les versions non précompilées. Pour ce tutoriel, vous avez téléchargé une version précompilée. Appuyez sur CTRL + X pour quitter le README et revenir au dossier spark.

Si vous avez choisi une version non construite, exécutez la commande indiquée dans le README. Au moment de la rédaction, c'était :

$ build/mvn -DskipTests clean package runPropulsé par

Notez que cette commande peut être longue.

Applications Spark vs shell Spark

Le shell interactif est un environnement de type REPL (Read-Eval-Print-Loop) : tout ce que vous tapez est lu, exécuté puis affiché pour que vous puissiez enchaîner vos analyses. Cela rappelle IPython, le shell Python interactif que vous utilisez peut-être avec Jupyter. Pour en savoir plus, lisez l'article de DataCamp IPython or Jupyter.

Vous pouvez utiliser ce shell, disponible pour Python et Scala, pour tout travail interactif.

Outre le shell, vous pouvez écrire et déployer des applications Spark. Contrairement à une application, le SparkSession est déjà créé dans le shell pour vous faire gagner du temps.

Vous vous demandez peut-être : qu'est-ce que SparkSession ?

C'est le point d'entrée principal de Spark : il représente la connexion à un cluster Spark et permet de créer des RDD et de diffuser des variables sur ce cluster. Avec Spark, tout commence et finit avec ce SparkSession. Remarque : avant Spark 2.0.0, les objets de connexion principaux étaient SparkContext, SqlContext et HiveContext.

Vous en verrez plus par la suite. Concentrez-vous pour l'instant sur le shell.

Le shell Spark pour Python

Depuis le dossier spark situé en /usr/local/spark, lancez :

$ ./bin/pysparkPropulsé par

Du texte s'affiche, puis « Spark » apparaît, comme ceci :

Python 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016, 12:39:47) 
[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
17/07/26 11:41:26 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/07/26 11:41:47 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.2.0
      /_/

Using Python version 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016 12:39:47)
SparkSession available as 'spark'.
>>>Propulsé par

Une fois à cet écran, vous pouvez commencer à expérimenter dans le shell interactif !

Astuce : si vous préférez IPython au shell Spark, définissez la variable d'environnement suivante :

export PYSPARK_DRIVER_PYTHON="/usr/local/ipython/bin/ipython"Propulsé par

>>> rdd1 = spark.sparkContext.parallelize([('a',7),('a',2),('b',2)])
>>> rdd2 = spark.sparkContext.parallelize([("a",["x","y","z"]), ("b",["p", "r"])])
>>> rdd3 = spark.sparkContext.parallelize(range(100))Propulsé par

>>> rdd1.reduce(lambda a,b: a+b)Propulsé par

>>> rdd2.flatMapValues(lambda x: x).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]Propulsé par

PySpark dans Jupyter Notebook

Pour cette partie, vous n'utiliserez pas le shell interactif, mais construirez votre propre application dans un Jupyter Notebook. Tout est déjà installé, vous n'avez presque rien à faire pour utiliser PySpark dans Jupyter.

Lancez l'application comme d'habitude avec $ jupyter notebook. Créez un nouveau notebook, importez la bibliothèque findspark et utilisez la fonction init(). Vous fournirez ici le chemin /usr/local/spark à init() car vous savez que Spark est installé à cet emplacement.

# Import findspark 
import findspark

# Initialize and provide path
findspark.init("/usr/local/spark")

# Or use this alternative
#findspark.init()Propulsé par

Astuce : si vous ne connaissez pas votre chemin Spark, utilisez findspark.find() pour le détecter automatiquement.

Pour d'autres façons d'utiliser Spark dans Jupyter, consultez notre Apache Spark in Python : Beginner’s Guide.

Tout est en place : créons votre premier programme Spark !

Créer votre premier programme Spark

Commencez par importer SparkSession depuis pyspark.sql et initialisez-le. Rappel : dans le shell interactif, il est créé automatiquement ! Ici, vous ferez un peu plus vous-même :

Importez le module SparkSession puis construisez-le avec builder(). Définissez l'URL du master, le nom de l'application, quelques paramètres comme la mémoire exécuteur, puis getOrCreate() pour récupérer ou créer la session.

# Import SparkSession
from pyspark.sql import SparkSession

# Build the SparkSession
spark = SparkSession.builder \
   .master("local") \
   .appName("Linear Regression Model") \
   .config("spark.executor.memory", "1gb") \
   .getOrCreate()
   
sc = spark.sparkContextPropulsé par

Remarque : si vous obtenez une erreur de type FileNotFoundError du style : « No such file or directory: '/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit' », vous devez (ré)définir votre PATH Spark. Allez au dossier personnel avec $ cd puis éditez .bash_profile avec $ nano .bash_profile.

Ajoutez par exemple en bas du fichier :

export SPARK_HOME="/usr/local/spark"Propulsé par

Quittez avec CTRL + X puis validez l'enregistrement des modifications avec Y. N'oubliez pas d'appliquer les changements avec source .bash_profile.

Astuce : vous pouvez définir d'autres variables d'environnement si besoin. Voici quelques exemples :

# Valeur fixe pour le hachage
export PYTHONHASHSEED=0

# Exécutable Python alternatif
export PYSPARK_PYTHON=/usr/local/ipython/bin/ipython

# Chemin de recherche des bibliothèques partagées
export LD_LIBRARY_PATH=/usr/local/ipython/bin/ipython

# Chemin de recherche des bibliothèques privées 
export PYTHONPATH=$SPARK_HOME/python/lib/py4j-*-src.zip:$PYTHONPATH:$SPARK_HOME/python/Propulsé par

Remarque : vous avez initialisé un SparkSession par défaut. Dans la plupart des cas, vous souhaiterez le configurer davantage, surtout avec de gros volumes de données. Pour en savoir plus, consultez cette page.

Charger vos données

Ce tutoriel utilise le jeu de données California Housing, issu d'un article de 1997 intitulé Sparse Spatial Autoregressions par R. Kelley Pace et Ronald Barry, publié dans Statistics and Probability Letters. Les auteurs l'ont construit à partir du recensement californien de 1990.

Chaque ligne correspond à un groupe d'îlots de recensement, plus petite unité géographique pour laquelle le U.S. Census Bureau publie des échantillons (en général de 600 à 3 000 personnes). Dans cet échantillon, un groupe compte en moyenne 1425,5 individus sur une zone géographique compacte. Vous trouverez ces informations sur cette page ou dans l'article mentionné ici.

Ces données spatiales contiennent 20 640 observations de prix de l'immobilier avec 9 variables économiques :

• Longitude : distance angulaire d'un lieu vers le nord ou le sud de l'équateur terrestre pour chaque groupe ;
• Latitude : distance angulaire d'un lieu vers l'est ou l'ouest de l'équateur terrestre pour chaque groupe ;
• Housing median age : âge médian des habitants du groupe. Remarque : la médiane est la valeur au centre de la distribution ;
• Total rooms : nombre total de pièces des logements par groupe ;
• Total bedrooms : nombre total de chambres des logements par groupe ;
• Population : nombre d'habitants du groupe ;
• Households : unités ménages et leurs occupants par groupe ;
• Median income : revenu médian des personnes du groupe ; et
• Median house value : variable dépendante, valeur médiane des logements par groupe.

Vous apprendrez également que tous les groupes avec des zéros sur les variables indépendantes et dépendantes ont été exclus.

Median house value est la variable cible de votre modèle de ML.

Téléchargez les données ici. Recherchez le dossier houses.zip, téléchargez et décompressez-le pour accéder aux fichiers.

Ensuite, utilisez textFile() pour lire les données depuis le dossier téléchargé vers des RDD. Cette méthode prend l'URI du fichier (ici un chemin local) et lit une collection de lignes. Par commodité, lisez aussi le fichier .domain contenant l'en-tête, pour vérifier l'ordre des variables.

# Load in the data
rdd = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.data')

# Load in the header
header = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.domain')Propulsé par

Exploration des données

Vous avez déjà glané beaucoup d'informations via la page du jeu de données, mais rien ne remplace une inspection directe avec Spark en Python.

Important : l'exécution de Spark est paresseuse : rien n'a encore été exécuté. Les variables rdd et header ne sont que des représentations. Vous devez « pousser » Spark à travailler. Utilisez collect() pour regarder header :

header.collect()Propulsé par

collect() ramène l'intégralité du RDD sur une seule machine. Vous verrez :

[u'longitude: continuous.', u'latitude: continuous.', u'housingMedianAge: continuous. ', u'totalRooms: continuous. ', u'totalBedrooms: continuous. ', u'population: continuous. ', u'households: continuous. ', u'medianIncome: continuous. ', u'medianHouseValue: continuous. ']Propulsé par

Astuce : attention avec collect() ! Cela peut saturer la mémoire du driver. Préférez take() pour n'afficher que quelques éléments. En général, limitez votre jeu de résultats autant que possible, comme en SQL.

Vous confirmez l'ordre des variables vu précédemment et que toutes les colonnes sont continues. Faisons travailler Spark pour vérifier cela sur les données California Housing.

Appelez take() sur votre RDD :

rdd.take(2)Propulsé par

Cette commande prend les 2 premiers éléments du RDD. Le résultat est logique : en lisant avec textFile(), les lignes sont lues telles quelles. Les entrées sont séparées par des virgules et les lignes aussi :

[u'-122.230000,37.880000,41.000000,880.000000,129.000000,322.000000,126.000000,8.325200,452600.000000', u'-122.220000,37.860000,21.000000,7099.000000,1106.000000,2401.000000,1138.000000,8.301400,358500.000000']Propulsé par

Il faut corriger cela. Pas besoin de découper chaque entrée tout de suite, mais les lignes doivent devenir des éléments distincts. Utilisez map() avec une lambda qui scinde la ligne sur la virgule, puis vérifiez avec take() :

Rappel : les lambdas sont des fonctions anonymes créées à l'exécution.

# Split lines on commas
rdd = rdd.map(lambda line: line.split(","))

# Inspect the first 2 lines 
rdd.take(2)Propulsé par

Vous obtenez :

[[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000'], [u'-122.220000', u'37.860000', u'21.000000', u'7099.000000', u'1106.000000', u'2401.000000', u'1138.000000', u'8.301400', u'358500.000000']]Propulsé par

Vous pouvez aussi utiliser :

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)Propulsé par

Si vous avez l'habitude de Pandas ou des data frames en R, vous vous attendiez peut-être à un en-tête ; il n'y en a pas. Pour vous faciliter la vie, passez des RDD aux DataFrames, préférables quand ils sont disponibles, notamment en Python où ils sont plus performants.

Mais quelle est la différence ?

Utilisez des RDD pour des transformations/actions bas niveau sur des données non structurées, sans schéma ni accès par nom de colonne, et quand vous n'avez pas besoin des gains de performance des DataFrames. Privilégiez les RDD si vous manipulez les données avec des constructions fonctionnelles plutôt que des expressions spécifiques au domaine.

En résumé, passons aux DataFrames pour utiliser des expressions de haut niveau, effectuer des requêtes SQL et profiter de l'accès colonne par colonne.

Allons-y.

Première étape : créer un SchemaRDD (RDD de Row) avec un schéma. Comme pour un DataFrame, vous voulez des lignes et des colonnes, avec des types.

Utilisez encore map() et une lambda pour associer chaque entrée à un champ nommé d'une Row. Par exemple, dans la première ligne :

[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000']Propulsé par

La lambda construit une Row dans un SchemaRDD en nommant l'élément d'indice 0 « longitude », etc.

Avec ce SchemaRDD, convertissez facilement le RDD en DataFrame via toDF().

# Import the necessary modules 
from pyspark.sql import Row

# Map the RDD to a DF
df = rdd.map(lambda line: Row(longitude=line[0], 
                              latitude=line[1], 
                              housingMedianAge=line[2],
                              totalRooms=line[3],
                              totalBedRooms=line[4],
                              population=line[5], 
                              households=line[6],
                              medianIncome=line[7],
                              medianHouseValue=line[8])).toDF()Propulsé par

Avec le DataFrame df, inspectez-le avec first(), take(), mais aussi head() et show() :

# Show the top 20 rows 
df.show()Propulsé par

Vous voyez immédiatement une différence nette avec le RDD :

Astuce : utilisez df.columns pour obtenir la liste des colonnes.

Les colonnes semblent correctes, mais qu'en est-il des types ? Spark tente d'inférer un schéma à la lecture, mais y est-il arrivé ? Utilisez df.dtypes ou df.printSchema() pour le savoir.

# Print the data types of all `df` columns
# df.dtypes

# Print the schema of `df`
df.printSchema()Propulsé par

Comme la première ligne n'est pas exécutée, vous obtenez :

root
 |-- households: string (nullable = true)
 |-- housingMedianAge: string (nullable = true)
 |-- latitude: string (nullable = true)
 |-- longitude: string (nullable = true)
 |-- medianHouseValue: string (nullable = true)
 |-- medianIncome: string (nullable = true)
 |-- population: string (nullable = true)
 |-- totalBedRooms: string (nullable = true)
 |-- totalRooms: string (nullable = true)Propulsé par

Toutes les colonnes sont encore en chaîne de caractères … Décevant !

Pour continuer, il faut corriger et affecter des types plus adaptés à chaque colonne, ce qui améliorera aussi les performances. Intuitivement, vous pourriez caster chaque colonne en FloatType() :

from pyspark.sql.types import *

df = df.withColumn("longitude", df["longitude"].cast(FloatType())) \
   .withColumn("latitude", df["latitude"].cast(FloatType())) \
   .withColumn("housingMedianAge",df["housingMedianAge"].cast(FloatType())) \
   .withColumn("totalRooms", df["totalRooms"].cast(FloatType())) \ 
   .withColumn("totalBedRooms", df["totalBedRooms"].cast(FloatType())) \ 
   .withColumn("population", df["population"].cast(FloatType())) \ 
   .withColumn("households", df["households"].cast(FloatType())) \ 
   .withColumn("medianIncome", df["medianIncome"].cast(FloatType())) \ 
   .withColumn("medianHouseValue", df["medianHouseValue"].cast(FloatType()))Propulsé par

Mais ces appels répétitifs sont verbeux et propices aux erreurs. Écrivons une fonction plus propre pour tout faire :

La fonction suivante prend un DataFrame, des noms de colonnes et le nouveau type souhaité. Pour chaque colonne, elle caste vers le nouveau type puis renvoie le DataFrame :

# Import all from `sql.types`
from pyspark.sql.types import *

# Write a custom function to convert the data type of DataFrame columns
def convertColumn(df, names, newType):
  for name in names: 
     df = df.withColumn(name, df[name].cast(newType))
  return df 

# Assign all column names to `columns`
columns = ['households', 'housingMedianAge', 'latitude', 'longitude', 'medianHouseValue', 'medianIncome', 'population', 'totalBedRooms', 'totalRooms']

# Conver the `df` columns to `FloatType()`
df = convertColumn(df, columns, FloatType())Propulsé par

Déjà plus lisible ! Vérifiez rapidement les types avec printSchema(), comme précédemment.

Maintenant, passons à l'exploration. L'accès colonne et les requêtes SQL sont deux atouts des DataFrames. Commençons petit : sélectionnez deux colonnes de df et affichez 10 lignes :

df.select('population','totalBedRooms').show(10)Propulsé par

Cette requête renvoie :

+----------+-------------+
|population|totalBedRooms|
+----------+-------------+
|     322.0|        129.0|
|    2401.0|       1106.0|
|     496.0|        190.0|
|     558.0|        235.0|
|     565.0|        280.0|
|     413.0|        213.0|
|    1094.0|        489.0|
|    1157.0|        687.0|
|    1206.0|        665.0|
|    1551.0|        707.0|
+----------+-------------+
only showing top 10 rowsPropulsé par

Vous pouvez complexifier, comme ci-dessous :

df.groupBy("housingMedianAge").count().sort("housingMedianAge",ascending=False).show()Propulsé par

Ce qui donne :

+----------------+-----+                                                        
|housingMedianAge|count|
+----------------+-----+
|            52.0| 1273|
|            51.0|   48|
|            50.0|  136|
|            49.0|  134|
|            48.0|  177|
|            47.0|  198|
|            46.0|  245|
|            45.0|  294|
|            44.0|  356|
|            43.0|  353|
|            42.0|  368|
|            41.0|  296|
|            40.0|  304|
|            39.0|  369|
|            38.0|  394|
|            37.0|  537|
|            36.0|  862|
|            35.0|  824|
|            34.0|  689|
|            33.0|  615|
+----------------+-----+
only showing top 20 rowsPropulsé par

Au-delà des requêtes, vous pouvez aussi décrire vos données et obtenir des statistiques résumées ; cela vous aidera pour la suite :

df.describe().show()Propulsé par

Observez les minimums et maximums des attributs numériques : plusieurs ont une large plage de valeurs. Vous devrez normaliser votre jeu de données.

Prétraitement des données

Avec ces premiers enseignements, vous pouvez prétraiter les données pour les injecter dans le modèle.

• Pas de valeurs manquantes à gérer : les zéros ont été exclus du jeu de données.
• Vous devrez probablement standardiser, car les écarts min/max sont importants.
• Vous pouvez créer des attributs supplémentaires (chambres par pièce, pièces par ménage, etc.).
• La variable cible a de grandes valeurs ; pour vous simplifier la vie, ajustez-les légèrement.

# Import all from `sql.functions` 
from pyspark.sql.functions import *

# Adjust the values of `medianHouseValue`
df = df.withColumn("medianHouseValue", col("medianHouseValue")/100000)

# Show the first 2 lines of `df`
df.take(2)Propulsé par

[Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0), Row(households=1138.0, housingMedianAge=21.0, latitude=37.86000061035156, longitude=-122.22000122070312, medianHouseValue=3.585, medianIncome=8.301400184631348, population=2401.0, totalBedRooms=1106.0, totalRooms=7099.0)]Propulsé par

# Import all from `sql.functions` if you haven't yet
from pyspark.sql.functions import *

# Divide `totalRooms` by `households`
roomsPerHousehold = df.select(col("totalRooms")/col("households"))

# Divide `population` by `households`
populationPerHousehold = df.select(col("population")/col("households"))

# Divide `totalBedRooms` by `totalRooms`
bedroomsPerRoom = df.select(col("totalBedRooms")/col("totalRooms"))

# Add the new columns to `df`
df = df.withColumn("roomsPerHousehold", col("totalRooms")/col("households")) \
   .withColumn("populationPerHousehold", col("population")/col("households")) \
   .withColumn("bedroomsPerRoom", col("totalBedRooms")/col("totalRooms"))
   
# Inspect the result
df.first()Propulsé par

Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0, roomsPerHousehold=6.984126984126984, populationPerHousehold=2.5555555555555554, bedroomsPerRoom=0.14659090909090908)Propulsé par

# Re-order and select columns
df = df.select("medianHouseValue", 
              "totalBedRooms", 
              "population", 
              "households", 
              "medianIncome", 
              "roomsPerHousehold", 
              "populationPerHousehold", 
              "bedroomsPerRoom")Propulsé par

# Import `DenseVector`
from pyspark.ml.linalg import DenseVector

# Define the `input_data` 
input_data = df.rdd.map(lambda x: (x[0], DenseVector(x[1:])))

# Replace `df` with the new DataFrame
df = spark.createDataFrame(input_data, ["label", "features"])Propulsé par

# Import `StandardScaler` 
from pyspark.ml.feature import StandardScaler

# Initialize the `standardScaler`
standardScaler = StandardScaler(inputCol="features", outputCol="features_scaled")

# Fit the DataFrame to the scaler
scaler = standardScaler.fit(df)

# Transform the data in `df` with the scaler
scaled_df = scaler.transform(df)

# Inspect the result
scaled_df.take(2)Propulsé par

[Row(label=4.526, features=DenseVector([129.0, 322.0, 126.0, 8.3252, 6.9841, 2.5556, 0.1466]), features_scaled=DenseVector([0.3062, 0.2843, 0.3296, 4.3821, 2.8228, 0.2461, 2.5264])), Row(label=3.585, features=DenseVector([1106.0, 2401.0, 1138.0, 8.3014, 6.2381, 2.1098, 0.1558]), features_scaled=DenseVector([2.6255, 2.1202, 2.9765, 4.3696, 2.5213, 0.2031, 2.6851]))]Propulsé par

# Split the data into train and test sets
train_data, test_data = scaled_df.randomSplit([.8,.2],seed=1234)Propulsé par

# Import `LinearRegression`
from pyspark.ml.regression import LinearRegression

# Initialize `lr`
lr = LinearRegression(labelCol="label", maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the data to the model
linearModel = lr.fit(train_data)Propulsé par

# Generate predictions
predicted = linearModel.transform(test_data)

# Extract the predictions and the "known" correct labels
predictions = predicted.select("prediction").rdd.map(lambda x: x[0])
labels = predicted.select("label").rdd.map(lambda x: x[0])

# Zip `predictions` and `labels` into a list
predictionAndLabel = predictions.zip(labels).collect()

# Print out first 5 instances of `predictionAndLabel` 
predictionAndLabel[:5]Propulsé par

[(1.4491508524918457, 0.14999), (1.5705029404692372, 0.14999), (2.148727956912464, 0.14999), (1.5831547768979277, 0.344), (1.5182107797955968, 0.398)]Propulsé par

# Coefficients for the model
linearModel.coefficients

# Intercept for the model
linearModel.interceptPropulsé par

# The coefficients
[0.0,0.0,0.0,0.276239709215,0.0,0.0,0.0]

# The intercept
0.990399577462Propulsé par

# Get the RMSE
linearModel.summary.rootMeanSquaredError

# Get the R2
linearModel.summary.r2Propulsé par

# RMSE
0.8692118678997669

# R2
0.4240895287218379
Propulsé par

spark.stop()Propulsé par