Големи данни с PostgreSQL и Apache Spark

PostgreSQL е добре известен като най-модерната база данни с отворен код и ви помага да управлявате данните си, без значение колко голям, малък или различен е наборът от данни, така че можете да го използвате за управление или анализ на вашите големи данни, и разбира се, има няколко начина да направите това възможно, например Apache Spark. В този блог ще видим какво представлява Apache Spark и как можем да го използваме за работа с нашата PostgreSQL база данни.

За анализ на големи данни имаме два различни типа анализи:

Пакетни анализи:Въз основа на данните, събрани за определен период от време.
Аналитика в реално време (поток):Въз основа на незабавни данни за незабавен резултат.

Какво е Apache Spark?

Apache Spark е унифициран аналитичен двигател за широкомащабна обработка на данни, който може да работи както върху пакетни анализи, така и в реално време по по-бърз и по-лесен начин.

Той предоставя API на високо ниво в Java, Scala, Python и R и оптимизиран двигател, който поддържа общи графики за изпълнение.

Компоненти на Apache Spark

Библиотеки на Apache Spark

Apache Spark включва различни библиотеки:

Spark SQL:Това е модул за работа със структурирани данни с помощта на SQL или API на DataFrame. Той предоставя общ начин за достъп до различни източници на данни, включително Hive, Avro, Parquet, ORC, JSON и JDBC. Можете дори да обедините данни от тези източници.
Spark Streaming:Улеснява изграждането на мащабируеми устойчиви на грешки приложения за стрийминг с помощта на интегриран в език API за обработка на поточно предаване, което ви позволява да пишете задачи за поточно предаване по същия начин, по който пишете пакетни задания. Поддържа Java, Scala и Python. Spark Streaming възстановява както загубената работа, така и състоянието на оператора извън кутията, без допълнителен код от ваша страна. Позволява ви да използвате повторно същия код за пакетна обработка, да обединявате потоци срещу исторически данни или да изпълнявате ad-hoc заявки в състоянието на потока.
MLib (Машинно обучение):Това е мащабируема библиотека за машинно обучение. MLlib съдържа висококачествени алгоритми, които използват итерация и могат да дадат по-добри резултати от еднопроходните приближения, използвани понякога в MapReduce.
GraphX:Това е API за графики и графично паралелно изчисление. GraphX обединява ETL, проучвателен анализ и итеративно изчисление на графики в рамките на една система. Можете да преглеждате едни и същи данни като графики и колекции, ефективно да трансформирате и обединявате графики с RDD и да пишете персонализирани итеративни графични алгоритми с помощта на API на Pregel.

Предимства на Apache Spark

Според официалната документация някои предимства на Apache Spark са:

Скорост:Изпълнявайте работни натоварвания 100 пъти по-бързо. Apache Spark постига висока производителност както за пакетни, така и за поточно предаване, като използва най-съвременен DAG (Direct Acyclic Graph) планировчик, оптимизатор на заявки и двигател за физическо изпълнение.
Леснота за използване:Пишете бързо приложения в Java, Scala, Python, R и SQL. Spark предлага над 80 оператора на високо ниво, които улесняват изграждането на паралелни приложения. Можете да го използвате интерактивно от Scala, Python, R и SQL обвивки.
Общ характер:Комбинирайте SQL, поточно предаване и сложни анализи. Spark захранва набор от библиотеки, включително SQL и DataFrames, MLlib за машинно обучение, GraphX и Spark Streaming. Можете да комбинирате тези библиотеки безпроблемно в едно и също приложение.
Изпълнява се навсякъде:Spark работи на Hadoop, Apache Mesos, Kubernetes, самостоятелно или в облака. Има достъп до различни източници на данни. Можете да стартирате Spark, като използвате неговия самостоятелен клъстерен режим, на EC2, на Hadoop YARN, на Mesos или на Kubernetes. Достъп до данни в HDFS, Alluxio, Apache Cassandra, Apache HBase, Apache Hive и стотици други източници на данни.

Сега нека видим как можем да интегрираме това с нашата PostgreSQL база данни.

Как да използвате Apache Spark с PostgreSQL

Ще приемем, че вашият PostgreSQL клъстер е стартиран и работи. За тази задача ще използваме PostgreSQL 11 сървър, работещ на CentOS7.

Първо, нека създадем нашата тестова база данни на нашия PostgreSQL сървър:

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

Сега ще създадем таблица, наречена t1:

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

И вмъкнете там някои данни:

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

Проверете създадените данни:

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

За да свържем Apache Spark към нашата PostgreSQL база данни, ще използваме JDBC конектор. Можете да го изтеглите от тук.

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

Сега нека инсталираме Apache Spark. За целта трябва да изтеглим пакетите spark от тук.

$ 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/

За да стартираме Spark shell, ще ни трябва JAVA, инсталирана на нашия сървър:

$  yum install java

Така че сега можем да стартираме нашия 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>

Имаме достъп до нашия контекстен уеб интерфейс на Spark, наличен в порт 4040 на нашия сървър:

Потребителски интерфейс на Apache Spark

В Spark черупката трябва да добавим 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

И добавете JDBC информацията, която да се използва от 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

Сега можем да изпълняваме SQL заявки. Първо, нека дефинираме query1 като SELECT * FROM t1, нашата тестова таблица.

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

И създайте DataFrame:

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

Така че сега можем да извършим действие над този 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>

Можем да добавим още стойности и да го стартираме отново, само за да потвърдим, че връща текущите стойности.

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)

Искра

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

В нашия пример показваме само как Apache Spark работи с нашата PostgreSQL база данни, а не как управлява нашата информация за големи данни.

Заключение

В днешно време е доста често срещано предизвикателство да управляваме големи данни в една компания и както видяхме, можем да използваме Apache Spark, за да се справим с него и да използваме всички функции, които споменахме по-рано. Големите данни са огромен свят, така че можете да проверите официалната документация за повече информация относно използването на Apache Spark и PostgreSQL и да я съобразите с вашите изисквания.