[無母] K Nearest neighbour estimator

這是最近寫論文用到的東西。

冰島的火山坑(我當下其實看不出來那是什麼)

首先直接看公式

$\hat{p}\left(x_{i}\right)=\frac{1}{n-1}\frac{k}{c_{d}\ \epsilon_{k}\left(i\right)^{d}}$

其中$d$是維度，$\epsilon$則是該點的與第$k$個鄰居的距離，並且

$c_{d}=\frac{\pi^{d/2}}{\Gamma\left(d/2 + 1\right)}$

如果單獨把第二個component的分母拿出來看，其實$c_{d}\ \epsilon_{k}\left(i\right)^{d}$就是半徑為$\epsilon_{k}\left(i\right)$球體的體積，若簡寫為$V\left(i\right)$，則上式可寫成

$\frac{k}{n-1}\frac{1}{V\left(i\right)}$

整個基本概念就很清楚了，可以把這個estimator想成你在這個資料點畫出一個半徑為$\epsilon_{k}\left(i\right)$的球，並且觀察有多少點落在球內。如果你讓半徑是距離第$k$個點的距離，那麼落在球內的點當然就是$k$個。

[無關程式] 說書(1) Data and Social Good

最近網路上似乎刮起了一陣說書炫風，於是乎我也想來跟風一下。

波里(Pori), 芬蘭

緣起是之前在電子郵箱裡收到一封O'Reilly的廣告信，說只要註冊還幹嘛的就可以下載免費的電子書看，因著歐巴桑心態當然是要下載一下啦。

這本書很好玩，叫做Data and Social Good，內容是關於資料科學在社會公益上的應用。這本書沒有太多的硬知識，當做消遣的讀物蠻適合的，並且相較於目前大部分討論資料科學應用的書，因為較著重在社會公益上，這本書的內容算是蠻清流的，銅臭味比較沒那麼重。

書很薄，只有19頁，可能比多數的paper還短。內容主要是實際案例，以下列出幾個我覺得比較有趣的。

1. DataKind and Simpa Networks 

DataKind是一間非營利公司，主要工作是媒合資料科學家和社會企業或公益組織，Simpa Networks就是一個有趣的案例。

Simpa Networks是一間致力於提供印度鄉村地區太陽能發電設備的公司。能源問題在印度是一個相當嚴重的問題，將近七千五百萬個家庭沒有電力供應，或者必須使用煤油等燃料，造成許多健康問題，如眼睛疾病等。

DataKind和Simpa Networks合作，追蹤並且分析客戶的交易歷史資料，幫助Simpa Networks有效挑選新的合作客戶。

2. DataKind and CTL (Crisis Text Line)

CTL是一個提供心理支持的非營利組織，當人們在遇到重大危機的時候，往往面臨不知所措的景況，這時就可以發簡訊給CTL，受過訓練的專員透過回發簡訊提供支持、安慰或是轉介的服務。

CTL遇到的問題是，因為許多人重複的寄發簡訊，義工們花了將近34%的時間在處理3%的使用者的簡訊。DataKind和CTL合作，幫助CTL建立一套篩選出即興需求的機制，已將資源花在真正的刀口上。經過努力後，原本的34%有效將至8%，因此幫助CTL拯救了更多人的生命。

3. New York City Department of Health and Mental Hygiene (DOHMH), Yelp.com and Columbia University 

紐約市健康與心理衛生局(New York City Department of Health and Mental Hygiene, DOHMH)發現，餐廳評論網Yelp上的文章可以做為食物傳染疾病的重要資料。

因此DOHMH找上了Yelp.com和Columbia University的資料科學家，建立了一套機制，利用文字探勘工具(Text Mining Tools)篩選出有可能有問題的餐廳並介入調查。

其實書裡面提到的案例還很多，有興趣的自己去看吧！(爛尾)

[貝氏] Paper digest (4) 煩死人的Variational Bayes續集 -- Blackbox Variational Inference

這篇接著很久很久寫的煩死人的 Variational Bayes 初探往下寫的。如果用食記部落客的語法，應該改成「再訪！煩死人的Variational Bayes」(超無聊)。

好，會寫這篇的原因是因為最近拜讀了一篇由大神David Blei團隊所寫的這篇文章，順便邊讀邊寫，吸收快又好。題目叫做Blackbox Variational Inference，你沒看錯Variaional Bayes也有黑箱版的，這篇文章刊出的時間是2014年，那年台灣高中生也剛好在反黑箱課綱，真的蠻潮的。

這個方法的賣點是不需要手動計算複雜的積分也可以做Variational Inference。

先複習一下，Variational Bayes的精神就是找一個替代的替身 $q(\theta)$ 並且讓它和posterior $p(\theta, x)$ 越像越好，也就是盡全力把 $KL[q(\theta), p(\theta, x)]$ 變得很小就對了！接下來，

$KL \left[q \left( \theta \right),p \left( \theta \right|x) \right] $

$=\int q \left( \theta \right) \log \frac{q \left( \theta \right)}{p \left( \theta,x \right)} d \theta- \int q \left( \theta \right) \log p \left(x\right) d\theta$

$=\int q\left(\theta\right)\log\frac{q\left(\theta\right)}{p\left(\theta,x\right)} d\theta-\log p\left(x\right)$

$=E_{q}[q(\theta)] - E_{q}[\log p(\theta, x)]-C$

上式的$ E_{q}[q(\theta)] - E_{q}[\log p(\theta, x)]$取負號以後叫做 ELBO (Evidence lower bound)，也就是

$ ELBO = E_{q}[\log p(\theta, x)]-E_{q}[q(\theta)]$

只要不斷讓ELBO越來越大，  $KL[q(\theta), p(\theta, x)]$ 就會越來越小。

好現在假設我們的其中一個替身叫做$q(\theta|\lambda)$，最佳化 $\lambda$ 的梯度(gradient)就是

$\nabla_{\lambda}\int (\log p(\theta, x)-\log q(\theta|\lambda))q(\theta|\lambda)d\theta$

$=\int \nabla_{\lambda}(\log p(\theta, x)-\log q(\theta|\lambda))q(\theta|\lambda)d\theta$

$+\int \nabla_{\lambda}q(\theta|\lambda)(\log p(\theta, x)-\log q(\theta|\lambda))d\theta$

其中，

$\int \nabla_{\lambda}(\log p(\theta, x)-\log q(\theta|\lambda))q(\theta|\lambda)d\theta$

$=-E_{q}[\nabla_{\lambda}\log q(\theta|\lambda)]=0$

剩下另外一個，因為

$\nabla_{\lambda}q(\theta|\lambda)=q(\theta|\lambda) \nabla_{\lambda}\log q(\theta|\lambda)$

就可以寫成

$\int \nabla_{\lambda}\log q(\theta|\lambda)(\log p(\theta, x)-\log q(\theta|\lambda))q(\theta|\lambda)d\theta$

$=E_{q}[ \nabla_{\lambda}\log q(\theta|\lambda)(\log p(\theta, x)-\log q(\theta|\lambda))]$

這個東西是可以用Monte Carlo integration估計的，只要從 $q$ 中抽樣就可以了。

因此說了一堆，黑箱版本的好處其實就是不需要手動計算複雜的積分(再次強調)也可以找出梯度，然後一次做最佳化就可以了。

[Scala] Big Data(6) submit spark job

這篇是這篇(默默改版了)的Spark版，就這樣而已ㄏㄏ。

冰島，隨便一拍都可以當啤酒廣告。沒錯，我在炫耀。

劇情設定是這樣的，假設我有三個檔案在hdfs上，分別是temp/input_1.txt, temp/input_2.txt和 temp/input_3.txt，各自格式都不一樣，然後我想要計算各科的平均分數。

$ hdfs dfs -cat temp/input_1.txt
WUJ-360100;math;56
WPY-802007;math;98
FKT-670008;science;67
$ hdfs dfs -cat temp/input_2.txt
{Number:FJB-004150, Subject:math, Score:96}
{Number:QDG-300700, Subject:chinese, Score:90}
{Number:JVY-030140, Subject:chinese, Score:71}
$ hdfs dfs -cat temp/input_3.txt
[Number=>ITM-501806; Subject=>science; Score=>82]
[Number=>QBE-003981; Subject=>math; Score=>85]
[Number=>EUJ-017009; Subject=>chinese; Score=>63]

相較於Hadoop的JAVA file總是要寫得的叨叨噓噓，Spark明顯親民多了，根本就不用什麼Multipleinputs啊啊啊很棒！所有的程式碼只有這樣：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.hadoop.fs.FileSystem

object MultiInput {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("Multiple Input Example")
    val hadoopConf = new org.apache.hadoop.conf.Configuration()
    val fs = org.apache.hadoop.fs.FileSystem.get(hadoopConf)

    val inPath1 = new org.apache.hadoop.fs.Path(args(0))
    val inPath2 = new org.apache.hadoop.fs.Path(args(1))
    val inPath3 = new org.apache.hadoop.fs.Path(args(2))
    val outPath = new org.apache.hadoop.fs.Path(args(3))

    if(fs.exists(outPath)){
        fs.delete(outPath)
    }

    val sc = new SparkContext(conf)
    val file1 = sc.textFile(inPath1.toString)
    var rdd1 = file1.map(line => (line.split(";")(1), line.split(";")(2)))

    val file2 = sc.textFile(inPath2.toString)
    var rdd2 = file2.filter(_.contains(",")).map(s => (
        s.split(",")(1).split(":")(1), 
        s.split(",")(2).split(":")(1).replace("}", "")))

    val file3 = sc.textFile(inPath3.toString)
    var rdd3 = file3.map(s => (
        s.split(";")(1).split("=>")(1), 
        s.split(";")(2).split("=>")(1).replace("]", "")))

    val rdd = rdd1.union(rdd2).union(rdd3)

    val res = rdd.combineByKey(
        (x: String) => (x.toInt, 1),
        (acc:(Int, Int), x) => (acc._1.toInt + x.toInt, acc._2.toInt + 1),
        (acc1:(Int, Int), acc2:(Int, Int)) => (
            acc1._1.toInt + acc2._1.toInt, 
            acc1._2.toInt + acc2._2.toInt)
        ).map(k => (k._1, k._2._1.toDouble / k._2._2.toDouble))

    res.saveAsTextFile(outPath.toString)
    
    }
}

然後把以上程式碼存成一個叫做MultiInput.scala的檔案。

之後進入打包流程，就比較麻煩了。
首先先建立一個打包需要的資料夾，這裡就姑且叫做multiInput吧！

$ mkdir ./multiInput
$ mkdir -p ./multiInput/src/main/scala

然後把剛剛的寫好的那的scala檔案丟到/src/main/scala目錄下

$ mv MultiInput.scala ./multiInput/src/main/scala/MultiInput.scala

創建一個叫做multiInput.sbt的檔案(極重要！)

$ vim ./multiInput/simple.sbt

內容如下：

$ name := "Multiple Input"

version := "1.0"

scalaVersion := "2.10.5"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"

然後進到資料夾multiInput進行一個打包的動作(要先安裝好sbt喔)

$ cd multiInput
$ sbt package

成功打包以後，會在target/scala-2.10底下搞到一個打包好的.jar檔案

$ ls target/scala-2.10/
classes  multiple-input_2.10-1.0.jar

submit spark job的方法如下

$ spark-submit --class "MultiInput" target/scala-2.10/multiple-input_2.10-1.0.jar \ 
temp/input_1.txt temp/input_2.txt temp/input_3.txt temp/output_mp

路徑記得要打對！路徑記得要打對！路徑記得要打對！很重要！

跑完之後就可以到hdfs上看結果了

$ hdfs dfs -getmerge temp/output_mp sp_test
$ cat sp_test
(math,83.75)
(chinese,74.66666666666667)
(science,74.5)

[Scala] Big Data(5) 雜記，Cloudera Quickstart VM和萬年word count

其實蠻久之前就有用過Spark這個東西了，只是最近再拿出來玩玩發現發現好多東西都忘記，忘得精光那種，筆記這種東西真的好重要啊啊啊啊啊。

Cloudera Quickstart VM

為了練習Spark自己搞一個server雖然很霸氣，但不是所有人都有錢有閒這樣做的，如果只是為了練習，或是用少量資料驗證一下想法可不可行，誠心建議使用Cloudera Quickstart VM就好了。這個東西真的超級佛心的，一包拿來裡面該有的都有(Hadoop, Spark, Hive...)，然後設定也算簡單，尤其是對於我這種對Architecture很沒轍的技術廢物(無誤)來說。

題外話，中視《飢餓遊戲》好看！

喔對了，為了平衡報導，其實另外一家叫做Hortonworks提供的Sandbox也不錯，有興趣的人也可以試試看。

然後我選用Docker VM，因為設定超級無敵簡單的就一頁而已，指令也沒幾行，連我都搞得定簡直太棒！不過得先安裝Docker就是了。設定步驟很簡單，以下也不過是把網頁上的Instruction翻成中文而已就醬。

在安裝好Docker之後，就打開terminal，用docker pull指令開始下載Cloudera QuickstartVM。

$ docker pull cloudera/quickstart

整包東西大概4GB左右，要稍微等一下。

然後鍵入指令，裡就會發現有一個image叫做cloudera/quickstart耶！(訝異個屁)

$ docker images
EPOSITORY            TAG                 IMAGE ID            CREATED             SIZE
cloudera/quickstart  latest              4239cd2958c6        10 months ago       6.34 GB

記下對應的IMAGE ID，以上圖為例也就是4239cd2958c6，用來啟動Docker VM囉！

$ docker run \
--hostname=quickstart.cloudera \
--privileged=true \
-t -i -p 8888 \
[IMAGE ID] /usr/bin/docker-quickstart

然後那個[IMAGE ID]記得要換一下。然後基本上經過一段時間的等在就可以成功進入VM的根目錄了，我啟動的時候hue沒有成功，但是在Docker VM開始run之後重複幾次restart神奇的事情就發生了，像這樣：

[root@quickstart /]# service hue restart
Shutting down hue:
Starting hue:                                              [  OK  ]

但如果不是很在意hue的話也可以不理它啦，反正天下事只要牽涉到UI就是神煩(翻白眼)。至於確切原因我也不知到為何，反正它就是好了，很煩(知道原因的拜託留言或是托夢告訴我拜託！)。

看到Hue web UI就是心曠神怡而已

然後注意，如果這時候關掉terminal的話，所有資料都會消失，如果不想資料消失的話，請在剛剛的啟動加上-p選項或者用ctrl+p -> ctrl+q關掉terminal，這樣VM就會繼續運行。如果要重新接上，請用docker ps找到對應的conatiner id

$ docker ps
CONTAINER ID        IMAGE               COMMAND                  CREATED    
56703ef28dc3        4239cd2958c6        "/usr/bin/docker-q..."   2 hours ago

然後docker attach

$ docker attach [CONTAINER ID]

如果要在執行其他指令的話，可以用docker exec，概念上和ssh差不多

$ sudo docker exec -i -t [CONTAINER ID] /bin/bash

千篇一律的Word count problem

接下來就是用Spark跑跑看最白爛(對，即便就是白爛我還是要跑)的word count，就是計算字數啦。首先得把檔案用docker cp從本機丟過去docker VM的container裡面。

$ docker cp word_count.txt [CONTAINER ID]:/word_count.txt

然後進入docker VM，把這個剛剛丟過來的檔案推到HDFS上。

[root@quickstart /]# hdfs dfs -mkdir temp
[root@quickstart /]# hdfs dfs -put word_count.txt temp/word_count.txt

之後進入Spark shell

[root@quickstart /]# spark-shell

之後就開始跑罐頭程式

scala> val file = sc.textFile("temp/word_count.txt")
file: org.apache.spark.rdd.RDD[String] = temp/word_count.txt MapPartitionsRDD[1] at textFile at :27

scala> val counts = file.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)
counts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at :29

scala> counts.collect().foreach(println)
(Groovy..,1)
(found,1)
(Check,1)...

scala> counts.saveAsTextFile("temp/output")

寫資料路徑的時候要稍微小心一點，其實不一定像網路上的demo一樣寫完整的路徑，像這樣。

scala> val file = sc.textFile("hdfs://quickstart.cloudera:8020/user/root/temp/word_count.txt")

然後看一下HDFS看看有沒有存成功

[root@quickstart /]# hdfs dfs -ls temp/output
Found 3 items
-rw-r--r--   1 root supergroup          0 2017-02-25 19:56 temp/output/_SUCCESS
-rw-r--r--   1 root supergroup         93 2017-02-25 19:56 temp/output/part-00000
-rw-r--r--   1 root supergroup        127 2017-02-25 19:56 temp/output/part-00001