Apache Spark ट्यूटोरियल: PySpark के साथ ML

डाउन्लोडिंग pyspark विथ pip

फिर, आप pip की मदद से PySpark डाउनलोड और इंस्टॉल कर सकते हैं। यह काफ़ी आसान है और किसी अन्य पैकेज को इंस्टॉल करने जैसा ही है। आपको बस सामान्य कमांड चलानी है और भारी काम आपके लिए हो जाएगा:

$ pip install pysparkसंचालित द्वारा

वैकल्पिक रूप से, आप Spark डाउनलोड पेज पर भी जा सकते हैं। पहले तीन चरणों में डिफॉल्ट विकल्प रखें और चरण 4 में आपको एक डाउनलोड करने योग्य लिंक मिलेगा। उसे क्लिक करके डाउनलोड करें। इस ट्यूटोरियल के लिए, आप 2.2.0 Spark रिलीज़ और “Pre-built for Apache Hadoop 2.7 and later” पैकेज प्रकार डाउनलोड करेंगे।

नोट कि डाउनलोड पूरा होने में कुछ समय लग सकता है!

Homebrew के साथ Spark डाउनलोड करना

आप Homebrew, जो एक मुक्त और ओपन-सोर्स पैकेज मैनेजर है, के साथ भी Spark इंस्टॉल कर सकते हैं। यदि आप macOS के साथ काम कर रहे हैं तो यह विशेष रूप से उपयोगी है।

Spark को खोजने, अधिक जानकारी प्राप्त करने और अंततः अपने पर्सनल कंप्यूटर पर इंस्टॉल करने के लिए बस निम्न कमांड चलाएँ:

# Search for spark
$ brew search spark

# Get more information on apache-spark
$ brew info apache-spark

# Install apache-spark
$ brew install apache-sparkसंचालित द्वारा

Spark डाउनलोड करें और सेट अप करें

अगला, सुनिश्चित करें कि आप अपने Downloads फ़ोल्डर में दिखाई देने वाली डायरेक्टरी को अनटार कर लें। यह आपके लिए स्वतः भी हो सकता है, spark-2.2.0-bin-hadoop2.7.tgz आर्काइव पर डबल-क्लिक करके, या अपने टर्मिनल को खोलकर निम्न कमांड चलाकर:

$ tar xvf spark-2.2.0-bin-hadoop2.7.tgzसंचालित द्वारा

फिर, निम्न लाइन चलाकर अनटार किए गए फ़ोल्डर को /usr/local/spark में मूव करें:

$ mv spark-2.1.0-bin-hadoop2.7 /usr/local/sparkसंचालित द्वारा

नोट कि यदि आपको यह त्रुटि मिलती है कि इस फ़ोल्डर को नए लोकेशन पर मूव करने की अनुमति नहीं है, तो आपको इस कमांड के आगे sudo जोड़ना चाहिए। ऊपर की लाइन तब $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark बन जाएगी। आपसे पासवर्ड मांगा जाएगा, जो आमतौर पर वही होता है जो आप अपने पीसी को स्टार्ट करते समय अनलॉक करने के लिए उपयोग करते हैं :)

अब जब सब सेट है, तो /usr/local/spark पाथ में README फ़ाइल खोलें। आप इसे रन करके कर सकते हैं

$ cd /usr/local/sparkसंचालित द्वारा

यह आपको उस फ़ोल्डर में ले आएगा जहाँ आपको होना चाहिए। फिर, आप फ़ोल्डर को देखना शुरू कर सकते हैं और उसमें शामिल README फ़ाइल पढ़ सकते हैं।

सबसे पहले, $ ls का उपयोग करके उन फाइलों और फ़ोल्डरों की सूची प्राप्त करें जो इस spark फ़ोल्डर में हैं। आप देखेंगे कि वहाँ एक README.md फ़ाइल है। आप इसे निम्न कमांड में से किसी एक को चलाकर खोल सकते हैं:

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.mdसंचालित द्वारा

टिप फाइल का नाम टाइप करते समय ऑटोकम्प्लीट के लिए अपने कीबोर्ड पर टैब बटन का उपयोग करें :) इससे आपका समय बचेगा।

आप देखेंगे कि यह README आपको Spark के बारे में कुछ सामान्य जानकारी, ऑनलाइन डॉक्यूमेंटेशन, Spark बिल्ड करना, इंटरएक्टिव Scala और Python शेल्स, उदाहरण प्रोग्राम्स और बहुत कुछ प्रदान करता है।

यहाँ जो चीज़ आपको सबसे अधिक रुचिकर लग सकती है, वह है Spark को कैसे बिल्ड करें वाला सेक्शन, लेकिन ध्यान दें कि यह खासकर तब ही प्रासंगिक होगा जब आपने प्री-बिल्ट वर्ज़न डाउनलोड नहीं किया हो। इस ट्यूटोरियल के लिए, हालांकि, आपने प्री-बिल्ट वर्ज़न डाउनलोड किया है। आप README से बाहर निकलने के लिए CTRL + X दबा सकते हैं, जो आपको वापस spark फ़ोल्डर में ले आएगा।

यदि आपने कोई ऐसा वर्ज़न चुना है जो अभी तक बिल्ड नहीं हुआ है, तो सुनिश्चित करें कि आप README में दी गई कमांड चलाएँ। लेखन के समय, वह निम्न थी:

$ build/mvn -DskipTests clean package runसंचालित द्वारा

ध्यान दें कि इस कमांड को चलने में समय लग सकता है।

Spark एप्लिकेशन बनाम Spark शेल

इंटरएक्टिव शेल, Read-Eval(uate)-Print-Loop (REPL) एनवायरनमेंट का एक उदाहरण है; इसका मतलब है कि आप जो कुछ भी टाइप करते हैं, उसे पढ़ा जाता है, इवैल्यूएट किया जाता है और आपको प्रिंट करके दिखाया जाता है ताकि आप अपना विश्लेषण जारी रख सकें। यह आपको IPython की याद दिला सकता है, जो एक शक्तिशाली इंटरएक्टिव Python शेल है जिसे आप Jupyter के साथ काम करने से जानते होंगे। यदि आप अधिक जानना चाहते हैं, तो DataCamp का IPython या Jupyter ब्लॉग पोस्ट पढ़ें।

इसका मतलब है कि आप इस शेल का उपयोग, जो Python और Scala दोनों के लिए उपलब्ध है, उन सभी इंटरएक्टिव कार्यों के लिए कर सकते हैं जिनकी आपको आवश्यकता है।

इस शेल के अलावा, आप Spark एप्लिकेशन लिख और डिप्लॉय भी कर सकते हैं। Spark एप्लिकेशन लिखने के विपरीत, शेल में SparkSession पहले से ही आपके लिए बनाया गया है ताकि आप सीधे काम शुरू कर सकें और एक बनाने में समय बर्बाद न करें।

अब आप सोच सकते हैं: SparkSession क्या है?

खैर, यह Spark की फंक्शनैलिटी का मुख्य एंट्री पॉइंट है: यह Spark क्लस्टर से कनेक्शन का प्रतिनिधित्व करता है और आप इसका उपयोग उस क्लस्टर पर RDDs बनाने और वेरिएबल्स को ब्रॉडकास्ट करने के लिए कर सकते हैं। जब आप Spark के साथ काम करते हैं, तो सब कुछ इसी SparkSession से शुरू और खत्म होता है। ध्यान दें कि Spark 2.0.0 से पहले, तीन मुख्य कनेक्शन ऑब्जेक्ट्स SparkContext, SqlContext और HiveContext थे।

आप आगे चलकर इसके बारे में और देखेंगे। फिलहाल, शेल पर ही फोकस करते हैं।

Python Spark शेल

/usr/local/spark पर स्थित spark फ़ोल्डर से, आप चला सकते हैं

$ ./bin/pysparkसंचालित द्वारा

शुरुआत में आपको कुछ टेक्स्ट दिखाई देगा। और फिर, आपको “Spark” दिखाई देगा, बिल्कुल ऐसे:

Python 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016, 12:39:47) 
[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
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).
17/07/26 11:41:26 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/07/26 11:41:47 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.2.0
      /_/

Using Python version 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016 12:39:47)
SparkSession available as 'spark'.
>>>संचालित द्वारा

जब आप यह देख लेते हैं, तो आप जानते हैं कि आप इंटरएक्टिव शेल में प्रयोग शुरू करने के लिए तैयार हैं!

टिप: यदि आप Spark शेल की बजाय IPython शेल का उपयोग करना पसंद करते हैं, तो आप निम्न एनवायरनमेंट वेरिएबल सेट करके ऐसा कर सकते हैं:

export PYSPARK_DRIVER_PYTHON="/usr/local/ipython/bin/ipython"संचालित द्वारा

>>> rdd1 = spark.sparkContext.parallelize([('a',7),('a',2),('b',2)])
>>> rdd2 = spark.sparkContext.parallelize([("a",["x","y","z"]), ("b",["p", "r"])])
>>> rdd3 = spark.sparkContext.parallelize(range(100))संचालित द्वारा

>>> rdd1.reduce(lambda a,b: a+b)संचालित द्वारा

>>> rdd2.flatMapValues(lambda x: x).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]संचालित द्वारा

Jupyter Notebook में PySpark

ट्यूटोरियल के इस हिस्से के लिए, आप ishell का उपयोग नहीं करेंगे बल्कि अपनी खुद की एप्लिकेशन बनाएँगे। आप यह एक Jupyter Notebook में करेंगे। आपके पास पहले से ही सभी आवश्यक चीजें इंस्टॉल हैं, इसलिए Jupyter में PySpark को काम में लाने के लिए आपको बहुत कुछ करने की ज़रूरत नहीं है।

आप नोटबुक एप्लिकेशन हमेशा की तरह ही लॉन्च कर सकते हैं, $ jupyter notebook चलाकर। फिर, आप एक नया नोटबुक बनाते हैं और बस findspark लाइब्रेरी इम्पोर्ट करके init() फ़ंक्शन का उपयोग करते हैं। इस मामले में, आप init() को पाथ /usr/local/spark देंगे क्योंकि आपको यक़ीन है कि आपने Spark यहीं इंस्टॉल किया है।

# Import findspark 
import findspark

# Initialize and provide path
findspark.init("/usr/local/spark")

# Or use this alternative
#findspark.init()संचालित द्वारा

टिप: यदि आपको अंदाज़ा नहीं है कि आपका पाथ सही सेट है या आपने अपने पीसी पर Spark कहाँ इंस्टॉल किया है, तो आप हमेशा findspark.find() का उपयोग करके Spark की लोकेशन स्वतः डिटेक्ट कर सकते हैं।

यदि आप Jupyter में Spark के साथ काम करने के वैकल्पिक तरीकों की तलाश में हैं, तो हमारा Apache Spark in Python: Beginner’s Guide देखें।

अब जब यह सब सेट हो गया है, तो आप आखिरकार अपना पहला Spark प्रोग्राम बनाना शुरू कर सकते हैं!

अपना पहला Spark प्रोग्राम बनाना

सबसे पहले, आप pyspark पैकेज से SparkContext इम्पोर्ट करना और उसे इनिशियलाइज़ करना चाहेंगे। याद रखें कि आपको पहले यह करने की ज़रूरत नहीं थी क्योंकि इंटरएक्टिव Spark शेल ने इसे आपके लिए स्वतः बनाया और इनिशियलाइज़ किया था! यहाँ, आपको थोड़ा और काम खुद करना होगा :)

pyspark.sql से SparkSession मॉड्यूल इम्पोर्ट करें और builder() मेथड के साथ एक SparkSession बनाएं। उसके बाद, आप कनेक्ट करने के लिए मास्टर URL, एप्लिकेशन का नाम सेट कर सकते हैं, एक्ज़ीक्यूटर मेमोरी जैसी कुछ अतिरिक्त कॉन्फ़िगरेशन जोड़ सकते हैं और अंत में, getOrCreate() का उपयोग करके या तो वर्तमान Spark सेशन प्राप्त कर सकते हैं या यदि कोई नहीं चल रहा है तो एक बना सकते हैं।

# Import SparkSession
from pyspark.sql import SparkSession

# Build the SparkSession
spark = SparkSession.builder \
   .master("local") \
   .appName("Linear Regression Model") \
   .config("spark.executor.memory", "1gb") \
   .getOrCreate()
   
sc = spark.sparkContextसंचालित द्वारा

नोट कि यदि आपको FileNotFoundError जैसी कोई त्रुटि मिलती है: “No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’”, तो आपको पता है कि आपको अपना Spark PATH (दोबारा) सेट करना होगा। $ cd चलाकर अपने होम डायरेक्टरी में जाएँ और फिर $ nano .bash_profile चलाकर .bash_profile फ़ाइल एडिट करें।

फ़ाइल के अंत में निम्न जैसा कुछ जोड़ें

export SPARK_HOME="/usr/local/spark"संचालित द्वारा

फ़ाइल से बाहर निकलने के लिए CTRL + X का उपयोग करें, लेकिन सुनिश्चित करें कि परिवर्तन सहेजने के लिए Y भी दबाएँ। अगला, source .bash_profile चलाकर परिवर्तनों को प्रभावी करना न भूलें।

टिप: यदि आप चाहें तो आप अतिरिक्त एनवायरनमेंट वेरिएबल्स भी सेट कर सकते हैं; आपको शायद उनकी ज़रूरत नहीं होगी, लेकिन यह जानना अच्छा है कि इच्छा होने पर आप उन्हें सेट कर सकते हैं। निम्न उदाहरण देखें:

# Set a fixed value for the hash seed secret
export PYTHONHASHSEED=0

# Set an alternate Python executable
export PYSPARK_PYTHON=/usr/local/ipython/bin/ipython

# Augment the default search path for shared libraries
export LD_LIBRARY_PATH=/usr/local/ipython/bin/ipython

# Augment the default search path for private libraries 
export PYTHONPATH=$SPARK_HOME/python/lib/py4j-*-src.zip:$PYTHONPATH:$SPARK_HOME/python/संचालित द्वारा

नोट यह भी कि अब आपने एक डिफॉल्ट SparkSession इनिशियलाइज़ कर लिया है। हालाँकि, अधिकांश मामलों में, आप इसे और कॉन्फ़िगर करना चाहेंगे। आप देखेंगे कि वास्तव में इसकी आवश्यकता तब होगी जब आप बिग डेटा के साथ काम करना शुरू करेंगे। यदि आप इसके बारे में अधिक जानना चाहते हैं, तो यह पेज देखें।

अपना डेटा लोड करना

यह ट्यूटोरियल California Housing डेटा सेट का उपयोग करता है। यह 1997 के पेपर Sparse Spatial Autoregressions में आया था, जिसे Pace, R. Kelley और Ronald Barry ने लिखा था और Statistics and Probability Letters जर्नल में प्रकाशित किया गया था। शोधकर्ताओं ने यह डेटा सेट 1990 के कैलिफ़ोर्निया जनगणना डेटा का उपयोग करके बनाया।

डेटा में प्रति जनगणना ब्लॉक समूह एक पंक्ति है। ब्लॉक समूह वह सबसे छोटा भौगोलिक यूनिट है जिसके लिए U.S. Census Bureau सैंपल डेटा प्रकाशित करता है (एक ब्लॉक समूह में आम तौर पर 600 से 3,000 लोगों की आबादी होती है)। इस सैंपल में एक ब्लॉक समूह में औसतन 1425.5 व्यक्ति शामिल हैं जो भौगोलिक रूप से कॉम्पैक्ट क्षेत्र में रहते हैं। आप यह जानकारी इस वेब पेज से या ऊपर उल्लिखित पेपर, जो आपको यहाँ मिल जाएगा, पढ़कर प्राप्त करेंगे।

इन स्थानिक डेटा में 9 आर्थिक वेरिएबल्स के साथ हाउसिंग प्राइस पर 20,640 ऑब्ज़र्वेशन शामिल हैं:

• Longitude प्रत्येक ब्लॉक समूह के लिए पृथ्वी के भूमध्य रेखा के उत्तर या दक्षिण में किसी भौगोलिक स्थान की कर्ण दूरी को संदर्भित करता है;
• Latitude प्रत्येक ब्लॉक समूह के लिए पृथ्वी के भूमध्य रेखा के पूर्व या पश्चिम में किसी भौगोलिक स्थान की कर्ण दूरी को संदर्भित करता है;
• Housing median age उस ब्लॉक समूह से संबंधित लोगों की माध्य आयु है। नोट कि माध्य वह मान है जो प्रेक्षित मूल्यों के फ्रीक्वेंसी वितरण के मध्य बिंदु पर होता है;
• Total rooms प्रति ब्लॉक समूह घरों में कमरों की कुल संख्या है;
• Total bedrooms प्रति ब्लॉक समूह घरों में बेडरूम की कुल संख्या है;
• Population किसी ब्लॉक समूह के निवासियों की संख्या है;
• Households प्रति ब्लॉक समूह घरों की इकाइयों और उनके निवासियों को संदर्भित करता है;
• Median income उस ब्लॉक समूह से संबंधित लोगों की माध्य आय को दर्ज करने के लिए उपयोग होता है; और,
• Median house value आश्रित वेरिएबल है और प्रति ब्लॉक समूह माध्य घर मूल्य को संदर्भित करता है।

इसके अलावा, आप यह भी जानेंगे कि जिन सभी ब्लॉक समूहों में स्वतंत्र और आश्रित वेरिएबल्स के लिए शून्य प्रविष्टियाँ थीं, उन्हें डेटा से बाहर कर दिया गया है।

Median house value आश्रित वेरिएबल है और आपके ML मॉडल में टारगेट वेरिएबल की भूमिका निभाएगा।

आप डेटा यहाँ से डाउनलोड कर सकते हैं। houses.zip फ़ोल्डर ढूँढें, उसे डाउनलोड करें और अनटार करें ताकि आप डेटा फ़ोल्डरों तक पहुँच सकें।

अगला, आप textFile() मेथड का उपयोग करके उस फ़ोल्डर से डेटा को RDDs में पढ़ेंगे जहाँ आपने इसे डाउनलोड किया है। यह मेथड फ़ाइल के लिए एक URI लेता है, जो इस मामले में आपकी मशीन का लोकल पाथ है, और उसे लाइनों के कलेक्शन के रूप में पढ़ता है। सुविधा के लिए, आप न केवल .data फ़ाइल पढ़ेंगे, बल्कि .domain फ़ाइल भी, जिसमें हेडर है। इससे आपको वेरिएबल्स के क्रम की डबल-चेकिंग करने में सुविधा होगी।

# Load in the data
rdd = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.data')

# Load in the header
header = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.domain')संचालित द्वारा

डेटा एक्सप्लोरेशन

आपने पहले ही उस वेब पेज को देखकर बहुत सी जानकारी जुटा ली है जहाँ से आपने डेटा सेट पाया, लेकिन Spark के साथ Python की मदद से अपने डेटा का व्यावहारिक रूप से निरीक्षण करना हमेशा बेहतर होता है।

यहाँ समझना महत्वपूर्ण है कि, क्योंकि Spark का निष्पादन “लेज़ी” है, अभी तक कुछ भी निष्पादित नहीं हुआ है। आपका डेटा वास्तव में पढ़ा नहीं गया है। rdd और header वेरिएबल वास्तव में अभी सिर्फ़ वैचारिक ऑब्जेक्ट हैं। आपको Spark को अपने लिए काम करने के लिए धकेलना होगा, तो चलिए header को देखने के लिए collect() मेथड का उपयोग करते हैं:

header.collect()संचालित द्वारा

collect() मेथड पूरे RDD को एक ही मशीन पर ले आता है, और आपको निम्न परिणाम दिखाई देगा:

[u'longitude: continuous.', u'latitude: continuous.', u'housingMedianAge: continuous. ', u'totalRooms: continuous. ', u'totalBedrooms: continuous. ', u'population: continuous. ', u'households: continuous. ', u'medianIncome: continuous. ', u'medianHouseValue: continuous. ']संचालित द्वारा

टिप: collect() का उपयोग करते समय सावधानी बरतें! इस लाइन को चलाने से संभवतः ड्राइवर की मेमोरी खत्म हो सकती है। यही कारण है कि यदि आप केवल RDD के कुछ एलिमेंट्स प्रिंट करना चाहते हैं, तो take() मेथड वाला निम्न तरीका अधिक सुरक्षित है। सामान्यतः, जहाँ संभव हो, अपने रिज़ल्ट सेट को सीमित रखना एक अच्छा सिद्धांत है, बिल्कुल वैसे ही जैसे जब आप SQL इस्तेमाल करते हैं।

आप सीखते हैं कि वेरिएबल्स का क्रम वही है जो आपने ऊपर डेटा सेट की प्रस्तुति में देखा था, और आप यह भी सीखते हैं कि सभी कॉलम्स में कंटीन्युअस वैल्यूज होनी चाहिए। आइए Spark को थोड़ा और काम करने के लिए बाध्य करें और California housing डेटा पर नज़र डालें ताकि इसे कन्फर्म किया जा सके।

अपने RDD पर take() मेथड कॉल करें:

rdd.take(2)संचालित द्वारा

पिछली लाइन चलाने से, आप RDD के पहले 2 एलिमेंट्स लेते हैं। परिणाम वैसा ही है जैसा आपने सोचा था: क्योंकि आपने फ़ाइलों को textFile() फ़ंक्शन से पढ़ा, लाइन्स बस एक साथ पढ़ ली गई हैं। एंट्रीज़ एक सिंगल कॉमा से अलग की गई हैं और रो भी एक कॉमा से अलग हैं:

[u'-122.230000,37.880000,41.000000,880.000000,129.000000,322.000000,126.000000,8.325200,452600.000000', u'-122.220000,37.860000,21.000000,7099.000000,1106.000000,2401.000000,1138.000000,8.301400,358500.000000']संचालित द्वारा

आपको निश्चित रूप से इसे हल करने की आवश्यकता है। अब, आपको एंट्रीज़ को स्प्लिट नहीं करना है, लेकिन आपको यह सुनिश्चित करने की ज़रूरत है कि आपके डेटा की पंक्तियाँ अलग-अलग एलिमेंट्स हों। इसे हल करने के लिए, आप map() फ़ंक्शन का उपयोग करेंगे जिसमें आप एक लैम्ब्डा फ़ंक्शन पास करेंगे जो कॉमा पर लाइन को स्प्लिट करता है। फिर, पहले की तरह ही take() के साथ परिणाम की जाँच करें:

याद रखें कि लैम्ब्डा फ़ंक्शन गुमनाम फ़ंक्शन होते हैं जो रनटाइम पर बनाए जाते हैं।

# Split lines on commas
rdd = rdd.map(lambda line: line.split(","))

# Inspect the first 2 lines 
rdd.take(2)संचालित द्वारा

आपको निम्न परिणाम मिलेगा:

[[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000'], [u'-122.220000', u'37.860000', u'21.000000', u'7099.000000', u'1106.000000', u'2401.000000', u'1138.000000', u'8.301400', u'358500.000000']]संचालित द्वारा

वैकल्पिक रूप से, आप अपना डेटा देखने के लिए निम्न फ़ंक्शंस का भी उपयोग कर सकते हैं:

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)संचालित द्वारा

यदि आप Pandas या R में डेटा फ्रेम्स के साथ काम करने के आदी हैं, तो आपने शायद हेडर देखने की उम्मीद भी की होगी, लेकिन यहाँ कोई हेडर नहीं है। अपनी ज़िंदगी आसान बनाने के लिए, आप RDD से आगे बढ़ेंगे और उसे DataFrame में बदल देंगे। जहाँ भी संभव हो, DataFrames को RDDs पर प्राथमिकता दी जाती है। खासकर जब आप Python के साथ काम कर रहे हों, तो DataFrames का प्रदर्शन RDDs से बेहतर होता है।

लेकिन दोनों में अंतर क्या है?

जब आप अपने अनस्ट्रक्चर्ड डेटा पर लो-लेवल ट्रांसफॉर्मेशन और एक्शन करना चाहते हैं तब आप RDDs का उपयोग कर सकते हैं। इसका मतलब है कि आप प्रोसेसिंग के दौरान किसी स्कीमा को थोपने या एट्रिब्यूट्स को नाम या कॉलम से एक्सेस करने की परवाह नहीं करते। प्रदर्शन के बारे में पहले कही गई बातों से जुड़ते हुए, RDDs का उपयोग करके, आप (सेमी-) स्ट्रक्चर्ड डेटा के लिए DataFrames जो प्रदर्शन लाभ देते हैं, उन्हें आवश्यक रूप से नहीं चाहते। RDDs का उपयोग तब करें जब आप डोमेन-विशिष्ट अभिव्यक्तियों की बजाय फंक्शनल प्रोग्रामिंग कंस्ट्रक्ट्स से डेटा मैनिप्युलेट करना चाहते हैं।

संक्षेप में, अब आप DataFrames पर स्विच करेंगे ताकि हाई-लेवल एक्सप्रेशंस का उपयोग कर सकें, अपने डेटा को और आगे एक्सप्लोर करने के लिए SQL क्वेरी कर सकें और कॉलमनर एक्सेस प्राप्त कर सकें।

तो चलिए यह करते हैं।

पहला कदम एक SchemaRDD या स्कीमा के साथ Row ऑब्जेक्ट्स का RDD बनाना है। यह सामान्य है, क्योंकि बिल्कुल DataFrame की तरह, आप अंततः ऐसी स्थिति में आना चाहते हैं जहाँ आपके पास रो और कॉलम हों। प्रत्येक एंट्री एक रो और किसी निश्चित कॉलम से जुड़ी होती है और कॉलम्स के डेटा टाइप होते हैं।

आप फिर से map() फ़ंक्शन और एक अन्य लैम्ब्डा फ़ंक्शन का उपयोग करेंगे जिसमें आप प्रत्येक एंट्री को Row के एक फ़ील्ड से मैप करेंगे। इसे और दृश्यात्मक बनाने के लिए, इस पहली लाइन पर विचार करें:

[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000']संचालित द्वारा

लैम्ब्डा फ़ंक्शन कहता है कि आप SchemaRDD में एक रो बनाने जा रहे हैं और index 0 पर मौजूद एलिमेंट का नाम “longitude” होगा, और इसी तरह आगे।

इस SchemaRDD के साथ, आप toDF() मेथड के साथ आसानी से RDD को DataFrame में कन्वर्ट कर सकते हैं।

# Import the necessary modules 
from pyspark.sql import Row

# Map the RDD to a DF
df = rdd.map(lambda line: Row(longitude=line[0], 
                              latitude=line[1], 
                              housingMedianAge=line[2],
                              totalRooms=line[3],
                              totalBedRooms=line[4],
                              population=line[5], 
                              households=line[6],
                              medianIncome=line[7],
                              medianHouseValue=line[8])).toDF()संचालित द्वारा

अब जब आपके पास अपना DataFrame df है, आप इसे पहले उपयोग की गई मेथड्स, यानी first() और take(), के साथ-साथ head() और show() से भी देख सकते हैं:

# Show the top 20 rows 
df.show()संचालित द्वारा

आप तुरंत देखेंगे कि यह उस RDD से बहुत अलग दिखता है जिसके साथ आप पहले काम कर रहे थे:

टिप: अपने DataFrame के कॉलम्स वापस पाने के लिए df.columns का उपयोग करें।

डेटा सभी अच्छे से कॉलम्स में क्रमबद्ध दिखता है, लेकिन डेटा टाइप्स का क्या? अपने डेटा को पढ़ने पर, Spark स्कीमा का अनुमान लगाने की कोशिश करेगा, लेकिन क्या यहाँ यह सफल हुआ है? अपने DataFrame में शामिल डेटा टाइप्स के बारे में अधिक जानने के लिए df.dtypes या df.printSchema() में से किसी एक का उपयोग करें।

# Print the data types of all `df` columns
# df.dtypes

# Print the schema of `df`
df.printSchema()संचालित द्वारा

क्योंकि आप पहली लाइन निष्पादित नहीं करते, आपको केवल निम्न परिणाम मिलेगा:

root
 |-- households: string (nullable = true)
 |-- housingMedianAge: string (nullable = true)
 |-- latitude: string (nullable = true)
 |-- longitude: string (nullable = true)
 |-- medianHouseValue: string (nullable = true)
 |-- medianIncome: string (nullable = true)
 |-- population: string (nullable = true)
 |-- totalBedRooms: string (nullable = true)
 |-- totalRooms: string (nullable = true)संचालित द्वारा

सभी कॉलम अब भी string डेटा टाइप के हैं… यह निराशाजनक है!

यदि आप इस DataFrame के साथ आगे बढ़ना चाहते हैं, तो आपको इस स्थिति को सुधारना होगा और सभी कॉलम्स को “बेहतर” या अधिक सटीक डेटा टाइप असाइन करने होंगे। आपके प्रदर्शन को भी इससे लाभ होगा। सहज रूप से, आप निम्न जैसी किसी समाधान की ओर जा सकते हैं, जहाँ आप डिक्लेयर करते हैं कि DataFrame df का हर कॉलम FloatType() में कास्ट किया जाना चाहिए:

from pyspark.sql.types import *

df = df.withColumn("longitude", df["longitude"].cast(FloatType())) \
   .withColumn("latitude", df["latitude"].cast(FloatType())) \
   .withColumn("housingMedianAge",df["housingMedianAge"].cast(FloatType())) \
   .withColumn("totalRooms", df["totalRooms"].cast(FloatType())) \ 
   .withColumn("totalBedRooms", df["totalBedRooms"].cast(FloatType())) \ 
   .withColumn("population", df["population"].cast(FloatType())) \ 
   .withColumn("households", df["households"].cast(FloatType())) \ 
   .withColumn("medianIncome", df["medianIncome"].cast(FloatType())) \ 
   .withColumn("medianHouseValue", df["medianHouseValue"].cast(FloatType()))संचालित द्वारा

लेकिन ये दोहरावदार कॉल्स काफ़ी उलझाऊ, गलती-प्रवण हैं और अच्छे नहीं दिखते। क्यों न आप एक फ़ंक्शन लिखें जो यह सब आपके लिए अधिक साफ़ तरीके से कर दे?

निम्न यूज़र-डिफाइंड फ़ंक्शन (UDF) एक DataFrame, कॉलम नाम और नया डेटा टाइप लेता है जो आप कॉलम्स को देना चाहते हैं। आप कहते हैं कि हर कॉलम नाम के लिए, आप कॉलम लेते हैं और उसे नए डेटा टाइप में कास्ट करते हैं। फिर, आप DataFrame वापस करते हैं:

# Import all from `sql.types`
from pyspark.sql.types import *

# Write a custom function to convert the data type of DataFrame columns
def convertColumn(df, names, newType):
  for name in names: 
     df = df.withColumn(name, df[name].cast(newType))
  return df 

# Assign all column names to `columns`
columns = ['households', 'housingMedianAge', 'latitude', 'longitude', 'medianHouseValue', 'medianIncome', 'population', 'totalBedRooms', 'totalRooms']

# Conver the `df` columns to `FloatType()`
df = convertColumn(df, columns, FloatType())संचालित द्वारा

यह पहले से काफी बेहतर दिखता है! आप पहले की तरह ही printSchema() मेथड से df के डेटा टाइप्स को जल्दी से देख सकते हैं।

अब जब यह सब सुलझ गया है, तो अब सच में डेटा एक्सप्लोरेशन शुरू करने का समय है। आपने देखा कि कॉलमनर एक्सेस और SQL क्वेरी करना DataFrames उपयोग करने के दो फायदे थे। अब इसमें थोड़ा और गहराई तक जाने का समय है। छोटा शुरू करते हैं और df से केवल दो कॉलम सिलेक्ट करते हैं जिनकी केवल 10 रो आपको देखनी हैं:

df.select('population','totalBedRooms').show(10)संचालित द्वारा

यह क्वेरी आपको निम्न रिज़ल्ट देगी:

+----------+-------------+
|population|totalBedRooms|
+----------+-------------+
|     322.0|        129.0|
|    2401.0|       1106.0|
|     496.0|        190.0|
|     558.0|        235.0|
|     565.0|        280.0|
|     413.0|        213.0|
|    1094.0|        489.0|
|    1157.0|        687.0|
|    1206.0|        665.0|
|    1551.0|        707.0|
+----------+-------------+
only showing top 10 rowsसंचालित द्वारा

आप अपनी क्वेरीज को और जटिल भी बना सकते हैं, जैसा कि निम्न उदाहरण में दिखता है:

df.groupBy("housingMedianAge").count().sort("housingMedianAge",ascending=False).show()संचालित द्वारा

जो आपको यह परिणाम देता है:

+----------------+-----+                                                        
|housingMedianAge|count|
+----------------+-----+
|            52.0| 1273|
|            51.0|   48|
|            50.0|  136|
|            49.0|  134|
|            48.0|  177|
|            47.0|  198|
|            46.0|  245|
|            45.0|  294|
|            44.0|  356|
|            43.0|  353|
|            42.0|  368|
|            41.0|  296|
|            40.0|  304|
|            39.0|  369|
|            38.0|  394|
|            37.0|  537|
|            36.0|  862|
|            35.0|  824|
|            34.0|  689|
|            33.0|  615|
+----------------+-----+
only showing top 20 rowsसंचालित द्वारा

क्वेरी करने के अलावा, आप अपने डेटा का वर्णन करने और कुछ समरी स्टैटिस्टिक्स प्राप्त करने का विकल्प भी चुन सकते हैं। यह आगे आपकी निश्चित रूप से मदद करेगा!

df.describe().show()संचालित द्वारा

सभी (सांख्यिकीय) एट्रिब्यूट्स के न्यूनतम और अधिकतम मूल्यों पर नज़र डालें। आप देखते हैं कि कई एट्रिब्यूट्स में मूल्यों की व्यापक रेंज है: आपको अपने डेटा सेट को नॉर्मलाइज़ करने की आवश्यकता होगी।

डेटा प्रीप्रोसेसिंग

अपने छोटे एक्सप्लोरेटरी डेटा विश्लेषण से जुटाई गई इस सारी जानकारी के साथ, आप डेटा को मॉडल में फीड करने के लिए प्रीप्रोसेस करने के लिए पर्याप्त जानते हैं।

• आपको मिसिंग वैल्यूज़ की परवाह नहीं करनी चाहिए; सभी शून्य वैल्यूज़ को डेटा सेट से बाहर रखा गया है।
• आपको शायद अपना डेटा स्टैंडर्डाइज़ करना चाहिए, जैसा कि आपने देखा कि न्यूनतम और अधिकतम मूल्यों की रेंज काफ़ी बड़ी है।
• संभवतः कुछ अतिरिक्त एट्रिब्यूट्स हैं जिन्हें आप जोड़ सकते हैं, जैसे प्रति रूम बेडरूम की संख्या या प्रति हाउसहोल्ड रूम्स की संख्या।
• आपका आश्रित वेरिएबल भी काफ़ी बड़ा है; अपनी ज़िंदगी आसान बनाने के लिए, आपको मूल्यों को थोड़ा समायोजित करना होगा।

# Import all from `sql.functions` 
from pyspark.sql.functions import *

# Adjust the values of `medianHouseValue`
df = df.withColumn("medianHouseValue", col("medianHouseValue")/100000)

# Show the first 2 lines of `df`
df.take(2)संचालित द्वारा

[Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0), Row(households=1138.0, housingMedianAge=21.0, latitude=37.86000061035156, longitude=-122.22000122070312, medianHouseValue=3.585, medianIncome=8.301400184631348, population=2401.0, totalBedRooms=1106.0, totalRooms=7099.0)]संचालित द्वारा

# Import all from `sql.functions` if you haven't yet
from pyspark.sql.functions import *

# Divide `totalRooms` by `households`
roomsPerHousehold = df.select(col("totalRooms")/col("households"))

# Divide `population` by `households`
populationPerHousehold = df.select(col("population")/col("households"))

# Divide `totalBedRooms` by `totalRooms`
bedroomsPerRoom = df.select(col("totalBedRooms")/col("totalRooms"))

# Add the new columns to `df`
df = df.withColumn("roomsPerHousehold", col("totalRooms")/col("households")) \
   .withColumn("populationPerHousehold", col("population")/col("households")) \
   .withColumn("bedroomsPerRoom", col("totalBedRooms")/col("totalRooms"))
   
# Inspect the result
df.first()संचालित द्वारा

Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0, roomsPerHousehold=6.984126984126984, populationPerHousehold=2.5555555555555554, bedroomsPerRoom=0.14659090909090908)संचालित द्वारा

# Re-order and select columns
df = df.select("medianHouseValue", 
              "totalBedRooms", 
              "population", 
              "households", 
              "medianIncome", 
              "roomsPerHousehold", 
              "populationPerHousehold", 
              "bedroomsPerRoom")संचालित द्वारा

# Import `DenseVector`
from pyspark.ml.linalg import DenseVector

# Define the `input_data` 
input_data = df.rdd.map(lambda x: (x[0], DenseVector(x[1:])))

# Replace `df` with the new DataFrame
df = spark.createDataFrame(input_data, ["label", "features"])संचालित द्वारा

# Import `StandardScaler` 
from pyspark.ml.feature import StandardScaler

# Initialize the `standardScaler`
standardScaler = StandardScaler(inputCol="features", outputCol="features_scaled")

# Fit the DataFrame to the scaler
scaler = standardScaler.fit(df)

# Transform the data in `df` with the scaler
scaled_df = scaler.transform(df)

# Inspect the result
scaled_df.take(2)संचालित द्वारा

[Row(label=4.526, features=DenseVector([129.0, 322.0, 126.0, 8.3252, 6.9841, 2.5556, 0.1466]), features_scaled=DenseVector([0.3062, 0.2843, 0.3296, 4.3821, 2.8228, 0.2461, 2.5264])), Row(label=3.585, features=DenseVector([1106.0, 2401.0, 1138.0, 8.3014, 6.2381, 2.1098, 0.1558]), features_scaled=DenseVector([2.6255, 2.1202, 2.9765, 4.3696, 2.5213, 0.2031, 2.6851]))]संचालित द्वारा

# Split the data into train and test sets
train_data, test_data = scaled_df.randomSplit([.8,.2],seed=1234)संचालित द्वारा

# Import `LinearRegression`
from pyspark.ml.regression import LinearRegression

# Initialize `lr`
lr = LinearRegression(labelCol="label", maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the data to the model
linearModel = lr.fit(train_data)संचालित द्वारा

# Generate predictions
predicted = linearModel.transform(test_data)

# Extract the predictions and the "known" correct labels
predictions = predicted.select("prediction").rdd.map(lambda x: x[0])
labels = predicted.select("label").rdd.map(lambda x: x[0])

# Zip `predictions` and `labels` into a list
predictionAndLabel = predictions.zip(labels).collect()

# Print out first 5 instances of `predictionAndLabel` 
predictionAndLabel[:5]संचालित द्वारा

[(1.4491508524918457, 0.14999), (1.5705029404692372, 0.14999), (2.148727956912464, 0.14999), (1.5831547768979277, 0.344), (1.5182107797955968, 0.398)]संचालित द्वारा

# Coefficients for the model
linearModel.coefficients

# Intercept for the model
linearModel.interceptसंचालित द्वारा

# The coefficients
[0.0,0.0,0.0,0.276239709215,0.0,0.0,0.0]

# The intercept
0.990399577462संचालित द्वारा

# Get the RMSE
linearModel.summary.rootMeanSquaredError

# Get the R2
linearModel.summary.r2संचालित द्वारा

# RMSE
0.8692118678997669

# R2
0.4240895287218379
संचालित द्वारा

spark.stop()संचालित द्वारा