تطوير نموذج في PySpark ML على مجموعة بيانات بأنواع بيانات مختلفة للدمى الصدئة

هل تعرف بالفعل كيفية التعامل مع أنواع بيانات متعددة في PySpark ML؟ لا؟ فأنت بحاجة ماسة لزيارتنا.







مرحبا! أريد أن أغطي بالتفصيل موضوعًا مثيرًا للاهتمام ، لكن للأسف ، ليس موضوعًا في وثائق Spark: كيف تدرب نموذجًا في PySpark ML على مجموعة بيانات مع أنواع بيانات مختلفة (سلاسل وأرقام)؟ كانت الرغبة في كتابة هذه المقالة ناتجة عن الحاجة إلى تصفح الإنترنت لعدة أيام بحثًا عن المقالة الضرورية مع الكود ، لأن البرنامج التعليمي الرسمي من Spark يقدم مثالًا للعمل ليس فقط مع علامات نوع بيانات واحد ، ولكن بشكل عام بعلامة واحدة ، ولكن معلومات عن كيفية العمل مع عدة أعمدة وأنواع مختلفة من البيانات ، لا يوجد. ومع ذلك ، بعد أن درست بالتفصيل إمكانيات PySpark للعمل مع البيانات ، تمكنت من كتابة كود العمل وفهم كيف يحدث كل شيء ، والذي أريد مشاركته معك. بأقصى سرعة للأمام ، أيها الأصدقاء!



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

#  
from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
from pyspark.ml import Pipeline
from pyspark.ml.feature import HashingTF, Tokenizer
from pyspark.sql.functions import UserDefinedFunction
from pyspark.sql.types import *
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
import pyspark.sql.functions as sf
from pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler
from pyspark.ml import Pipeline
from pyspark.ml.regression import GBTRegressor
#other types of regression models
#     
#from pyspark.ml.regression import LinearRegression
#from pyspark.ml.regression import RandomForestRegressor
#from pyspark.ml.regression import GeneralizedLinearRegression
#from pyspark.ml.regression import DecisionTreeRegressor
from pyspark.ml.feature import VectorIndexer
from pyspark.ml.evaluation import RegressionEvaluator

الآن دعنا ننشئ سياق Spark (محلي) وجلسة Spark ونتحقق مما إذا كان كل شيء يعمل عن طريق عرضه على الشاشة. يعد إنشاء جلسة Spark نقطة البداية للعمل مع مجموعات البيانات في Spark:

#  
sc = SparkContext('local')
spark = SparkSession(sc)
spark

هناك أداة للعمل مع البيانات ، فلنقم الآن بتحميلها. تستخدم المقالة مجموعة بيانات مأخوذة من موقع مسابقة التعلم الآلي Kaggle:

https://www.kaggle.com/unitednations/international-greenhouse-gas-emissions

والتي يتم تخزينها بعد التنزيل في path_csv بتنسيق .csv وبها الخيارات التالية:

• header: إذا كان السطر الأول في ملفنا هو header ، فإننا نضع "true"
• محدد: نضع إشارة تفصل بيانات سطر واحد بعلامات ، غالبًا ما تكون "،" أو "؛"
• inferSchema: إذا كان "صحيحًا" ، فسيكتشف PySpark تلقائيًا نوع كل عمود ، وإلا فسيتعين عليك كتابته بنفسك

#   .csv  path_csv
path_csv = 'greenhouse_gas_inventory_data_data.csv'
data = spark.read.format("csv")\
        .option("header", "true")\
        .option("delimiter", ",")\
        .option("inferSchema", "true")\
        .load(path_csv)

لفهم نوع البيانات التي نتعامل معها بشكل أفضل ، دعنا نلقي نظرة على بعض أسطرها:

#   
data.show()

لنرى أيضًا عدد الصفوف الموجودة في مجموعة البيانات:

#  
data.select('year').count()

وأخيرًا ، دعنا نستنتج أنواع بياناتنا ، والتي ، كما نتذكر ، طلبنا من PySpark تحديدها تلقائيًا باستخدام الخيار ("inferSchema" ، "true"):

#     
data.printSchema()

الآن دعنا ننتقل إلى الطبق الرئيسي - العمل مع عدة علامات لأنواع بيانات مختلفة. يمكن لـ Spark تدريب النموذج على البيانات المحولة ، حيث يكون العمود المتوقع متجهًا والأعمدة ذات الميزات أيضًا متجه ، مما يعقد المهمة ... لكننا لا نستسلم ، ولتدريب النموذج في PySpark ، سنستخدم خط الأنابيب ، حيث سنمرر خطة عمل معينة (متغير) مراحل):

1. step label_stringIdx: نقوم بتحويل عمود مجموعة بيانات القيمة التي نريد أن نتنبأ بها إلى سلسلة متجهية Spark وإعادة تسميتها للتسمية بالمعامل handleInvalid = 'keep' ، مما يعني أن عمودنا المتوقع يدعم قيمة خالية
2. step stringIndexer: تحويل أعمدة السلسلة إلى سلاسل Spark قاطعة
3. encoder: ()
4. assembler: Spark, , VectorAssembler(), ( ) (assemblerInputs) «features»
5. gbt: PySpark ML GBTRegressor,

#value -      - 
stages = []
label_stringIdx = StringIndexer(inputCol = 'value', outputCol = 'label', handleInvalid = 'keep')
stages += [label_stringIdx]

#depend on categorical columns: country and types of emission
#   :    
categoricalColumns = ['country_or_area', 'category']
for categoricalCol in categoricalColumns:
    #        
    stringIndexer = StringIndexer(inputCol = categoricalCol,
                                  outputCol = categoricalCol + 'Index',
                                  handleInvalid = 'keep')
    encoder = OneHotEncoder(inputCol=stringIndexer.getOutputCol(),
                            outputCol=categoricalCol + "classVec")
    stages += [stringIndexer, encoder]

#   : 
numericCols = ['year']
assemblerInputs = [c + "classVec" for c in categoricalColumns] + numericCols
#    - - 
assembler = VectorAssembler(inputCols=assemblerInputs, outputCol="features")
stages += [assembler]

دعنا نقسم مجموعة البيانات الخاصة بنا إلى عينات تدريب واختبار في النسبة المفضلة من 70٪ إلى 30٪ ، على التوالي ، ونبدأ تدريب النموذج باستخدام شجرة تعزيز الانحدار المتدرج (GBTRegressor) ، والتي يجب أن تتنبأ بمتجه "التسمية" بناءً على الميزات التي تم دمجها مسبقًا في ناقل "ميزات" واحد بحد أقصى متكرر = 10:

#       (30% )
(trainingData, testData) = data.randomSplit([0.7, 0.3])

#  (   )
gbt = GBTRegressor(labelCol="label", featuresCol="features", maxIter=10)
stages += [gbt]

#   stages    
pipeline = Pipeline(stages=stages)

والآن نحتاج فقط إلى إرسال خطة عمل ومجموعة بيانات تدريبية إلى الكمبيوتر:

#  
model = pipeline.fit(trainingData)

#     
predictions = model.transform(testData)

دعنا نحفظ نموذجنا حتى نتمكن دائمًا من العودة إلى استخدامه دون إعادة التدريب:

# 
pipeline.write().overwrite().save('model/gbtregr_model')

وإذا قررت البدء في استخدام النموذج المدرب للتنبؤات مرة أخرى ، فاكتب ببساطة:

#     
load_model = pipeline.read().load('model/gbtregr_model')

لذلك ، رأينا كيف يتم تطبيق PySpark في أداة للعمل مع البيانات الضخمة في لغة Python ، مع العديد من أعمدة الميزات لأنواع البيانات المختلفة.



حان الوقت الآن لتطبيق هذا على نماذجك ...