概述

流數(shù)據(jù)是一個在機器學習領域蓬勃發(fā)展的概念

學習如何使用PySpark來利用機器學習模型對流數(shù)據(jù)進行預測

我們將介紹流數(shù)據(jù)和Spark Streaming的基礎知識，然后深入到實現(xiàn)部分

引言

想象一下——每一秒都有8，500多條推文發(fā)布，900多張照片被上傳到Instagram，4，200多個Skype呼叫，78，000多次Google搜索，以及200多萬封電子郵件被發(fā)送（數(shù)據(jù)來自InternetLive Stats）。

我們正在以前所未有的速度和規(guī)模生產(chǎn)數(shù)據(jù)。這是在數(shù)據(jù)科學領域工作的大好時候！但是有了大量的數(shù)據(jù)后，接踵而至的是復雜的挑戰(zhàn)。

首要，如何收集大規(guī)模的數(shù)據(jù)？如何確保一旦生成并收集數(shù)據(jù)，機器學習管道就會繼續(xù)產(chǎn)生結果？這些都是業(yè)界面臨的重大挑戰(zhàn)，以及為什么流數(shù)據(jù)的概念在企業(yè)中越來越受到關注。

增加處理流數(shù)據(jù)的能力將極大地擴展當前的數(shù)據(jù)科學產(chǎn)品投資組合。這是業(yè)界急需的技能，若能熟練掌握它，將幫助你擔負起下一個數(shù)據(jù)科學角色。

因此，在本文中，我們將學習什么是流數(shù)據(jù)，了解Spark Streaming的基礎知識，然后在一個業(yè)界相關的數(shù)據(jù)集上使用Spark實現(xiàn)流數(shù)據(jù)。

什么是流數(shù)據(jù)？

社交媒體產(chǎn)生的數(shù)據(jù)是驚人的。你敢于想象存儲所有數(shù)據(jù)需要些什么嗎？這是一個復雜的過程！因此，在深入探討本文的Spark方面之前，先來理解什么是流數(shù)據(jù)。

流數(shù)據(jù)沒有離散的開始或結束。這些數(shù)據(jù)是每秒從數(shù)千個數(shù)據(jù)源中生成的，它們需要盡快進行處理和分析。大量流數(shù)據(jù)需要實時處理，例如Google搜索結果。

我們知道，在事件剛發(fā)生時一些見解會更有價值，而隨著時間的流逝它們會逐漸失去價值。以體育賽事為例——我們希望看到即時分析，即時統(tǒng)計見解，在那一刻真正享受比賽，對吧？

例如，假設你正在觀看一場羅杰·費德勒（Roger Federer）對戰(zhàn)諾瓦克·喬科維奇（Novak Djokovic）的激動人心的網(wǎng)球比賽。

這場比賽兩局打平，你想了解與費德勒的職業(yè)平均水平相比，其反手發(fā)球的百分比。是在幾天之后看到有意義，還是在決勝局開始前的那一刻看到有意義呢？

Spark Streaming的基礎知識

Spark Streaming是核心Spark API的擴展，可實現(xiàn)實時數(shù)據(jù)流的可伸縮和容錯流處理。

在轉到實現(xiàn)部分之前，先了解一下Spark Streaming的不同組成部分。

離散流

離散流（Dstream）是一個連續(xù)的數(shù)據(jù)流。對于離散流，其數(shù)據(jù)流可以直接從數(shù)據(jù)源接收，也可以在對原始數(shù)據(jù)進行一些處理后接收。

構建流應用程序的第一步是定義要從中收集數(shù)據(jù)的數(shù)據(jù)資源的批處理持續(xù)時間。如果批處理持續(xù)時間為2秒，則將每2秒收集一次數(shù)據(jù)并將其存儲在RDD中。這些RDD的連續(xù)序列鏈是一個DStream，它是不可變的，可以通過Spark用作一個分布式數(shù)據(jù)集。

考慮一個典型的數(shù)據(jù)科學項目。在數(shù)據(jù)預處理階段，我們需要轉換變量，包括將分類變量轉換為數(shù)字變量，創(chuàng)建分箱，去除異常值和很多其他的事。Spark保留了在數(shù)據(jù)上定義的所有轉換的歷史記錄。因此，無論何時發(fā)生故障，它都可以追溯轉換的路徑并重新生成計算結果。

我們希望Spark應用程序7 x 24小時持續(xù)運行。并且每當故障發(fā)生時，我們都希望它能盡快恢復。但是，在大規(guī)模處理數(shù)據(jù)的同時，Spark需要重新計算所有轉換以防出現(xiàn)故障。可以想象，這樣做的代價可能會非常昂貴。

緩存

這是應對該挑戰(zhàn)的一種方法。我們可以暫時存儲已計算（緩存）的結果，以維護在數(shù)據(jù)上定義的轉換的結果。這樣，當發(fā)生故障時，就不必一次又一次地重新計算這些轉換。

DStreams允許將流數(shù)據(jù)保留在內存中。當我們要對同一數(shù)據(jù)執(zhí)行多種運算時，這很有用。

檢查點

高速緩存在正常使用時非常有用，但是它需要大量內存。并不是每個人都有數(shù)百臺具有128 GB內存的計算機來緩存所有內容。

檢查點的概念能夠有所幫助。

檢查點是另一種保留轉換后的數(shù)據(jù)框結果的技術。它將不時地將正在運行的應用程序的狀態(tài)保存在任何可靠的存儲介質（如HDFS）上。但是，它比緩存慢，靈活性也更差。

在擁有流數(shù)據(jù)時可以使用檢查點。轉換結果取決于先前的轉換結果，并且需要保存以供使用。此外，我們還存儲檢查點元數(shù)據(jù)信息，例如用于創(chuàng)建流數(shù)據(jù)的配置以及一系列DStream操作的結果等。

流數(shù)據(jù)的共享變量

有時候需要為必須在多個集群上執(zhí)行的Spark應用程序定義諸如map，reduce或filter之類的函數(shù)。在函數(shù)中使用的變量會被復制到每臺機器（集群）中。

在這種情況下，每個集群都有一個不同的執(zhí)行器，我們想要一些可以賦予這些變量之間關系的東西。

例如：假設Spark應用程序在100個不同的集群上運行，它們捕獲了來自不同國家的人發(fā)布的Instagram圖片。

現(xiàn)在，每個集群的執(zhí)行者將計算該特定集群上的數(shù)據(jù)的結果。但是我們需要一些幫助這些集群進行交流的東西，以便獲得匯總結果。在Spark中，我們擁有共享變量，這些變量使此問題得以克服。

累加器變量

用例包括發(fā)生錯誤的次數(shù)，空白日志的數(shù)量，我們從特定國家收到請求的次數(shù)——所有這些都可以使用累加器解決。

每個集群上的執(zhí)行程序將數(shù)據(jù)發(fā)送回驅動程序進程，以更新累加器變量的值。 累加器僅適用于關聯(lián)和可交換的運算。例如，對求和和求最大值有用，而求平均值不起作用。

廣播變量

當我們使用位置數(shù)據(jù)（例如城市名稱和郵政編碼的映射）時，這些是固定變量，是吧？現(xiàn)在，如果每次在任意集群上的特定轉換都需要這種類型的數(shù)據(jù)，我們不需要向驅動程序發(fā)送請求，因為它會太昂貴。

相反，可以在每個集群上存儲此數(shù)據(jù)的副本。這些類型的變量稱為廣播變量。

廣播變量允許程序員在每臺計算機上保留一個只讀變量。通常，Spark使用高效的廣播算法自動分配廣播變量，但是如果有任務需要多個階段的相同數(shù)據(jù)，也可以定義它們。

使用PySpark對流數(shù)據(jù)進行情感分析

是時候啟動你最喜歡的IDE了！讓我們在本節(jié)中進行編碼，并以實踐的方式理解流數(shù)據(jù)。

理解問題陳述

在本節(jié)我們將使用真實數(shù)據(jù)集。我們的目標是檢測推文中的仇恨言論。為了簡單起見，如果一條推文包含帶有種族主義或性別歧視情緒的言論，我們就認為該推文包含仇恨言論。

因此，任務是將種族主義或性別歧視的推文從其他推文中區(qū)分出來。我們將使用包含推文和標簽的訓練樣本，其中標簽“1”表示推文是種族主義/性別歧視的，標簽“0”則表示其他種類。

為什么這是一個與主題相關的項目？因為社交媒體平臺以評論和狀態(tài)更新的形式接收龐大的流數(shù)據(jù)。該項目將幫助我們審核公開發(fā)布的內容。

設置項目工作流程

1. 模型構建：構建邏輯回歸模型管道，對推文中是否包含仇恨言論進行分類。在這里，我們的重點不是建立一個完全準確的分類模型，而是了解如何在流數(shù)據(jù)上使用任意模型并返回結果

2. 初始化Spark Streaming的環(huán)境：一旦模型構建完成，需要定義獲取流數(shù)據(jù)的主機名和端口號

3. 流數(shù)據(jù)：接下來，從定義的端口添加來自netcat服務器的推文，SparkStreaming API將在指定的持續(xù)時間后接收數(shù)據(jù)

4. 預測并返回結果：一旦接收到推文，就將數(shù)據(jù)傳遞到創(chuàng)建的機器學習管道中，并從模型中返回預測的情緒

這是對工作流程的簡潔說明：

訓練數(shù)據(jù)以建立邏輯回歸模型

我們在一個CSV文件中存儲推文數(shù)據(jù)及其相應的標簽。使用邏輯回歸模型來預測推文是否包含仇恨言論。如果是，則模型預測標簽為1（否則為0）。你可以參考“面向初學者的PySpark”來設置Spark環(huán)境。

可以在這里下載數(shù)據(jù)集和代碼。

首先，需要定義CSV文件的模式。否則，Spark會將每列數(shù)據(jù)的類型都視為字符串。讀取數(shù)據(jù)并檢查模式是否符合定義：

# importing required libraries

from pyspark import SparkContext

from pyspark.sql.session import SparkSession

from pyspark.streaming import StreamingContext

import pyspark.sql.types as tp

from pyspark.ml import Pipeline

from pyspark.ml.feature import StringIndexer， OneHotEncoderEstimator， VectorAssembler

from pyspark.ml.feature import StopWordsRemover， Word2Vec， RegexTokenizer

from pyspark.ml.classification import LogisticRegression

from pyspark.sql import Row

# initializing spark session

sc = SparkContext（appName=“PySparkShell”）

spark = SparkSession（sc）

# define the schema

my_schema = tp.StructType（［

tp.StructField（name=‘id’， dataType= tp.IntegerType（）， nullable=True），

tp.StructField（name=‘label’， dataType= tp.IntegerType（）， nullable=True），

tp.StructField（name=‘tweet’， dataType= tp.StringType（）， nullable=True）

］）

# read the dataset

my_data = spark.read.csv（‘twitter_sentiments.csv’，

schema=my_schema，

header=True）

# view the data

my_data.show（5）

# print the schema of the file

my_data.printSchema（）

定義機器學習管道的各個階段

現(xiàn)在已經(jīng)將數(shù)據(jù)保存在Spark數(shù)據(jù)框中，需要定義轉換數(shù)據(jù)的不同階段，然后使用它從模型中獲取預測的標簽。

在第一階段，使用RegexTokenizer將推特文本轉換為單詞列表。然后，從單詞列表中刪除停用詞并創(chuàng)建詞向量。在最后階段，使用這些詞向量來構建邏輯回歸模型并獲得預測的情緒。

記住——重點不是建立一個完全準確的分類模型，而是要看看如何在流數(shù)據(jù)上使用預測模型來獲取結果。

# define stage 1： tokenize the tweet text

stage_1 = RegexTokenizer（inputCol=‘tweet’ ， outputCol=‘tokens’， pattern=‘\\W’）

# define stage 2： remove the stop words

stage_2 = StopWordsRemover（inputCol=‘tokens’， outputCol=‘filtered_words’）

# define stage 3： create a word vector of the size 100

stage_3 = Word2Vec（inputCol=‘filtered_words’， outputCol=‘vector’， vectorSize=100）

# define stage 4： Logistic Regression Model

model = LogisticRegression（featuresCol=‘vector’， labelCol=‘label’）

設置機器學習管道

讓我們在Pipeline對象中添加階段，然后按順序執(zhí)行這些轉換。用訓練數(shù)據(jù)集擬合管道，現(xiàn)在，每當有了新的推文，只需要將其傳遞給管道對象并轉換數(shù)據(jù)即可獲取預測：

# setup the pipeline

pipeline = Pipeline（stages= ［stage_1， stage_2， stage_3， model］）

# fit the pipeline model with the training data

pipelineFit = pipeline.fit（my_data）

流數(shù)據(jù)和返回結果

假設每秒收到數(shù)百條評論，我們希望通過阻止用戶發(fā)布仇恨言論來保持平臺整潔。因此，每當我們收到新文本，都會將其傳遞到管道中并獲得預測的情緒。

我們將定義一個函數(shù)get_prediction，該函數(shù)將刪除空白句子并創(chuàng)建一個數(shù)據(jù)框，其中每一行都包含一條推文。

初始化Spark Streaming的環(huán)境并定義3秒的批處理持續(xù)時間。這意味著我們將對每3秒收到的數(shù)據(jù)進行預測：

# define a function to compute sentiments of the received tweets

defget_prediction（tweet_text）：

try：

# filter the tweets whose length is greater than 0

tweet_text = tweet_text.filter（lambda x： len（x） 》0）

# create a dataframe with column name ‘tweet’ and each row will contain the tweet

rowRdd = tweet_text.map（lambda w： Row（tweet=w））

# create a spark dataframe

wordsDataFrame = spark.createDataFrame（rowRdd）

# transform the data using the pipeline and get the predicted sentiment

pipelineFit.transform（wordsDataFrame）.select（‘tweet’，‘prediction’）.show（）

except ：

print（‘No data’）

# initialize the streaming context

ssc = StreamingContext（sc， batchDuration=3）

# Create a DStream that will connect to hostname:port， like localhost:9991

lines = ssc.socketTextStream（sys.argv［1］， int（sys.argv［2］））

# split the tweet text by a keyword ‘TWEET_APP’ so that we can identify which set of words is from a single tweet

words = lines.flatMap（lambda line ： line.split（‘TWEET_APP’））

# get the predicted sentiments for the tweets received

words.foreachRDD（get_prediction）

# Start the computation

ssc.start（）

# Wait for the computation to terminate

ssc.awaitTermination（）

在一個終端上運行該程序，然后使用Netcat（用于將數(shù)據(jù)發(fā)送到定義的主機名和端口號的實用工具）。你可以使用以下命令啟動TCP連接：

nc -lk port_number

最后，在第二個終端中鍵入文本，你將在另一個終端中實時獲得預測。

完美！

結語

流數(shù)據(jù)在未來幾年只會越來越熱門，因此應該真正開始熟悉這一主題。請記住，數(shù)據(jù)科學不只是建立模型——整個流程都需要關注。

本文介紹了SparkStreaming的基礎知識以及如何在真實的數(shù)據(jù)集上實現(xiàn)它。我鼓勵大家使用另一個數(shù)據(jù)集或抓取實時數(shù)據(jù)來實現(xiàn)剛剛介紹的內容（你也可以嘗試其他模型）。