Big Data avec PostgreSQL et Apache Spark

PostgreSQL est bien connu comme la base de données open source la plus avancée, et il vous aide à gérer vos données, peu importe la taille, la taille ou la différence de l'ensemble de données, afin que vous puissiez l'utiliser pour gérer ou analyser vos mégadonnées, et bien sûr, il y a plusieurs façons de rendre cela possible, par exemple Apache Spark. Dans ce blog, nous verrons ce qu'est Apache Spark et comment nous pouvons l'utiliser pour travailler avec notre base de données PostgreSQL.

Pour l'analyse de données volumineuses, nous proposons deux types d'analyse différents :

Analyse par lots :basée sur les données collectées sur une période donnée.
Analyse en temps réel (flux) :basée sur des données immédiates pour un résultat instantané.

Qu'est-ce qu'Apache Spark ?

Apache Spark est un moteur d'analyse unifié pour le traitement de données à grande échelle qui peut fonctionner à la fois sur des analyses par lots et en temps réel de manière plus rapide et plus simple.

Il fournit des API de haut niveau en Java, Scala, Python et R, ainsi qu'un moteur optimisé prenant en charge les graphiques d'exécution généraux.

Composants Apache Spark

Bibliothèques Apache Spark

Apache Spark comprend différentes bibliothèques :

Spark SQL :il s'agit d'un module permettant de travailler avec des données structurées à l'aide de SQL ou d'une API DataFrame. Il fournit un moyen commun d'accéder à diverses sources de données, notamment Hive, Avro, Parquet, ORC, JSON et JDBC. Vous pouvez même joindre des données de ces sources.
Spark Streaming :il facilite la création d'applications de streaming évolutives et tolérantes aux pannes à l'aide d'une API intégrée au langage pour le traitement des flux, ce qui vous permet d'écrire des tâches de streaming de la même manière que vous écrivez des tâches par lots. Il prend en charge Java, Scala et Python. Spark Streaming récupère à la fois le travail perdu et l'état de l'opérateur, sans aucun code supplémentaire de votre part. Il vous permet de réutiliser le même code pour le traitement par lots, de joindre des flux à des données historiques ou d'exécuter des requêtes ad hoc sur l'état du flux.
MLib (Machine Learning) :il s'agit d'une bibliothèque de machine learning évolutive. MLlib contient des algorithmes de haute qualité qui exploitent l'itération et peuvent donner de meilleurs résultats que les approximations en une passe parfois utilisées sur MapReduce.
GraphX :il s'agit d'une API pour les graphes et le calcul parallèle aux graphes. GraphX unifie l'ETL, l'analyse exploratoire et le calcul de graphe itératif au sein d'un système unique. Vous pouvez afficher les mêmes données sous forme de graphiques et de collections, transformer et joindre efficacement des graphiques avec des RDD et écrire des algorithmes de graphique itératifs personnalisés à l'aide de l'API Pregel.

Avantages d'Apache Spark

Selon la documentation officielle, certains avantages d'Apache Spark sont :

Vitesse :exécutez les charges de travail 100 fois plus rapidement. Apache Spark atteint des performances élevées pour les données par lots et en continu, à l'aide d'un planificateur DAG (Direct Acyclic Graph) de pointe, d'un optimiseur de requêtes et d'un moteur d'exécution physique.
Facilité d'utilisation :écrivez rapidement des applications en Java, Scala, Python, R et SQL. Spark propose plus de 80 opérateurs de haut niveau qui facilitent la création d'applications parallèles. Vous pouvez l'utiliser de manière interactive à partir des shells Scala, Python, R et SQL.
Généralité :combinez SQL, le streaming et des analyses complexes. Spark alimente une pile de bibliothèques, notamment SQL et DataFrames, MLlib pour l'apprentissage automatique, GraphX et Spark Streaming. Vous pouvez combiner ces bibliothèques de manière transparente dans la même application.
S'exécute partout :Spark s'exécute sur Hadoop, Apache Mesos, Kubernetes, de manière autonome ou dans le cloud. Il peut accéder à diverses sources de données. Vous pouvez exécuter Spark en mode cluster autonome, sur EC2, sur Hadoop YARN, sur Mesos ou sur Kubernetes. Accédez aux données dans HDFS, Alluxio, Apache Cassandra, Apache HBase, Apache Hive et des centaines d'autres sources de données.

Voyons maintenant comment nous pouvons intégrer cela à notre base de données PostgreSQL.

Comment utiliser Apache Spark avec PostgreSQL

Nous supposerons que votre cluster PostgreSQL est opérationnel. Pour cette tâche, nous utiliserons un serveur PostgreSQL 11 exécuté sur CentOS7.

Commençons par créer notre base de données de test sur notre serveur PostgreSQL :

postgres=# CREATE DATABASE testing;
CREATE DATABASE
postgres=# \c testing
You are now connected to database "testing" as user "postgres".

Maintenant, nous allons créer une table appelée t1 :

testing=# CREATE TABLE t1 (id int, name text);
CREATE TABLE

Et insérez-y quelques données :

testing=# INSERT INTO t1 VALUES (1,'name1');
INSERT 0 1
testing=# INSERT INTO t1 VALUES (2,'name2');
INSERT 0 1

Vérifiez les données créées :

testing=# SELECT * FROM t1;
 id | name
----+-------
  1 | name1
  2 | name2
(2 rows)

Pour connecter Apache Spark à notre base de données PostgreSQL, nous allons utiliser un connecteur JDBC. Vous pouvez le télécharger ici.

$ wget https://jdbc.postgresql.org/download/postgresql-42.2.6.jar

Maintenant, installons Apache Spark. Pour cela, nous devons télécharger les packages Spark à partir d'ici.

$ wget http://us.mirrors.quenda.co/apache/spark/spark-2.4.3/spark-2.4.3-bin-hadoop2.7.tgz
$ tar zxvf spark-2.4.3-bin-hadoop2.7.tgz
$ cd spark-2.4.3-bin-hadoop2.7/

Pour exécuter le shell Spark, nous aurons besoin de JAVA installé sur notre serveur :

$  yum install java

Alors maintenant, nous pouvons exécuter notre Spark Shell :

$ ./bin/spark-shell
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).
Spark context Web UI available at http://ApacheSpark1:4040
Spark context available as 'sc' (master = local[*], app id = local-1563907528854).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.3
      /_/

Using Scala version 2.11.12 (OpenJDK 64-Bit Server VM, Java 1.8.0_212)
Type in expressions to have them evaluated.
Type :help for more information.

scala>

Nous pouvons accéder à notre interface Web contextuelle Spark disponible dans le port 4040 sur notre serveur :

Interface utilisateur Apache Spark

Dans le shell Spark, nous devons ajouter le pilote PostgreSQL JDBC :

scala> :require /path/to/postgresql-42.2.6.jar
Added '/path/to/postgresql-42.2.6.jar' to classpath.
scala> import java.util.Properties
import java.util.Properties

Et ajoutez les informations JDBC à utiliser par Spark :

scala> val url = "jdbc:postgresql://localhost:5432/testing"
url: String = jdbc:postgresql://localhost:5432/testing
scala> val connectionProperties = new Properties()
connectionProperties: java.util.Properties = {}
scala> connectionProperties.setProperty("Driver", "org.postgresql.Driver")
res6: Object = null

Maintenant, nous pouvons exécuter des requêtes SQL. Tout d'abord, définissons query1 comme SELECT * FROM t1, notre table de test.

scala> val query1 = "(SELECT * FROM t1) as q1"
query1: String = (SELECT * FROM t1) as q1

Et créez le DataFrame :

scala> val query1df = spark.read.jdbc(url, query1, connectionProperties)
query1df: org.apache.spark.sql.DataFrame = [id: int, name: string]

Alors maintenant, nous pouvons effectuer une action sur ce DataFrame :

scala> query1df.show()
+---+-----+
| id| name|
+---+-----+
|  1|name1|
|  2|name2|
+---+-----+

scala> query1df.explain
== Physical Plan ==
*(1) Scan JDBCRelation((SELECT * FROM t1) as q1) [numPartitions=1] [id#19,name#20] PushedFilters: [], ReadSchema: struct<id:int,name:string>

Nous pouvons ajouter plus de valeurs et l'exécuter à nouveau juste pour confirmer qu'il renvoie les valeurs actuelles.

PostgreSQL

testing=# INSERT INTO t1 VALUES (10,'name10'), (11,'name11'), (12,'name12'), (13,'name13'), (14,'name14'), (15,'name15');
INSERT 0 6
testing=# SELECT * FROM t1;
 id |  name
----+--------
  1 | name1
  2 | name2
 10 | name10
 11 | name11
 12 | name12
 13 | name13
 14 | name14
 15 | name15
(8 rows)

Étincelle

scala> query1df.show()
+---+------+
| id|  name|
+---+------+
|  1| name1|
|  2| name2|
| 10|name10|
| 11|name11|
| 12|name12|
| 13|name13|
| 14|name14|
| 15|name15|
+---+------+

Dans notre exemple, nous montrons uniquement comment Apache Spark fonctionne avec notre base de données PostgreSQL, pas comment il gère nos informations Big Data.

Conclusion

De nos jours, il est assez courant d'avoir le défi de gérer le Big Data dans une entreprise, et comme nous avons pu le voir, nous pouvons utiliser Apache Spark pour y faire face et utiliser toutes les fonctionnalités que nous avons mentionnées précédemment. Le Big Data est un monde immense, vous pouvez donc consulter la documentation officielle pour plus d'informations sur l'utilisation d'Apache Spark et de PostgreSQL et l'adapter à vos besoins.