يعد Apache Spark البرنامج الأكثر نجاحًا لمؤسسة Apache Software Foundation وهو مصمم للحوسبة السريعة. تستخدم العديد من الصناعات Apache Spark للعثور على حلولها. PySpark SQL عبارة عن وحدة نمطية في Spark تدمج المعالجة العلائقية مع واجهة برمجة تطبيقات البرمجة الوظيفية الخاصة بـ Spark. يمكننا استخراج البيانات باستخدام لغة استعلام SQL. يمكننا استخدام الاستعلامات نفس لغة SQL.

إذا كان لديك فهم أساسي لنظام RDBMS، فسيكون PySpark SQL سهل الاستخدام، حيث يمكنك توسيع حدود معالجة البيانات العلائقية التقليدية. يدعم Spark أيضًا لغة Hive Query Language، ولكن هناك قيود على قاعدة بيانات Hive. تم تطوير Spark SQL لإزالة عيوب قاعدة بيانات Hive. دعونا نلقي نظرة على العيوب التالية للخلية:

عيوب الخلية

• لا يمكن استئناف المعالجة، مما يعني أنه إذا فشل التنفيذ في منتصف سير العمل، فلا يمكنك الاستئناف من حيث توقف.
• لا يمكننا إسقاط قواعد البيانات المشفرة بشكل متتالي عند تمكين سلة المهملات. يؤدي إلى خطأ في التنفيذ. لإسقاط هذا النوع من قاعدة البيانات، يتعين على المستخدمين استخدام خيار التطهير.
• يتم تنفيذ الاستعلامات المخصصة باستخدام MapReduce، الذي يتم تشغيله بواسطة Hive، ولكن عندما نقوم بتحليل قاعدة البيانات متوسطة الحجم، فإنه يؤخر الأداء.
• لا تدعم الخلية عملية التحديث أو الحذف.
• ويقتصر على دعم الاستعلام الفرعي.

هذه العيوب هي أسباب تطوير Apache SQL.

مقدمة موجزة عن PySpark SQL

يدعم PySpark المعالجة العلائقية المتكاملة مع البرمجة الوظيفية لـ Spark. فهو يوفر الدعم لمصادر البيانات المختلفة ليجعل من الممكن نسج استعلامات SQL مع تحويلات التعليمات البرمجية، مما يؤدي إلى أداة قوية جدًا.

ينشئ PySpark SQL الاتصال بين RDD والجدول العلائقي. إنه يوفر تكاملًا أوثق بين المعالجة العلائقية والإجرائية من خلال Dataframe API التعريفي، والذي يتكامل مع رمز Spark.

باستخدام SQL، يمكن الوصول إليها بسهولة لعدد أكبر من المستخدمين وتحسين الأداء للمستخدمين الحاليين. كما أنه يدعم مجموعة واسعة من مصادر البيانات والخوارزميات في البيانات الضخمة.

ميزة PySpark SQL

ميزات PySpark SQL مذكورة أدناه:

1) الوصول إلى البيانات الاتساق

إنه يوفر وصولاً متسقًا إلى البيانات مما يعني أن SQL يدعم طريقة مشتركة للوصول إلى مجموعة متنوعة من مصادر البيانات مثل Hive، وAvro، وParquet، وJSON، وJDBC. إنه يلعب دورًا مهمًا في استيعاب جميع المستخدمين الحاليين في Spark SQL.

2) الاندماج مع سبارك

تم دمج استعلامات PySpark SQL مع برامج Spark. يمكننا استخدام الاستعلامات داخل برامج سبارك.

إحدى أهم مزاياه هي أن المطورين لا يضطرون إلى إدارة فشل الحالة يدويًا أو إبقاء التطبيق متزامنًا مع المهام المجمعة.

3) الاتصال القياسي

فهو يوفر اتصالاً من خلال JDBC أو ODBC، وهما معياران صناعيان للاتصال بأدوات ذكاء الأعمال.

4) الوظائف المحددة من قبل المستخدم

يحتوي PySpark SQL على وظيفة محددة من قبل المستخدم (UDFs) مدمجة في اللغة. يتم استخدام UDF لتحديد وظيفة جديدة قائمة على الأعمدة تعمل على توسيع مفردات DSL الخاصة بـ Spark SQL لتحويل DataFrame.

5) توافق الخلية

يقوم PySpark SQL بتشغيل استعلامات Hive غير معدلة على البيانات الحالية. فهو يسمح بالتوافق الكامل مع بيانات الخلية الحالية.

وحدة PySpark SQL

فيما يلي بعض الفئات المهمة من Spark SQL وDataFrames:

pyspark.sql.SparkSession:وهو يمثل نقطة الدخول الرئيسية ل إطار البيانات ووظيفة SQL.pyspark.sql.DataFrame:وهو يمثل مجموعة موزعة من البيانات المجمعة في أعمدة مسماة.pyspark.sql.العمود:وهو يمثل تعبير عمود في إطار البيانات. pyspark.sql.الصف:يمثل صفًا من البيانات في ملف إطار البيانات. pyspark.sql.GroupedData:طرق التجميع، التي تم إرجاعها بواسطة DataFrame.groupBy(). وظائف pyspark.sql.DataFrameNa:يمثل طرقًا لمعالجة البيانات المفقودة (القيم الخالية).وظائف pyspark.sql.DataFrameStat:وهو يمثل طرقًا لوظيفة الإحصائيات.وظائف pysark.sql:إنه يمثل قائمة بالوظائف المضمنة المتاحة لـ إطار البيانات. أنواع pyspark.sql:وهو يمثل قائمة بأنواع البيانات المتاحة.pyspark.sql.نافذة:يتم استخدامه للعمل مع وظائف النافذة.

خذ بعين الاعتبار المثال التالي لـ 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() 

انتاج:

 +-----+ |hello| +-----+ |spark| +-----+ 

شرح الكود:

في الكود أعلاه، قمنا باستيراد ملف findpark وحدة ودعا يجدبارك.init() البناء؛ بعد ذلك، قمنا باستيراد وحدة SparkSession لإنشاء جلسة شرارة.

من pyspark.sql قم باستيراد SparkSession

يمكن استخدام جلسة شرارة لإنشاء Dataset وDataFrame API. يمكن أيضًا استخدام SparkSession لإنشاء DataFrame، وتسجيل DataFrame كجدول، وتنفيذ SQL على الجداول، وجدول التخزين المؤقت، وقراءة ملف الباركيه.

منشئ الصف

الماسح الضوئي التالي

إنه منشئ جلسة Spark.

getOrCreate()

يتم استخدامه للحصول على موجود جلسة سبارك, أو إذا لم يكن هناك واحد موجود، فقم بإنشاء واحد جديد بناءً على الخيارات المحددة في المنشئ.

بعض الطرق الأخرى

فيما يلي بعض طرق PySpark SQL:

1. اسم التطبيق (الاسم)

يتم استخدامه لتعيين اسم التطبيق الذي سيتم عرضه في Spark web UI. المعلمة اسم يقبل اسم المعلمة.

2. التكوين (مفتاح = لا شيء، القيمة = لا شيء، أسيوط = لا شيء)

يتم استخدامه لتعيين خيار التكوين. يتم نشر الخيارات المعينة باستخدام هذه الطريقة تلقائيًا إلى كليهما سبارك كونف و SparkSession التكوين.

 from pyspark.conf import SparkConfSparkSession.builder.config(conf=SparkConf()) 

حدود:

مفتاح-سلسلة اسم مفتاح لخاصية التكوين.قيمة-وهو يمثل قيمة خاصية التكوين.مؤتمر -مثال على SparkConf.

3. سيد (سيد)

يقوم بتعيين عنوان URL الرئيسي للاتصال به، مثل 'local' للتشغيل محليًا، و'local[4]' للتشغيل محليًا باستخدام 4 مراكز.

حدود:

يتقن:عنوان url لـ spark master.

4.SparkSession.catalog

إنها واجهة يمكن للمستخدم إنشاء أو إسقاط أو تغيير أو الاستعلام عن قاعدة البيانات والجداول والوظائف الأساسية وما إلى ذلك.

5.SparkSession.conf

إنها واجهة تكوين وقت التشغيل لـ Spark. هذه هي الواجهة التي يمكن للمستخدم من خلالها الحصول على جميع تكوينات Spark وHadoop ذات الصلة بـ Spark SQL وتعيينها.

فئة pyspark.sql.DataFrame

إنها مجموعة موزعة من البيانات المجمعة في أعمدة مسماة. يشبه DataFrame الجدول العلائقي في Spark SQL، ويمكن إنشاؤه باستخدام وظائف مختلفة في SQLContext.

 student = sqlContext.read.csv('...') 

بعد إنشاء إطار البيانات، يمكننا معالجته باستخدام العديد من اللغات الخاصة بالمجال (DSL) والتي تعد وظائف محددة مسبقًا في DataFrame. النظر في المثال التالي.

 # 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'}) 

لنتأمل المثال التالي:

الاستعلام باستخدام Spark SQL

في التعليمة البرمجية التالية، نقوم أولاً بإنشاء DataFrame وتنفيذ استعلامات SQL لاسترداد البيانات. خذ بعين الاعتبار الكود التالي:

 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() 

انتاج:

 +----------------+ | 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| +---+--------------------+-------------+-----+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+ 

باستخدام الدالة groupBy()

تقوم الدالة groupBy() بجمع بيانات الفئة المشابهة.

 songdf.groupBy('Genre').count().show() 

انتاج:

 +----------------+-----+ | 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| +----------------+-----+ 

التوزيع (الأعداد، *الأعمدة)

ال توزيع() تقوم بإرجاع DataFrame جديد وهو تعبير تقسيم. تقبل هذه الوظيفة معلمتين أقسام الأرقام و * العقيد. ال أقسام الأرقام تحدد المعلمة العدد المستهدف من الأعمدة.

 song_spotify.repartition(10).rdd.getNumPartitions() data = song_spotify.union(song_spotify).repartition('Energy') data.show(5) 

انتاج:

 +---+--------------------+-----------+-------+----------------+------+------------+--------------+--------+--------+-------+--------------+------------+----------+ |_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