Apache Spark je najúspešnejší softvér Apache Software Foundation a je navrhnutý pre rýchle výpočty. Niekoľko priemyselných odvetví používa Apache Spark na nájdenie svojich riešení. PySpark SQL je modul v Spark, ktorý integruje relačné spracovanie s funkčným programovacím API Spark. Údaje môžeme extrahovať pomocou dotazovacieho jazyka SQL. Dotazy môžeme použiť rovnako ako jazyk SQL.
Ak máte základné znalosti o RDBMS, PySpark SQL bude ľahko použiteľný, kde môžete rozšíriť obmedzenia tradičného spracovania relačných údajov. Spark podporuje aj Hive Query Language, ale existujú obmedzenia databázy Hive. Spark SQL bol vyvinutý na odstránenie nedostatkov databázy Hive. Pozrime sa na nasledujúce nevýhody Hive:
Nevýhody Hive
- Nemôže obnoviť spracovanie, čo znamená, že ak spustenie zlyhá uprostred pracovného toku, nemôžete pokračovať od miesta, kde sa zaseklo.
- Keď je povolený kôš, nemôžeme zahodiť šifrované databázy do kaskády. Vedie to k chybe vykonávania. Na zrušenie takéhoto typu databázy musia používatelia použiť možnosť Vyčistiť.
- Dotazy ad-hoc sa vykonávajú pomocou MapReduce, ktorý spúšťa úľ, ale keď analyzujeme databázu strednej veľkosti, oneskoruje výkon.
- Úľ nepodporuje operáciu aktualizácie alebo odstránenia.
- Je obmedzená na podporu poddotazov.
Tieto nevýhody sú dôvodom na vývoj Apache SQL.
Stručný úvod PySpark SQL
PySpark podporuje integrované relačné spracovanie s funkčným programovaním Spark. Poskytuje podporu pre rôzne zdroje údajov, aby bolo možné spájať dotazy SQL s transformáciami kódu, výsledkom čoho je veľmi výkonný nástroj.
PySpark SQL vytvára spojenie medzi RDD a relačnou tabuľkou. Poskytuje oveľa užšiu integráciu medzi relačným a procedurálnym spracovaním prostredníctvom deklaratívneho API Dataframe, ktoré je integrované s kódom Spark.
Pomocou SQL môže byť ľahko dostupný pre viac používateľov a zlepšiť optimalizáciu pre tých súčasných. Podporuje tiež širokú škálu zdrojov údajov a algoritmov v Big-data.
Funkcia PySpark SQL
Funkcie PySpark SQL sú uvedené nižšie:
1) Konzistentný prístup k údajom
Poskytuje konzistentný prístup k údajom, čo znamená, že SQL podporuje zdieľaný spôsob prístupu k rôznym zdrojom údajov, napr Úľ, Avro, Parkety, JSON a JDBC. Hrá významnú úlohu pri prispôsobení všetkých existujúcich používateľov do Spark SQL.
2) Začlenenie so Sparkom
Dotazy PySpark SQL sú integrované s programami Spark. Môžeme použiť dotazy v programoch Spark.
Jednou z jeho najväčších výhod je, že vývojári nemusia manuálne spravovať zlyhanie stavu alebo udržiavať aplikáciu v synchronizácii s dávkovými úlohami.
3) Štandardné pripojenie
Poskytuje pripojenie cez JDBC alebo ODBC a tieto dva sú priemyselnými štandardmi pre konektivitu pre nástroje business intelligence.
4) Používateľom definované funkcie
PySpark SQL má jazykovú kombinovanú používateľsky definovanú funkciu (UDF). UDF sa používa na definovanie novej funkcie založenej na stĺpcoch, ktorá rozširuje slovnú zásobu DSL Spark SQL na transformáciu DataFrame.
5) Kompatibilita úľa
PySpark SQL spúšťa neupravené Hive dotazy na aktuálne dáta. Umožňuje plnú kompatibilitu s aktuálnymi údajmi Hive.
Modul PySpark SQL
Niektoré dôležité triedy Spark SQL a DataFrames sú nasledujúce:
Zvážte nasledujúci príklad PySpark SQL.
import findspark findspark.init() import pyspark # only run after findspark.init() from pyspark.sql import SparkSession spark = SparkSession.builder.getOrCreate() df = spark.sql('''select 'spark' as hello ''') df.show()
Výkon:
normalizácia v databáze
+-----+ |hello| +-----+ |spark| +-----+
Vysvetlenie kódu:
Vo vyššie uvedenom kóde sme importovali findspark modul a zavolal findspark.init() konštruktér; potom sme importovali modul SparkSession na vytvorenie relácie sparku.
z pyspark.sql importujte SparkSession
Spark session možno použiť na vytvorenie Dataset a DataFrame API. SparkSession možno použiť aj na vytvorenie DataFrame, registráciu DataFrame ako tabuľky, spustenie SQL cez tabuľky, cache tabuľky a čítanie parketového súboru.
staviteľ triedy
Je to staviteľ Spark Session.
getOrCreate()
Používa sa na získanie existujúceho SparkSession, alebo ak neexistuje žiadny, vytvorte nový na základe možností nastavených v nástroji.
Niekoľko ďalších metód
Niekoľko metód PySpark SQL je nasledujúcich:
1. appName(name)
Slúži na nastavenie názvu aplikácie, ktorý sa bude zobrazovať vo webovom UI Spark. Parameter názov akceptuje názov parametra.
2. config(key=None, value = None, conf = None)
Používa sa na nastavenie možnosti konfigurácie. Možnosti nastavené pomocou tejto metódy sa automaticky rozšíria do oboch SparkConf a SparkSession konfigurácia používateľa.
from pyspark.conf import SparkConfSparkSession.builder.config(conf=SparkConf())
Parametre:
3. majster (majster)
Nastavuje hlavnú adresu URL sparku, ku ktorej sa má pripojiť, napríklad „local“ na lokálne spustenie, „local[4]“ na lokálne spustenie so 4 jadrami.
Parametre:
4. SparkSession.katalóg
Je to rozhranie, ktoré môže používateľ vytvárať, rušiť, meniť alebo dotazovať na základnú databázu, tabuľky, funkcie atď.
5. SparkSession.conf
Je to runtime konfiguračné rozhranie pre spark. Toto je rozhranie, cez ktoré môže používateľ získať a nastaviť všetky konfigurácie Spark a Hadoop, ktoré sú relevantné pre Spark SQL.
triedy pyspark.sql.DataFrame
Ide o distribuovanú kolekciu údajov zoskupených do pomenovaných stĺpcov. DataFrame je podobný ako relačná tabuľka v Spark SQL, dá sa vytvoriť pomocou rôznych funkcií v SQLContext.
student = sqlContext.read.csv('...')
Po vytvorení dátového rámca s ním môžeme manipulovať pomocou niekoľkých doménovo špecifických jazykov (DSL), ktoré sú preddefinovanými funkciami DataFrame. Zvážte nasledujúci príklad.
# To create DataFrame using SQLContext student = sqlContext.read.parquet('...') department = sqlContext.read.parquet('...') student.filter(marks > 55).join(department, student.student_Id == department.id) .groupBy(student.name, 'gender').({'name': 'student_Id', 'mark': 'department'}) Uvažujme o nasledujúcom príklade:
Dotazovanie pomocou Spark SQL
V nasledujúcom kóde najprv vytvoríme DataFrame a spustíme SQL dotazy na získanie údajov. Zvážte nasledujúci kód:
from pyspark.sql import * #Create DataFrame songdf = spark.read.csv(r'C:UsersDEVANSH SHARMA op50.csv', inferSchema = True, header = True) #Perform SQL queries songdf.select('Genre').show() songdf.filter(songdf['Genre']=='pop').show()
Výkon:
java math.random
+----------------+ | Genre| +----------------+ | canadian pop| | reggaeton flow| | dance pop| | pop| | dfw rap| | pop| | trap music| | pop| | country rap| | electropop| | reggaeton| | dance pop| | pop| | panamanian pop| |canadian hip hop| | dance pop| | latin| | dfw rap| |canadian hip hop| | escape room| +----------------+ only showing top 20 rows +---+--------------------+-------------+-----+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+ |_c0| Track.Name| Artist.Name|Genre|Beats.Per.Minute|Energy|Danceability|Loudness..dB..|Liveness|Valence.|Length.|Acousticness..|Speechiness.|Popularity| +---+--------------------+-------------+-----+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+ | 4|Beautiful People ...| Ed Sheeran| pop| 93| 65| 64| -8| 8| 55| 198| 12| 19| 86| | 6|I Don't Care (wit...| Ed Sheeran| pop| 102| 68| 80| -5| 9| 84| 220| 9| 4| 84| | 8| How Do You Sleep?| Sam Smith| pop| 111| 68| 48| -5| 8| 35| 202| 15| 9| 90| | 13| Someone You Loved|Lewis Capaldi| pop| 110| 41| 50| -6| 11| 45| 182| 75| 3| 88| | 38|Antisocial (with ...| Ed Sheeran| pop| 152| 82| 72| -5| 36| 91| 162| 13| 5| 87| | 44| Talk| Khalid| pop| 136| 40| 90| -9| 6| 35| 198| 5| 13| 84| | 50|Cross Me (feat. C...| Ed Sheeran| pop| 95| 79| 75| -6| 7| 61| 206| 21| 12| 82| +---+--------------------+-------------+-----+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+
Použitie funkcie groupBy().
Funkcia groupBy() zhromažďuje podobné údaje kategórie.
songdf.groupBy('Genre').count().show()
Výkon:
+----------------+-----+ | Genre|count| +----------------+-----+ | boy band| 1| | electropop| 2| | pop| 7| | brostep| 2| | big room| 1| | pop house| 1| | australian pop| 1| | edm| 3| | r&b en espanol| 1| | dance pop| 8| | reggaeton| 2| | canadian pop| 2| | trap music| 1| | escape room| 1| | reggaeton flow| 2| | panamanian pop| 2| | atl hip hop| 1| | country rap| 2| |canadian hip hop| 3| | dfw rap| 2| +----------------+-----+
rozdelenie (numpartitions, *cols)
The distribúcia() vráti nový DataFrame, ktorý je výrazom na rozdelenie. Táto funkcia akceptuje dva parametre numpartitions a *col. The numpartitions parameter určuje cieľový počet stĺpcov.
song_spotify.repartition(10).rdd.getNumPartitions() data = song_spotify.union(song_spotify).repartition('Energy') data.show(5)
Výkon:
+---+--------------------+-----------+-------+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+ |_c0| Track.Name|Artist.Name| Genre|Beats.Per.Minute|Energy|Danceability|Loudness..dB..|Liveness|Valence.|Length.|Acousticness..|Speechiness.|Popularity| +---+--------------------+-----------+-------+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+ | 4|Beautiful People ...| Ed Sheeran| pop| 93| 65| 64| -8| 8| 55| 198| 12| 19| 86| | 5|Goodbyes (Feat. Y...|Post Malone|dfw rap| 150| 65| 58| -4| 11| 18| 175| 45| 7| 94| | 17| LA CANCI?N| J Balvin| latin| 176| 65| 75| -6| 11| 43| 243| 15| 32| 90| | 4|Beautiful People ...| Ed Sheeran| pop| 93| 65| 64| -8| 8| 55| 198| 12| 19| 86| | 5|Goodbyes (Feat. Y...|Post Malone|dfw rap| 150| 65| 58| -4| 11| 18| 175| 45| 7| 94| +---+--------------------+-----------+-------+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+ only showing top 5 rows