Ó¢ÐÛϧӢÐÛ-µ±SparkÓöÉÏZeppelin֮ʵս°¸Àý

cd $ZEPPELIN_HOME/scripts/docker/spark-cluster-managers/spark_standalonedocker build -t "spark_standalone" .

docker run -it \-p 8080:8080 \-p 7077:7077 \-p 8888:8888 \-p 8081:8081 \-h sparkmaster \--name spark_standalone \spark_standalone bash;

ps -ef | grep spark

age:Integer, job:String, marital : String, education : String, balance : Integer20;teacher;single;±¾¿Æ;2000025;plumber;single;±¾¿Æ;1000021;doctor;single;±¾¿Æ;2500023;singer;single;±¾¿Æ;20000...

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 classval 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 tablebank.toDF().registerTempTable("bank")

println(¡°%table column_1\tcolumn_2\n¡±+value_1\tvalue_2\n+¡­)

%sparkdf3.write.mode("overwrite").format("jdbc").option("driver","com.mysql.jdbc.Driver").option("user","root").option("password","password").option("url","jdbc:mysql://localhost:3306/spark_demo").option("dbtable","record").save()

import org.apache.spark.streaming._import org.apache.spark.streaming.twitter._import org.apache.spark.storage.StorageLevelimport scala.io.Sourceimport scala.collection.mutable.HashMapimport java.io.Fileimport org.apache.log4j.Loggerimport org.apache.log4j.Levelimport 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 credentialsval 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()

%sparkimport _root_.kafka.serializer.DefaultDecoderimport _root_.kafka.serializer.StringDecoderimport org.apache.spark.streaming.kafka.KafkaUtilsimport org.apache.spark.storage.StorageLevelimport org.apache.spark.streaming._ // prevent INFO logging from pollution outputsc.setLogLevel("INFO") // creating the StreamingContext with 5 seconds intervalval ssc = new StreamingContext(sc, Seconds(5)) val kafkaConf = Map(    "metadata.broker.list" -> "localhost:9092",    "zookeeper.connect" -> "localhost:2181",    "group.id" -> "kafka-streaming-example",    "zookeeper.connection.timeout.ms" -> "1000") val lines = KafkaUtils.createStream[Array[Byte], String, DefaultDecoder, StringDecoder](    ssc,    kafkaConf,    Map("test" -> 1),   // subscripe to topic and partition 1    StorageLevel.MEMORY_ONLY) val words = lines.flatMap{ case(x, y) => y.split(" ")}?import spark.implicits._?val w=words.map(x=> (x,1L)).reduceByKey(_+_)w.foreachRDD(rdd => rdd.toDF.registerTempTable("counts"))ssc.start()

select * from counts order by _2 desc limit 10