استخدام Spark للعمل مع ملفات البيانات

8 دقائق

بعد إعداد دفتر ملاحظات وإرفاقه بنظام مجموعة، يمكنك استخدام Spark لقراءة ملفات البيانات ومعالجتها. يدعم Spark مجموعة واسعة من التنسيقات - مثل CSV وJSON وParquet وORC وAvro وDelta - ويوفر Databricks موصلات مضمنة للوصول إلى الملفات المخزنة في مساحة العمل أو في Azure Data Lake أو مخزن البيانات الثنائية الكبيرة أو في أنظمة خارجية أخرى.

عادة ما يتبع سير العمل ثلاث خطوات:

اقرأ ملفا في Spark DataFrame باستخدام spark.read بالتنسيق والمسار الصحيحين. عند قراءة تنسيقات النص الأولي مثل CSV أو JSON ، يمكن ل Spark استنتاج المخطط (أسماء الأعمدة وأنواع البيانات) ، ولكن هذا يكون بطيئا أو غير موثوق به في بعض الأحيان. تتمثل أفضل ممارسة في الإنتاج في تحديد المخطط بشكل صريح بحيث يتم تحميل البيانات باستمرار وكفاءة.
استكشاف DataFrame وتحويله باستخدام عمليات SQL أو DataFrame (على سبيل المثال، تصفية الصفوف، وتحديد الأعمدة، وتجميع القيم).
اكتب النتائج مرة أخرى إلى التخزين بتنسيق محدد.

تم تصميم العمل مع الملفات في Spark ليكون متسقا عبر مجموعات البيانات الصغيرة والكبيرة. ستعمل نفس التعليمات البرمجية المستخدمة لاختبار ملف CSV صغير أيضا على مجموعات بيانات أكبر بكثير ، نظرا لأن Spark يوزع العمل عبر نظام المجموعة. هذا يجعل من السهل التوسع من الاستكشاف السريع إلى معالجة البيانات الأكثر تعقيدا.

تحميل البيانات في إطار البيانات

دعونا نستكشف مثالاً افتراضياً لمعرفة كيف يمكنك استخدام إطار بيانات للعمل مع البيانات. افترض أن لديك البيانات التالية في ملف نصي محدد بفاصل يدعى products.csv مخزنة في مجلد كتالوج Unity:

ProductID,ProductName,Category,ListPrice
771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

في دفتر ملاحظات Spark، يمكنك استخدام التعليمات البرمجية PySpark التالية لتحميل البيانات في إطار بيانات وعرض أول 10 صفوف:

%pyspark
df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/products.csv',
    format='csv',
    header=True
)
display(df.limit(10))

%pyspark يسمى السطر في البداية سحرا، ويخبر Spark أن اللغة المستخدمة في هذه الخلية هي PySpark. فيما يلي التعليمة البرمجية المكافئة لـ Scala لمثال بيانات المنتجات:

%spark
val df = spark.read.format("csv").option("header", "true").load("/Volumes/my_catalog/my_schema/my_volume/products.csv")
display(df.limit(10))

إشعار

يتم الوصول إلى الملفات في Azure Databricks عبر مجلدات كتالوج Unity باستخدام مسارات بصيغة /Volumes/<catalog>/<schema>/<volume>/<path>. هذا هو النمط الموصى به لتخزين الملفات والوصول إليها. DBFS (/data/...) هو نهج قديم ولم يعد يوصى به.

يتم استخدام السحر %spark لتحديد Scala.

تلميح

يمكنك أيضاً تحديد اللغة التي تريد استخدامها لكل خلية في واجهة دفتر الملاحظات.

سيؤدي كلا المثالين الموضحين سابقاً إلى إخراج مثل هذا:

معرّف المنتج	ProductName	الفئة	قائمة الأسعار
771	ماونتن 100 فضي، 38	دراجات جبلية	3399.9900
772	ماونتن 100 فضي، 42	دراجات جبلية	3399.9900
773	ماونتن 100 فضي، 44	دراجات جبلية	3399.9900
...	...	...	...

تحديد مخطط إطار بيانات

في المثال السابق، احتوى الصف الأول من ملف CSV على أسماء الأعمدة، وتمكن Spark من استنتاج نوع بيانات كل عمود من البيانات التي يحتوي عليها. يمكنك أيضاً تحديد مخطط صريح للبيانات، وهو أمر مفيد عندما لا يتم تضمين أسماء الأعمدة في ملف البيانات، مثل مثال CSV هذا:

771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

يوضح مثال PySpark التالي كيفية تحديد مخطط لإطار البيانات المراد تحميله من ملف يسمى product-data.csv بهذا التنسيق:

from pyspark.sql.types import *
from pyspark.sql.functions import *

productSchema = StructType([
    StructField("ProductID", IntegerType()),
    StructField("ProductName", StringType()),
    StructField("Category", StringType()),
    StructField("ListPrice", FloatType())
    ])

df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/product-data.csv',
    format='csv',
    schema=productSchema,
    header=False)
display(df.limit(10))

وستكون النتائج مرة أخرى مشابهة لما يلي:

معرّف المنتج	ProductName	الفئة	قائمة الأسعار
771	ماونتن 100 فضي، 38	دراجات جبلية	3399.9900
772	ماونتن 100 فضي، 42	دراجات جبلية	3399.9900
773	ماونتن 100 فضي، 44	دراجات جبلية	3399.9900
...	...	...	...

تصفية إطارات البيانات وتجميعها

يمكنك استخدام أساليب فئة Dataframe لتصفية البيانات التي تحتوي عليها وفرزها وتجميعها ومعالجتها بطريقة أخرى. على سبيل المثال، يستخدم select مثال التعليمات البرمجية التالي الأسلوب لاسترداد أعمدة ProductNameوListPrice من إطار بيانات df الذي يحتوي على بيانات المنتج في المثال السابق:

pricelist_df = df.select("ProductID", "ListPrice")

ستبدو النتائج من مثال التعليمات البرمجية هذا كما يلي:

معرّف المنتج	قائمة الأسعار
771	3399.9900
772	3399.9900
773	3399.9900
...	...

كما هو الحال مع معظم طرق معالجة البيانات، select ترجع كائن إطار بيانات جديد.

تلميح

يعد تحديد مجموعة فرعية من الأعمدة من إطار البيانات عملية شائعة، والتي يمكن تحقيقها أيضا باستخدام بناء الجملة الأقصر التالي:

pricelist_df = df["ProductID", "ListPrice"]

يمكنك "ربط" الأساليب معا لتنفيذ سلسلة من المعالجات التي تؤدي إلى تحويل إطار البيانات. على سبيل المثال، يقوم select هذا المثال من التعليمات البرمجية بتسلسل وأساليب where إنشاء إطار بيانات جديد يحتوي على عمودي ProductNameوListPrice للمنتجات التي تحتوي على فئة من الدراجات الجبلية أو دراجات الطرق:

bikes_df = df.select("ProductName", "ListPrice").where((df["Category"]=="Mountain Bikes") | (df["Category"]=="Road Bikes"))
display(bikes_df)

ستبدو النتائج من مثال التعليمات البرمجية هذا كما يلي:

ProductName	قائمة الأسعار
ماونتن 100 فضي، 38	3399.9900
رود-750 أسود، 52	539.9900
...	...

لتجميع البيانات وتجميعها، يمكنك استخدام groupby الأسلوب ووظائف التجميع. على سبيل المثال، تحسب التعليمات البرمجية PySpark التالية عدد المنتجات لكل فئة:

counts_df = df.select("ProductID", "Category").groupBy("Category").count()
display(counts_df)

ستبدو النتائج من مثال التعليمات البرمجية هذا كما يلي:

الفئة	عدد
سماعة الرأس	3
العجلات	14
دراجات جبلية	32
...	...

إشعار

إطارات بيانات Spark تقريرية وغير قابلة للتغيير. يقوم كل تحويل (مثل select، filterأو أو groupBy) بإنشاء إطار بيانات جديد يمثل ما تريده، وليس كيفية تشغيله. هذا يجعل الكود قابلا لإعادة الاستخدام وقابلا للتحسين وخاليا من الآثار الجانبية. ولكن لا يتم تنفيذ أي من هذه التحولات فعليا حتى تقوم بتشغيل إجراء (على سبيل المثال ، display، collect، write) ، وعند هذه النقطة يقوم Spark بتشغيل الخطة المحسنة الكاملة.

استخدام تعبيرات SQL في Spark

تعد واجهة برمجة تطبيقات Dataframe جزءاً من مكتبة Spark المسماة Spark SQL، والتي تمكن محللي البيانات من استخدام تعبيرات SQL للاستعلام عن البيانات ومعالجتها.

إنشاء كائنات قاعدة البيانات في كتالوج Spark

كتالوج Spark هو مخزن بيانات تعريفي لكائنات البيانات الارتباطية مثل طرق العرض والجداول. يمكن لوقت تشغيل Spark استخدام الكتالوج لدمج التعليمات البرمجية المكتوبة بأي لغة مدعومة من Spark بسلاسة مع تعبيرات SQL التي قد تكون أكثر طبيعية لبعض محللي البيانات أو المطورين.

تتمثل إحدى أبسط الطرق لتوفير البيانات في إطار بيانات للاستعلام في كتالوج Spark في إنشاء طريقة عرض مؤقتة، كما هو موضح في مثال التعليمات البرمجية التالي:

df.createOrReplaceTempView("products")

طريقة العرض مؤقتة، ما يعني أنه يتم حذفها تلقائيا في نهاية جلسة العمل الحالية. يمكنك أيضا إنشاء جداول ثابتة في الكتالوج لتعريف قاعدة بيانات يمكن الاستعلام فيها باستخدام Spark SQL.

إشعار

لن نستكشف جداول كتالوج Spark بعمق في هذه الوحدة، ولكن من المفيد أخذ الوقت لتمييز بعض النقاط الرئيسية:

يمكنك إنشاء جدول فارغ باستخدام spark.catalog.createTable الأسلوب. الجداول هي بنيات بيانات التعريف التي تخزن بياناتها الأساسية في موقع التخزين المقترن بالكتالوج. يؤدي حذف جدول أيضاً إلى حذف بياناته الأساسية.
يمكنك حفظ إطار بيانات كجدول باستخدام saveAsTable أسلوبه.
يمكنك إنشاء جدول خارجي باستخدام spark.catalog.createExternalTable الأسلوب . تحدد الجداول الخارجية بيانات التعريف في الكتالوج ولكن تحصل على بياناتها الأساسية من موقع تخزين خارجي؛ عادة ما يكون مجلداً في مستودع بيانات. لا يؤدي حذف جدول خارجي إلى حذف البيانات الأساسية.

استخدام Spark SQL API للاستعلام عن البيانات

يمكنك استخدام Spark SQL API في التعليمات البرمجية المكتوبة بأي لغة للاستعلام عن البيانات في الكتالوج. على سبيل المثال، تستخدم التعليمات البرمجية PySpark التالية استعلام SQL لإرجاع البيانات من طريقة عرض المنتجات كإطار بيانات.

bikes_df = spark.sql("SELECT ProductID, ProductName, ListPrice \
                      FROM products \
                      WHERE Category IN ('Mountain Bikes', 'Road Bikes')")
display(bikes_df)

ستبدو النتائج من مثال التعليمات البرمجية مشابهة للجدول التالي:

ProductName	قائمة الأسعار
ماونتن 100 فضي، 38	3399.9900
رود-750 أسود، 52	539.9900
...	...

استخدام التعليمات البرمجية SQL

يوضح المثال السابق كيفية استخدام Spark SQL API لتضمين تعبيرات SQL في تعليمة Spark البرمجية. في دفتر الملاحظات، يمكنك أيضاً استخدام %sql السحر لتشغيل التعليمات البرمجية SQL التي تقوم بالاستعلامات عن الكائنات في الكتالوج، مثل هذا:

%sql

SELECT Category, COUNT(ProductID) AS ProductCount
FROM products
GROUP BY Category
ORDER BY Category

يرجع مثال التعليمات البرمجية SQL مجموعة نتائج يتم عرضها تلقائيا في دفتر الملاحظات كجدول، مثل الجدول أدناه:

الفئة	عدد المنتجات
Bib-Shorts	3
رفوف الدراجات	1
مدرجات الدراجة	1
...	...

الملاحظات

هل كانت هذه الصفحة مفيدة؟