# 教程 原文鏈接 : [http://zeppelin.apache.org/docs/0.7.2/quickstart/tutorial.html](http://zeppelin.apache.org/docs/0.7.2/quickstart/tutorial.html) 譯文鏈接 : [http://www.apache.wiki/pages/viewpage.action?pageId=10030571](http://www.apache.wiki/pages/viewpage.action?pageId=10030571) 貢獻者 : [片刻](/display/~jiangzhonglian) [ApacheCN](/display/~apachecn) [Apache中文網](/display/~apachechina) 本教程將引導您了解Zeppelin的一些基本概念。我們假設你已經安裝了Zeppelin。如果沒有，請先看[這里](http://zeppelin.apache.org/docs/0.7.1/install/install.html)。 Zeppelin當前的主要后端處理引擎是[Apache Spark](https://spark.apache.org/)。如果您剛剛接觸到該系統，您可能希望首先了解如何處理數據以充分利用Zeppelin。 ## 本地文件教程 ### 數據優化 在開始Zeppelin教程之前，您需要下載[bank.zip](http://archive.ics.uci.edu/ml/machine-learning-databases/00222/bank.zip)。 首先，將csv格式的數據轉換成RDD?`Bank`對象，運行以下腳本。這也將使用`filter`功能刪除標題。 ``` val bankText = sc.textFile("yourPath/bank/bank-full.csv") case class Bank(age:Integer, job:String, marital : String, education : String, balance : Integer) // split each line, filter out header (starts with "age"), and map it into Bank case class val bank = bankText.map(s=>s.split(";")).filter(s=>s(0)!="\"age\"").map( s=>Bank(s(0).toInt, s(1).replaceAll("\"", ""), s(2).replaceAll("\"", ""), s(3).replaceAll("\"", ""), s(5).replaceAll("\"", "").toInt ) ) // convert to DataFrame and create temporal table bank.toDF().registerTempTable("bank") ``` ### 數據檢索 假設我們想看到年齡分布`bank`。為此，運行： ``` %sql select age, count(1) from bank where age < 30 group by age order by age ``` ?您可以輸入框通過更換設置年齡條件`30`用`${maxAge=30}`。 ``` %sql select age, count(1) from bank where age < ${maxAge=30} group by age order by age ``` ?現在我們要看到具有某種婚姻狀況的年齡分布，并添加組合框來選擇婚姻狀況。跑： ``` %sql select age, count(1) from bank where marital="${marital=single,single|divorced|married}" group by age order by age ``` ## 具有流數據的教程 ### 數據優化 由于本教程基于Twitter的示例tweet流，您必須使用Twitter帳戶配置身份驗證。要做到這一點，看看[Twitter Credential Setup](https://databricks-training.s3.amazonaws.com/realtime-processing-with-spark-streaming.html#twitter-credential-setup)。當您得到API密鑰，您應填寫證書相關的值（`apiKey`，`apiSecret`，`accessToken`，`accessTokenSecret`與下面的腳本您的API密鑰）。 這將創建一個`Tweet`對象的RDD?并將這些流數據注冊為一個表： ``` import org.apache.spark.streaming._ import org.apache.spark.streaming.twitter._ import org.apache.spark.storage.StorageLevel import scala.io.Source import scala.collection.mutable.HashMap import java.io.File import org.apache.log4j.Logger import org.apache.log4j.Level import sys.process.stringSeqToProcess /** Configures the Oauth Credentials for accessing Twitter */ def configureTwitterCredentials(apiKey: String, apiSecret: String, accessToken: String, accessTokenSecret: String) { val configs = new HashMap[String, String] ++= Seq( "apiKey" -> apiKey, "apiSecret" -> apiSecret, "accessToken" -> accessToken, "accessTokenSecret" -> accessTokenSecret) println("Configuring Twitter OAuth") configs.foreach{ case(key, value) => if (value.trim.isEmpty) { throw new Exception("Error setting authentication - value for " + key + " not set") } val fullKey = "twitter4j.oauth." + key.replace("api", "consumer") System.setProperty(fullKey, value.trim) println("\tProperty " + fullKey + " set as [" + value.trim + "]") } println() } // Configure Twitter credentials val apiKey = "xxxxxxxxxxxxxxxxxxxxxxxxx" val apiSecret = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" val accessToken = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" val accessTokenSecret = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" configureTwitterCredentials(apiKey, apiSecret, accessToken, accessTokenSecret) import org.apache.spark.streaming.twitter._ val ssc = new StreamingContext(sc, Seconds(2)) val tweets = TwitterUtils.createStream(ssc, None) val twt = tweets.window(Seconds(60)) case class Tweet(createdAt:Long, text:String) twt.map(status=> Tweet(status.getCreatedAt().getTime()/1000, status.getText()) ).foreachRDD(rdd=> // Below line works only in spark 1.3.0. // For spark 1.1.x and spark 1.2.x, // use rdd.registerTempTable("tweets") instead. rdd.toDF().registerAsTable("tweets") ) twt.print ssc.start()? ``` ### 數據檢索 對于每個以下腳本，每次單擊運行按鈕，您將看到不同的結果，因為它是基于實時數據。 我們開始提取包含單詞**girl的**最多10個tweets?。 ``` %sql select * from tweets where text like '%girl%' limit 10 ``` ?這次假設我們想看看在過去60秒內每秒創建的tweet有多少。為此，運行： ``` %sql select createdAt, count(1) from tweets group by createdAt order by createdAt ``` ?您可以在Spark SQL中進行用戶定義的功能并使用它們。讓我們通過命名函數來嘗試`sentiment`。該功能將返回參數中的三種態度之一（正，負，中性）。 ``` def sentiment(s:String) : String = { val positive = Array("like", "love", "good", "great", "happy", "cool", "the", "one", "that") val negative = Array("hate", "bad", "stupid", "is") var st = 0; val words = s.split(" ") positive.foreach(p => words.foreach(w => if(p==w) st = st+1 ) ) negative.foreach(p=> words.foreach(w=> if(p==w) st = st-1 ) ) if(st>0) "positivie" else if(st<0) "negative" else "neutral" } // Below line works only in spark 1.3.0. // For spark 1.1.x and spark 1.2.x, // use sqlc.registerFunction("sentiment", sentiment _) instead. sqlc.udf.register("sentiment", sentiment _) ``` ?要檢查人們如何看待使用`sentiment`上述功能的女孩，請運行以下操作： ``` %sql select sentiment(text), count(1) from tweets where text like '%girl%' group by sentiment(text) ```