2015年，全球網際網路廣告營收接近600億美元，比2014年增長了近20%。多家網際網路巨頭都依賴於廣告營收，如谷歌，百度，Facebook，網際網路新貴們也都開始試水廣告業，如Snapchat, Pinterest， Spotify.

作為網際網路廣告的老大哥，谷歌花了很大的力氣研發自己的社交網路，Google+，並期待能與Facebook，Twitter抗衡。然後事與願違，Google+的份額依然低於1% 。

2015年，谷歌終於不再強迫使用者把Google+和谷歌家的其他服務繫結，如Youtube。筆者認為谷歌花大量人力財力研發Google+並將其與其他服務繫結的原因之一是蒐集使用者的喜好資料來為自家的廣告業務服務。更具體的說是為了更好地預測廣告的點選率。在這方面，Facebook似乎更勝一籌。2015年第一季度，Facebook廣告的轉換率比Google Display Ads Network (GDN)高出了55%。筆者並不感到驚訝，Facebook顯然比谷歌更瞭解使用者的喜好，筆者猜測使用者的行為，包括點贊, 評論，關注等，都為廣告演算法提供了關於使用者的喜好或類別的資訊。這些資訊會用於廣告的點選率預測，而點選率預測又會應用於廣告排序演算法。

今天我就帶大家來用 Spark MLlib訓練一個廣告點選率預測的模型。

環境配置

Java

Spark 2.0.0

安裝很簡單，pre-built版本的spark下載下來即可

Downloads | Apache Spark

Python

資料

Kaggle Avazu挑戰賽中的廣告資料

Click-Through Rate Prediction

元資料

id: ad identifier

click: 0/1 for non-click/click

hour: format is YYMMDDHH, so 14091123 means 23:00 on Sept. 11, 2014 UTC.

C1 -- anonymized categorical variable

banner_pos

site_id

site_domain

site_category

app_id

app_domain

app_category

device_id

device_ip

device_model

device_type

device_conn_type

C14-C21 -- anonymized categorical variables

資料預處理

我們注意到資料中有字串型別的值，如site_id,site_domain,site_category,app_id,app_domain,app_category等。我們需要將他們轉換成數值型，公式如下：

程式碼：

import os
import sys
if __name__ == "__main__":
    input_file = sys.argv[1]
    preprocess_file = sys.argv[2]
    test_flag = sys.argv[3]
    print "input=" + input_file
    print "preprocess_file=" + preprocess_file
    output = open(preprocess_file, "w")
    with open(input_file, "r") as lines:
            next(lines)
            for line in lines:
                fields = line.split(",")
                index = 5
                end = 14
                if test_flag == "1":
                    index = 4
                    end = 13
                while index < end:
                    fields[index] = str(hash(fields[index]) % 1000000)
                    index += 1

                newline = ",".join(fields)
                output.write(newline)
    output.close()
 

此外，有必要將原來的訓練資料分割成兩部分：本地訓練資料，本地測試資料。由於原來的訓練資料是基於10天的資料，我們可以把前9天資料作為本地訓練資料，把最後一天的資料作為本地測試資料。

程式碼：

import os
import sys
if __name__ == "__main__":
   input_file = sys.argv[1]
   train_file = sys.argv[2]
   test_file = sys.argv[3]
   test_start_date = sys.argv[4]
   test_data_ouput = open(test_file, "w")
   train_data_output = open(train_file, "w")


   with open(input_file, "r") as lines:
           next(lines)
           for line in lines:
               fields = line.split(",")
              if fields[2].startswith(test_start_date):
                   test_data_ouput.write(line)
               else:
                   train_data_output.write(line)


   train_data_output.close()
   test_data_ouput.close()

Spark 入門

Apache Spark是一個分散式計算框架，旨在簡化運行於計算機叢集上的並行程式的編寫。該框架對資源排程，任務的提交、執行和跟蹤，節點間的通訊以及資料並行處理的內在底層操作都進行了抽象。它提供了一個更高級別的API用於處理分散式資料。從這方面說，它與Apache Hadoop等分散式處理框架類似。但在底層架構上，Spark與它們有所不同。

我們先介紹SparkContext物件。任何Spark程式的編寫都是從SparkContext（或用Java編寫時的JavaSparkContext）開始的。SparkContext的初始化需要一個SparkConf物件，後者包含了Spark叢集配置的各種引數（比如主節點的URL）。初始化後，我們便可用SparkContext物件所包含的各種方法來建立和操作分散式資料集和共享變數。若要用Python程式碼來實現的話，可參照下面的程式碼：

sconf = SparkConf().setAppName(“WordCount") .setMaster(“local[4]")
sc = SparkContext(conf=sconf)

這段程式碼會建立一個4執行緒的SparkContext物件，並將其相應的任務命名為WordCount。

RDD（Resilient Distributed Dataset，彈性分散式資料集）是Spark的核心概念之一。一個RDD代表一系列的“記錄”（嚴格來說，某種型別的物件）。這些記錄被分配或分割槽到一個叢集的多個節點上（在本地模式下，可以類似地理解為單個程序裡的多個執行緒上）。Spark中的RDD具備容錯性，即當某個節點或任務失敗時（因非使用者程式碼錯誤的原因而引起，如硬體故障、網路不通等），RDD會在餘下的節點上自動重建，以便任務能最終完成。我們可以把RDD看成資料和基於該資料計算的單元。

RDD也可以基於Hadoop的輸入源建立，比如本地檔案系統、HDFS和Amazon S3。基於Hadoop的RDD可以使用任何實現了Hadoop InputFormat介面的輸入格式，包括文字檔案、其他Hadoop標準格式、HBase、Cassandra等。以下舉例說明如何用一個本地檔案系統裡的檔案建立RDD：

rddFromTextFile = sc.textFile(“sample.txt")

上述程式碼中的textFile函式（方法）會返回一個RDD物件。該物件的每一條記錄都是一個表示文字檔案中某一行文字的String（字串）物件。

基於RDD的操作被分為轉換（transformation）和執行（action）兩種。一般來說，轉換操作是對一個數據集裡的所有記錄執行某種函式，從而使記錄發生改變；而執行通常是執行某些計算或聚合操作，並將結果返回執行SparkContext的那個驅動程式（driver）。

常見的轉換操作: map, filer

常見的執行操作: count, take, collect, saveAsTextFile

基於SparkMLliB的模型訓練

Spark提供豐富的機器學習庫，包括：分類，迴歸，聚類等。此外還提供特徵提取演算法庫。詳見：

MLlib: Main Guide

本文將logistic regression 應用於處理過的資料，把 loss 收斂在 21.7左右

程式碼：

from __future__ import print_function
from pyspark import SparkContext
from pyspark.mllib.classification import LogisticRegressionWithLBFGS,
                                              LogisticRegressionModel
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.util import MLUtils

if __name__ == "__main__":

    sc = SparkContext(appName="CTRLogisticRegression")

    # $example on$
    # Load and parse the data
    def parsePoint(line):
        values = [float(x) for x in line.split(',')]
        features = values[0:1]
        features.extend(values[2:])
        return LabeledPoint(values[1],features)

    data = sc.textFile("train_data")
    testData = sc.textFile("test_data")

    parsedTrainData = data.map(parsePoint)
    parsedTestData = testData.map(parsePoint)

    # Build the model
    model = LogisticRegressionWithLBFGS.train(parsedTrainData)

    # Evaluating the model on training data
    labelsAndPreds = parsedTestData.map(lambda p: (p.label,    
                                       model.predict(p.features)))
    trainErr = labelsAndPreds.filter(lambda (v, p): v != p).count() /
                                     float(parsedTestData.count())
    print("Training Error = " + str(trainErr))

    # Save and load model
    model.save(sc, "target/tmp/CTR")

程式碼執行命令

$SPARKPATH/bin/spark-submit kagggleLogisstic.py

$SPARKPATH 為spark安裝目錄

延伸

由於精度太低，上述例子訓練處的模型顯然是無法用於工業界的。特徵工程（Feature engineering） 可以幫助我們提高模型的可靠性和準確性，那麼如何進行特徵工程來提高廣告點選率預測的精度呢？有哪些額外的特徵可以生成？為何Facebook的廣告轉換率比谷歌的高？這兩者間有何聯絡？