Introduction à Apache Spark avec exemples et cas d'utilisation

J'ai entendu parler de Spark pour la première fois fin 2013 lorsque je me suis intéressé à Scala, le langage dans lequel Spark est écrit. Quelque temps plus tard, j'ai réalisé un projet de science des données amusant en essayant de prédire la survie sur le Titanic. Cela s'est avéré être un excellent moyen de se familiariser davantage avec les concepts et la programmation de Spark. Je le recommande vivement à tous les développeurs Spark en herbe à la recherche d'un endroit pour commencer.

Aujourd'hui, Spark est adopté par des acteurs majeurs comme Amazon, eBay et Yahoo! De nombreuses organisations exécutent Spark sur des clusters avec des milliers de nœuds. Selon la FAQ Spark, le plus grand cluster connu compte plus de 8 000 nœuds. En effet, Spark est une technologie qu'il vaut la peine de prendre en compte et d'apprendre.

Cet article fournit une introduction à Spark, y compris des cas d'utilisation et des exemples. Il contient des informations du site Web Apache Spark ainsi que du livre Learning Spark - Lightning-Fast Big Data Analysis.

Qu'est-ce qu'Apache Spark ? Une introduction

Spark est un projet Apache annoncé comme "l'informatique en cluster rapide comme l'éclair". Il a une communauté open source florissante et est le projet Apache le plus actif en ce moment.

Spark fournit une plate-forme de traitement de données plus rapide et plus générale. Spark vous permet d'exécuter des programmes jusqu'à 100 fois plus rapidement en mémoire ou 10 fois plus rapidement sur disque que Hadoop. L'année dernière, Spark a repris Hadoop en complétant le concours 100 To Daytona GraySort 3 fois plus vite sur un dixième du nombre de machines et il est également devenu le moteur open source le plus rapide pour trier un pétaoctet.

Spark permet également d'écrire du code plus rapidement puisque vous avez à votre disposition plus de 80 opérateurs de haut niveau. Pour le démontrer, regardons le "Hello World!" du BigData : l'exemple du nombre de mots. Écrit en Java pour MapReduce, il contient environ 50 lignes de code, alors que dans Spark (et Scala), vous pouvez le faire aussi simplement que ceci :

 sparkContext.textFile("hdfs://...") .flatMap(line => line.split(" ")) .map(word => (word, 1)).reduceByKey(_ + _) .saveAsTextFile("hdfs://...")

Un autre aspect important lors de l'apprentissage de l'utilisation d'Apache Spark est le shell interactif (REPL) qu'il fournit prêt à l'emploi. En utilisant REPL, on peut tester le résultat de chaque ligne de code sans avoir besoin de coder et d'exécuter tout le travail. Le chemin vers le code fonctionnel est donc beaucoup plus court et l'analyse de données ad hoc est rendue possible.

Les fonctionnalités clés supplémentaires de Spark incluent :

• Fournit actuellement des API en Scala, Java et Python, avec la prise en charge d'autres langages (tels que R) en cours de route
• S'intègre bien à l'écosystème Hadoop et aux sources de données (HDFS, Amazon S3, Hive, HBase, Cassandra, etc.)
• Peut fonctionner sur des clusters gérés par Hadoop YARN ou Apache Mesos, et peut également fonctionner de manière autonome

Le noyau Spark est complété par un ensemble de bibliothèques puissantes de niveau supérieur qui peuvent être utilisées de manière transparente dans la même application. Ces bibliothèques incluent actuellement SparkSQL, Spark Streaming, MLlib (pour l'apprentissage automatique) et GraphX, chacune étant détaillée dans cet article. Des bibliothèques et extensions Spark supplémentaires sont également en cours de développement.

Noyau d'étincelle

Spark Core est le moteur de base pour le traitement de données parallèle et distribué à grande échelle. Il est responsable de :

• gestion de la mémoire et récupération des pannes
• planification, distribution et surveillance des tâches sur un cluster
• interaction avec les systèmes de stockage

Spark introduit le concept d'un RDD (Resilient Distributed Dataset), une collection d'objets distribuée, immuable et tolérante aux pannes, qui peut être utilisée en parallèle. Un RDD peut contenir n'importe quel type d'objet et est créé en chargeant un jeu de données externe ou en distribuant une collection à partir du programme pilote.

Les RDD prennent en charge deux types d'opérations :

• Les transformations sont des opérations (telles que mappage, filtre, jointure, union, etc.) qui sont effectuées sur un RDD et qui génèrent un nouveau RDD contenant le résultat.
• Les actions sont des opérations (telles que réduire, compter, premier, etc.) qui renvoient une valeur après avoir exécuté un calcul sur un RDD.

Les transformations dans Spark sont "paresseuses", ce qui signifie qu'elles ne calculent pas leurs résultats tout de suite. Au lieu de cela, ils "se souviennent" simplement de l'opération à effectuer et de l'ensemble de données (par exemple, un fichier) sur lequel l'opération doit être effectuée. Les transformations ne sont réellement calculées que lorsqu'une action est appelée et que le résultat est renvoyé au programme pilote. Cette conception permet à Spark de fonctionner plus efficacement. Par exemple, si un gros fichier était transformé de différentes manières et transmis à la première action, Spark ne traiterait et ne renverrait le résultat que pour la première ligne, plutôt que de faire le travail pour l'ensemble du fichier.

Par défaut, chaque RDD transformé peut être recalculé chaque fois que vous exécutez une action dessus. Cependant, vous pouvez également conserver un RDD en mémoire à l'aide de la méthode de persistance ou de cache, auquel cas Spark conservera les éléments sur le cluster pour un accès beaucoup plus rapide la prochaine fois que vous l'interrogerez.

SparkSQL

SparkSQL est un composant Spark qui prend en charge l'interrogation de données via SQL ou via le langage de requête Hive. Il s'agissait à l'origine du port Apache Hive pour s'exécuter au-dessus de Spark (à la place de MapReduce) et est maintenant intégré à la pile Spark. En plus de fournir un support pour diverses sources de données, il permet de tisser des requêtes SQL avec des transformations de code, ce qui se traduit par un outil très puissant. Voici un exemple de requête compatible Hive :

 // sc is an existing SparkContext. val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc) sqlContext.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") sqlContext.sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src") // Queries are expressed in HiveQL sqlContext.sql("FROM src SELECT key, value").collect().foreach(println)

Diffusion d'étincelles

Spark Streaming prend en charge le traitement en temps réel des données de streaming, telles que les fichiers journaux du serveur Web de production (par exemple Apache Flume et HDFS/S3), les médias sociaux comme Twitter et diverses files d'attente de messagerie comme Kafka. Sous le capot, Spark Streaming reçoit les flux de données d'entrée et divise les données en lots. Ensuite, ils sont traités par le moteur Spark et génèrent le flux final de résultats par lots, comme illustré ci-dessous.

L'API Spark Streaming correspond étroitement à celle de Spark Core, ce qui permet aux programmeurs de travailler facilement dans le monde des données par lots et en continu.

MLlib

MLlib est une bibliothèque d'apprentissage automatique qui fournit divers algorithmes conçus pour évoluer sur un cluster pour la classification, la régression, le clustering, le filtrage collaboratif, etc. (consultez l'article de Toptal sur l'apprentissage automatique pour plus d'informations sur ce sujet). Certains de ces algorithmes fonctionnent également avec des données en continu, telles que la régression linéaire utilisant les moindres carrés ordinaires ou le clustering k-means (et plus encore). Apache Mahout (une bibliothèque d'apprentissage automatique pour Hadoop) s'est déjà détourné de MapReduce et s'est associé à Spark MLlib.

GraphX

GraphX ​​est une bibliothèque permettant de manipuler des graphiques et d'effectuer des opérations parallèles aux graphiques. Il fournit un outil uniforme pour l'ETL, l'analyse exploratoire et les calculs de graphes itératifs. Outre les opérations intégrées pour la manipulation de graphes, il fournit une bibliothèque d'algorithmes de graphes courants tels que PageRank.

Comment utiliser Apache Spark : cas d'utilisation de la détection d'événements

Maintenant que nous avons répondu à la question « Qu'est-ce qu'Apache Spark ? », réfléchissons au type de problèmes ou de défis pour lesquels il pourrait être utilisé le plus efficacement.

Je suis tombé récemment sur un article sur une expérience visant à détecter un tremblement de terre en analysant un flux Twitter. Fait intéressant, il a été démontré que cette technique était susceptible de vous informer d'un tremblement de terre au Japon plus rapidement que l'Agence météorologique japonaise. Même s'ils ont utilisé une technologie différente dans leur article, je pense que c'est un excellent exemple pour voir comment nous pourrions utiliser Spark avec des extraits de code simplifiés et sans le code de colle.

Tout d'abord, il faudrait filtrer les tweets qui semblent pertinents comme "tremblement de terre" ou "tremblement". Nous pourrions facilement utiliser Spark Streaming à cette fin comme suit :

 TwitterUtils.createStream(...) .filter(_.getText.contains("earthquake") || _.getText.contains("shaking"))

Ensuite, nous devrons effectuer une analyse sémantique des tweets pour déterminer s'ils semblent faire référence à un tremblement de terre actuel. Des tweets comme "Tremblement de terre !" ou "Maintenant ça tremble", par exemple, seraient considérés comme des correspondances positives, alors que des tweets comme "Participer à une conférence sur le tremblement de terre" ou "Le tremblement de terre d'hier était effrayant" ne le seraient pas. Les auteurs de l'article ont utilisé une machine à vecteurs de support (SVM) à cette fin. Nous ferons la même chose ici, mais nous pouvons également essayer une version en streaming. Un exemple de code résultant de MLlib ressemblerait à ceci :

 // We would prepare some earthquake tweet data and load it in LIBSVM format. val data = MLUtils.loadLibSVMFile(sc, "sample_earthquate_tweets.txt") // Split data into training (60%) and test (40%). val splits = data.randomSplit(Array(0.6, 0.4), seed = 11L) val training = splits(0).cache() val test = splits(1) // Run training algorithm to build the model val numIterations = 100 val model = SVMWithSGD.train(training, numIterations) // Clear the default threshold. model.clearThreshold() // Compute raw scores on the test set. val scoreAndLabels = test.map { point => val score = model.predict(point.features) (score, point.label) } // Get evaluation metrics. val metrics = new BinaryClassificationMetrics(scoreAndLabels) val auROC = metrics.areaUnderROC() println("Area under ROC = " + auROC)

Si nous sommes satisfaits du taux de prédiction du modèle, nous pourrions passer à l'étape suivante et réagir chaque fois que nous découvrons un tremblement de terre. Pour en détecter un, nous avons besoin d'un certain nombre (c'est-à-dire la densité) de tweets positifs dans une fenêtre de temps définie (comme décrit dans l'article). Notez que, pour les tweets avec les services de localisation Twitter activés, nous extrairions également l'emplacement du tremblement de terre. Forts de ces connaissances, nous pourrions utiliser SparkSQL et interroger une table Hive existante (stockant les utilisateurs intéressés à recevoir des notifications de tremblement de terre) pour récupérer leurs adresses e-mail et leur envoyer un e-mail d'avertissement personnalisé, comme suit :

 // sc is an existing SparkContext. val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc) // sendEmail is a custom function sqlContext.sql("FROM earthquake_warning_users SELECT firstName, lastName, city, email") .collect().foreach(sendEmail)

Autres cas d'utilisation d'Apache Spark

Les cas d'utilisation potentiels de Spark vont bien au-delà de la détection des tremblements de terre bien sûr.

Voici un échantillon rapide (mais certainement loin d'être exhaustif !) D'autres cas d'utilisation qui nécessitent de gérer la vitesse, la variété et le volume du Big Data, pour lequel Spark est si bien adapté :

Dans l'industrie du jeu, le traitement et la découverte de schémas à partir du tuyau d'incendie potentiel des événements en temps réel dans le jeu et la capacité d'y répondre immédiatement est une capacité qui pourrait générer une activité lucrative, à des fins telles que la rétention des joueurs, la publicité ciblée, l'auto -ajustement du niveau de complexité, et ainsi de suite.

Dans l'industrie du commerce électronique, les informations de transaction en temps réel pourraient être transmises à un algorithme de clustering en continu comme k-means ou à un filtrage collaboratif comme ALS. Les résultats pourraient même être combinés avec d'autres sources de données non structurées, telles que les commentaires des clients ou les avis sur les produits, et utilisés pour améliorer et adapter constamment les recommandations au fil du temps avec les nouvelles tendances.

Dans le secteur de la finance ou de la sécurité, la pile Spark pourrait être appliquée à un système de détection de fraude ou d'intrusion ou à une authentification basée sur les risques. Il pourrait obtenir des résultats de premier ordre en récoltant d'énormes quantités de journaux archivés, en les combinant avec des sources de données externes telles que des informations sur les violations de données et les comptes compromis (voir, par exemple, https://haveibeenpwned.com/) et des informations de la connexion/ demande telle que la géolocalisation IP ou l'heure.

Conclusion

En résumé, Spark aide à simplifier la tâche difficile et gourmande en calculs de traitement de gros volumes de données en temps réel ou archivées, structurées et non structurées, en intégrant de manière transparente des capacités complexes pertinentes telles que l'apprentissage automatique et les algorithmes de graphes. Spark apporte le traitement du Big Data aux masses. Vérifiez-le!