Hướng dẫn Apache Spark: ML với PySpark

Tải pyspark với pip

Sau đó, bạn có thể tải và cài đặt PySpark với sự trợ giúp của pip. Việc này khá dễ và giống như cài bất kỳ gói nào khác. Bạn chỉ cần chạy lệnh quen thuộc và phần nặng nhọc sẽ được xử lý giúp bạn:

$ pip install pyspark

Hoặc, bạn cũng có thể truy cập trang tải Spark. Giữ nguyên tùy chọn mặc định ở ba bước đầu và bạn sẽ thấy liên kết tải ở bước 4. Nhấp vào liên kết đó để tải. Trong hướng dẫn này, bạn sẽ tải Spark phiên bản 2.2.0 và gói “Pre-built for Apache Hadoop 2.7 and later”.

Lưu ý rằng việc tải xuống có thể mất một lúc!

Tải Spark với Homebrew

Bạn cũng có thể cài Spark với Homebrew, một trình quản lý gói miễn phí và mã nguồn mở. Điều này đặc biệt tiện lợi nếu bạn làm việc với macOS.

Chỉ cần chạy các lệnh sau để tìm Spark, lấy thêm thông tin và cuối cùng cài nó trên máy tính cá nhân của bạn:

# Search for spark
$ brew search spark

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

# Install apache-spark
$ brew install apache-spark

Tải và Thiết lập Spark

Tiếp theo, hãy chắc chắn rằng bạn giải nén thư mục xuất hiện trong thư mục Downloads của bạn. Việc này có thể diễn ra tự động cho bạn bằng cách nhấp đúp vào tệp nén spark-2.2.0-bin-hadoop2.7.tgz hoặc mở Terminal và chạy lệnh sau:

$ tar xvf spark-2.2.0-bin-hadoop2.7.tgz

Tiếp theo, di chuyển thư mục đã giải nén đến /usr/local/spark bằng cách chạy dòng sau:

$ mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark

Lưu ý rằng nếu bạn gặp lỗi báo không có quyền để di chuyển thư mục này đến vị trí mới, bạn nên thêm sudo trước lệnh. Dòng trên sẽ trở thành $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark. Bạn sẽ được nhắc nhập mật khẩu, thường là mật khẩu bạn dùng để mở khóa máy tính khi khởi động :)

Bây giờ mọi thứ đã sẵn sàng, mở tệp README trong đường dẫn /usr/local/spark. Bạn có thể làm điều này bằng cách thực thi

$ cd /usr/local/spark

Điều này sẽ đưa bạn đến đúng thư mục. Sau đó, bạn có thể bắt đầu kiểm tra thư mục và đọc tệp README được đính kèm.

Đầu tiên, dùng $ ls để liệt kê các tệp và thư mục trong thư mục spark này. Bạn sẽ thấy có tệp README.md ở đó. Bạn có thể mở nó bằng cách thực thi một trong các lệnh sau:

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.md

Mẹo hãy dùng phím tab trên bàn phím để tự động hoàn thành khi bạn đang gõ tên tệp :) Điều này sẽ tiết kiệm thời gian cho bạn.

Bạn sẽ thấy README này cung cấp thông tin tổng quan về Spark, tài liệu trực tuyến, cách build Spark, các shell tương tác Scala và Python, chương trình ví dụ và nhiều hơn nữa.

Điều có thể bạn quan tâm nhất ở đây là phần cách build Spark nhưng lưu ý rằng điều này chỉ thực sự liên quan nếu bạn chưa tải phiên bản dựng sẵn. Tuy nhiên, với hướng dẫn này, bạn đã tải một bản dựng sẵn. Bạn có thể nhấn CTRL + X để thoát README, quay lại thư mục spark.

Nếu bạn đã chọn một phiên bản chưa được build, hãy nhớ chạy lệnh được liệt kê trong tệp README. Tại thời điểm viết, đó là lệnh sau:

$ build/mvn -DskipTests clean package run

Lưu ý rằng lệnh này có thể chạy khá lâu.

Ứng dụng Spark và Spark Shell

Shell tương tác là một ví dụ về môi trường Read-Eval(uate)-Print-Loop (REPL); Nghĩa là bất cứ thứ gì bạn gõ vào sẽ được đọc, thực thi và in ra để bạn tiếp tục phân tích. Điều này có thể khiến bạn nhớ đến IPython, một shell Python mạnh mẽ mà bạn có thể quen thuộc khi làm việc với Jupyter. Nếu muốn biết thêm, hãy cân nhắc đọc bài viết IPython hay Jupyter của DataCamp.

Điều này có nghĩa là bạn có thể dùng shell, vốn có cho cả Python cũng như Scala, cho mọi công việc tương tác mà bạn cần làm.

Bên cạnh shell này, bạn cũng có thể viết và triển khai các ứng dụng Spark. Trái ngược với việc viết ứng dụng Spark, SparkSession đã được tạo sẵn cho bạn để bạn có thể bắt đầu làm việc ngay và không tốn thời gian tạo nó.

Bây giờ bạn có thể thắc mắc: SparkSession là gì?

Đó là điểm vào chính cho chức năng Spark: nó đại diện cho kết nối tới một cụm Spark và bạn có thể dùng nó để tạo RDD và broadcast biến trên cụm đó. Khi bạn làm việc với Spark, mọi thứ bắt đầu và kết thúc với SparkSession này. Lưu ý rằng trước Spark 2.0.0, ba đối tượng kết nối chính là SparkContext, SqlContext và HiveContext.

Bạn sẽ thấy thêm về điều này sau. Giờ hãy tập trung vào shell.

Python Spark Shell

Từ bên trong thư mục spark tại /usr/local/spark, bạn có thể chạy

$ ./bin/pyspark

Lúc đầu, bạn sẽ thấy một số dòng văn bản xuất hiện. Và sau đó, bạn sẽ thấy “Spark” hiện ra, như thế này:

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'.
>>>

Khi thấy như vậy, bạn biết rằng mình đã sẵn sàng bắt đầu thử nghiệm trong shell tương tác!

Mẹo: nếu bạn thích dùng shell IPython thay vì Spark shell, bạn có thể làm điều này bằng cách đặt biến môi trường sau:

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')]

PySpark trong Jupyter Notebook

Ở phần này của hướng dẫn, bạn sẽ không dùng ishell mà sẽ xây dựng ứng dụng của riêng bạn. Bạn sẽ làm điều này trong Jupyter Notebook. Bạn đã có sẵn mọi thứ cần thiết, nên không cần làm nhiều để PySpark hoạt động trong Jupyter.

Bạn có thể khởi chạy ứng dụng notebook như bình thường, bằng cách chạy $ jupyter notebook. Sau đó, bạn tạo một notebook mới và chỉ đơn giản import thư viện findspark và dùng hàm init(). Ở đây, bạn sẽ cung cấp đường dẫn /usr/local/spark cho init() vì bạn chắc chắn đây là đường dẫn nơi bạn đã cài Spark.

# Import findspark 
import findspark

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

# Or use this alternative
#findspark.init()

Mẹo: nếu bạn không chắc đường dẫn đã đặt đúng hoặc bạn đã cài Spark ở đâu trên máy, bạn luôn có thể dùng findspark.find() để tự động phát hiện vị trí cài Spark.

Nếu bạn tìm kiếm các cách khác để làm việc với Spark trong Jupyter, hãy xem Apache Spark in Python: Hướng dẫn cho người mới bắt đầu.

Bây giờ mọi thứ đã ổn thỏa, cuối cùng bạn có thể bắt đầu tạo chương trình Spark đầu tiên!

Tạo Chương trình Spark Đầu tiên

Điều đầu tiên bạn muốn làm là import SparkContext từ gói pyspark và khởi tạo nó. Nhớ rằng trước đó bạn không phải làm việc này vì Spark shell tương tác đã tự động tạo và khởi tạo cho bạn! Ở đây, bạn sẽ cần tự làm thêm một chút :)

Import mô-đun SparkSession từ pyspark.sql và xây dựng một SparkSession với phương thức builder(). Sau đó, bạn có thể đặt URL master để kết nối, tên ứng dụng, thêm một số cấu hình bổ sung như bộ nhớ executor và cuối cùng dùng getOrCreate() để lấy phiên Spark hiện có hoặc tạo một phiên nếu chưa có cái nào chạy.

# 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

Lưu ý rằng nếu bạn gặp lỗi FileNotFoundError tương tự: “No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’”, bạn biết rằng mình phải (đặt lại) biến PATH của Spark. Về thư mục home bằng cách chạy $ cd rồi chỉnh sửa tệp .bash_profile bằng $ nano .bash_profile.

Thêm nội dung như sau vào cuối tệp

export SPARK_HOME="/usr/local/spark"

Dùng CTRL + X để thoát tệp nhưng nhớ lưu thay đổi bằng cách nhập Y để xác nhận. Tiếp theo, đừng quên áp dụng thay đổi bằng cách chạy source .bash_profile.

Mẹo: bạn cũng có thể đặt thêm các biến môi trường nếu muốn; Có lẽ bạn không cần, nhưng biết rằng bạn có thể đặt khi cần là điều tốt. Xem các ví dụ sau:

# 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/

Lưu ý rằng bây giờ bạn đã khởi tạo một SparkSession mặc định. Tuy nhiên, trong hầu hết trường hợp, bạn sẽ muốn cấu hình thêm. Bạn sẽ thấy điều này thực sự cần khi bắt đầu làm việc với dữ liệu lớn. Nếu muốn biết thêm, xem trang này.

Nạp Dữ liệu của bạn

Hướng dẫn này sử dụng tập dữ liệu California Housing. Nó xuất hiện trong một bài báo năm 1997 có tiêu đề Sparse Spatial Autoregressions, do Pace, R. Kelley và Ronald Barry viết và xuất bản trên tạp chí Statistics and Probability Letters. Các nhà nghiên cứu đã xây dựng tập dữ liệu này bằng cách sử dụng dữ liệu điều tra dân số California năm 1990.

Dữ liệu chứa một dòng cho mỗi nhóm khối điều tra dân số. Một nhóm khối là đơn vị địa lý nhỏ nhất mà Cục Điều tra Dân số Hoa Kỳ công bố dữ liệu mẫu (một nhóm khối thường có dân số từ 600 đến 3.000 người). Trong mẫu này, một nhóm khối trung bình có 1425,5 người sống trong một khu vực địa lý nhỏ gọn. Bạn sẽ thu thập thông tin này từ trang web này hoặc bằng cách đọc bài báo đã đề cập ở trên và bạn có thể tìm tại đây.

Những dữ liệu không gian này chứa 20.640 quan sát về giá nhà với 9 biến kinh tế:

• Kinh độ là khoảng cách góc của một địa điểm địa lý về phía bắc hoặc nam so với xích đạo trái đất cho mỗi nhóm khối;
• Vĩ độ là khoảng cách góc của một địa điểm địa lý về phía đông hoặc tây so với xích đạo trái đất cho mỗi nhóm khối;
• Tuổi nhà trung vị là độ tuổi trung vị của những người thuộc một nhóm khối. Lưu ý rằng trung vị là giá trị nằm ở điểm giữa của phân phối tần suất các giá trị quan sát được;
• Tổng số phòng là tổng số phòng trong các ngôi nhà mỗi nhóm khối;
• Tổng số phòng ngủ là tổng số phòng ngủ trong các ngôi nhà mỗi nhóm khối;
• Dân số là số cư dân của một nhóm khối;
• Hộ gia đình là số đơn vị nhà ở và người ở trong mỗi nhóm khối;
• Thu nhập trung vị dùng để ghi nhận thu nhập trung vị của những người thuộc một nhóm khối; Và,
• Giá nhà trung vị là biến phụ thuộc và đề cập đến giá nhà trung vị mỗi nhóm khối.

Hơn nữa, bạn cũng biết rằng tất cả các nhóm khối có giá trị bằng 0 cho các biến độc lập và phụ thuộc đã bị loại khỏi dữ liệu.

Giá nhà trung vị là biến phụ thuộc và sẽ được gán vai trò biến mục tiêu trong mô hình ML của bạn.

Bạn có thể tải dữ liệu tại đây. Tìm thư mục houses.zip, tải về và giải nén để bạn có thể truy cập các thư mục dữ liệu.

Tiếp theo, bạn sẽ dùng phương thức textFile() để đọc dữ liệu từ thư mục bạn đã tải về thành RDD. Phương thức này nhận một URI cho tệp, trong trường hợp này là đường dẫn cục bộ trên máy bạn, và đọc nó thành một tập hợp các dòng. Để tiện lợi, bạn sẽ không chỉ đọc tệp .data mà cả tệp .domain chứa header. Điều này sẽ cho phép bạn kiểm tra thứ tự các biến.

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

Khám phá Dữ liệu

Bạn đã thu thập được khá nhiều thông tin chỉ bằng cách xem trang web nơi bạn tìm thấy tập dữ liệu, nhưng luôn tốt hơn khi trực tiếp kiểm tra dữ liệu với sự trợ giúp của Spark với Python, trong trường hợp này.

Điều quan trọng cần hiểu ở đây là vì thực thi của Spark là “lười” (lazy), nên chưa có gì được thực thi. Dữ liệu của bạn thực ra chưa được đọc vào. Các biến rdd và header thực chất chỉ là khái niệm trong đầu bạn. Bạn phải thúc đẩy Spark làm việc cho bạn, nên hãy dùng phương thức collect() để xem header:

header.collect()

Phương thức collect() đưa toàn bộ RDD về một máy đơn, và bạn sẽ thấy kết quả sau:

[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. ']

Mẹo: hãy cẩn thận khi dùng collect()! Chạy dòng này có thể khiến driver hết bộ nhớ. Đó là lý do cách tiếp cận sau với phương thức take() an toàn hơn nếu bạn chỉ muốn in vài phần tử của RDD. Nói chung, giới hạn tập kết quả bất cứ khi nào có thể là một nguyên tắc tốt, giống như khi bạn dùng SQL.

Bạn biết rằng thứ tự các biến giống như bạn đã thấy ở phần giới thiệu tập dữ liệu bên trên, và bạn cũng biết rằng tất cả các cột nên có giá trị liên tục. Hãy buộc Spark làm thêm một chút việc và xem dữ liệu nhà ở California để xác nhận điều này.

Gọi phương thức take() trên RDD của bạn:

rdd.take(2)

Bằng cách thực thi dòng trước, bạn lấy 2 phần tử đầu tiên của RDD. Kết quả đúng như mong đợi: vì bạn đã đọc tệp bằng hàm textFile(), các dòng chỉ được đọc vào cùng nhau. Các mục được phân tách bằng dấu phẩy và chính các hàng cũng được phân tách bằng dấu phẩy:

[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']

Bạn chắc chắn cần giải quyết điều này. Giờ, bạn không cần tách các mục, nhưng bạn cần đảm bảo rằng các hàng dữ liệu là các phần tử riêng biệt. Để giải quyết, bạn sẽ dùng hàm map() với một hàm lambda để tách dòng tại dấu phẩy. Sau đó, kiểm tra kết quả bằng cách chạy lại dòng với phương thức take(), giống như trước:

Nhớ rằng các hàm lambda là hàm ẩn danh được tạo tại thời điểm chạy.

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

# Inspect the first 2 lines 
rdd.take(2)

Bạn sẽ nhận được kết quả sau:

[[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']]

Ngoài ra, bạn cũng có thể dùng các hàm sau để kiểm tra dữ liệu:

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)

Nếu bạn quen làm việc với Pandas hoặc data frame trong R, có lẽ bạn cũng mong đợi thấy header, nhưng ở đây không có. Để cuộc sống dễ dàng hơn, bạn sẽ chuyển từ RDD sang DataFrame. DataFrame được ưu tiên hơn RDD bất cứ khi nào bạn có thể dùng chúng. Đặc biệt khi làm việc với Python, hiệu năng của DataFrame tốt hơn RDD.

Vậy sự khác nhau giữa hai cái là gì?

Bạn có thể dùng RDD khi muốn thực hiện các transformation và action mức thấp trên dữ liệu phi cấu trúc. Điều này có nghĩa là bạn không quan tâm đến việc áp một schema trong khi xử lý hoặc truy cập thuộc tính theo tên hay cột. Liên quan đến điều đã nói về hiệu năng, khi dùng RDD, bạn không nhất thiết muốn lợi ích hiệu năng mà DataFrame mang lại cho dữ liệu (bán) cấu trúc. Dùng RDD khi bạn muốn thao tác dữ liệu bằng các cấu trúc lập trình hàm thay vì các biểu thức đặc thù miền.

Tóm lại, bây giờ bạn sẽ chuyển sang DataFrame để dùng các biểu thức cấp cao, thực hiện truy vấn SQL để khám phá dữ liệu sâu hơn và có quyền truy cập theo cột.

Vậy hãy làm thôi.

Bước đầu tiên là tạo một SchemaRDD hay một RDD các đối tượng Row với một schema. Điều này bình thường, vì giống như một DataFrame, bạn cuối cùng muốn có các hàng và cột. Mỗi mục liên kết với một hàng và một cột nhất định và các cột có kiểu dữ liệu.

Bạn sẽ lại dùng hàm map() và một hàm lambda khác trong đó bạn ánh xạ mỗi mục tới một trường trong một Row. Để hình dung rõ hơn, hãy xem dòng đầu tiên này:

[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']

Hàm lambda nói rằng bạn sẽ tạo một hàng trong SchemaRDD và phần tử tại chỉ số 0 sẽ có tên “longitude”, và cứ thế.

Với SchemaRDD này, bạn có thể dễ dàng chuyển RDD thành DataFrame với phương thức toDF().

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

Bây giờ bạn đã có DataFrame df, bạn có thể kiểm tra nó với các phương thức bạn đã dùng trước, cụ thể là first() và take(), nhưng cũng có head() và show():

# Show the top 20 rows 
df.show()

Bạn sẽ thấy ngay rằng trông nó rất khác so với RDD trước đó:

Mẹo: dùng df.columns để trả về các cột của DataFrame.

Dữ liệu có vẻ được sắp xếp thành cột rất gọn, nhưng còn kiểu dữ liệu thì sao? Khi đọc dữ liệu, Spark sẽ cố suy diễn schema, nhưng liệu ở đây đã thành công? Dùng df.dtypes hoặc df.printSchema() để biết thêm về các kiểu dữ liệu trong DataFrame.

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

# Print the schema of `df`
df.printSchema()

Bởi vì bạn không thực thi dòng đầu tiên, bạn sẽ chỉ nhận lại kết quả sau:

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)

Tất cả các cột vẫn có kiểu dữ liệu chuỗi… Thật thất vọng!

Nếu bạn muốn tiếp tục với DataFrame này, bạn sẽ cần khắc phục tình huống và gán kiểu dữ liệu “tốt hơn” hoặc chính xác hơn cho tất cả các cột. Hiệu năng cũng sẽ được hưởng lợi. Theo trực giác, bạn có thể chọn một giải pháp như sau, nơi bạn khai báo rằng mỗi cột của DataFrame df nên được ép kiểu sang 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()))

Nhưng các lời gọi lặp lại này khá rối, dễ lỗi và không đẹp. Tại sao bạn không viết một hàm có thể làm tất cả việc này một cách gọn gàng hơn?

Hàm do người dùng định nghĩa (UDF) sau nhận một DataFrame, tên cột và kiểu dữ liệu mới mà bạn muốn các cột có. Bạn chỉ định rằng với mỗi tên cột, bạn lấy cột đó và ép kiểu nó sang kiểu mới. Sau đó, bạn trả về 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())

Trông đã ổn hơn nhiều! Bạn có thể nhanh chóng kiểm tra kiểu dữ liệu của df với phương thức printSchema(), giống như đã làm trước đó.

Bây giờ mọi thứ đã sẵn sàng, đã đến lúc thực sự bắt đầu khám phá dữ liệu. Bạn đã thấy rằng truy cập theo cột và truy vấn SQL là hai lợi thế của việc dùng DataFrame. Vậy giờ là lúc đào sâu hơn một chút. Hãy bắt đầu nhỏ và chỉ chọn hai cột từ df mà bạn chỉ muốn xem 10 hàng:

df.select('population','totalBedRooms').show(10)

Truy vấn này cho bạn kết quả sau:

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

Bạn cũng có thể làm truy vấn phức tạp hơn, như ví dụ sau:

df.groupBy("housingMedianAge").count().sort("housingMedianAge",ascending=False).show()

Cho ra kết quả sau:

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

Ngoài truy vấn, bạn cũng có thể mô tả dữ liệu và lấy một số thống kê tóm tắt. Điều này chắc chắn sẽ giúp bạn về sau!

df.describe().show()

Hãy xem các giá trị nhỏ nhất và lớn nhất của tất cả thuộc tính (dạng số). Bạn thấy rằng nhiều thuộc tính có dải giá trị rộng: bạn sẽ cần chuẩn hóa dữ liệu.

Tiền xử lý Dữ liệu

Với tất cả thông tin thu thập được từ phân tích khám phá nhỏ, bạn biết đủ để tiền xử lý dữ liệu nhằm đưa vào mô hình.

• Bạn không cần bận tâm về giá trị thiếu; tất cả giá trị 0 đã bị loại khỏi tập dữ liệu.
• Bạn có lẽ nên chuẩn hóa dữ liệu, như bạn đã thấy dải giá trị min và max khá lớn.
• Có thể có thêm một số thuộc tính bạn có thể thêm vào, như đặc trưng ghi nhận số phòng ngủ trên mỗi phòng hoặc số phòng trên mỗi hộ.
• Biến phụ thuộc của bạn cũng khá lớn; Để dễ làm việc hơn, bạn sẽ cần điều chỉnh các giá trị một chút.

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