val numbers = Array(, , , , )
val numberRDD = sc.parallelize(numbers, )
val multipleNumberRDD = numberRDD.map ( num => num * )
multipleNumberRDD.foreach ( num => println(num) )
val lineArray = Array("hello you", "hello me", "hello world")
val lines = sc.parallelize(lineArray, )
val words = lines.flatMap { line => line.split(" ") }
words.foreach { word => println(word) }
val factor =
val factorBroadcast = sc.broadcast(factor)
val arr =Array(,,,,)
val rdd = sc.parallelize(arr)
val mutiRdd = rdd.map(num => num*factorBroadcast.value())
mutiRdd.foreach(num => println(num))
Spark 作为一个优秀的大数据计算框架，必然少不了支持SQL的框架，基于Hive的性能以及它与Spark的兼容，Shark项目由此诞生，Shark即Hive on Spark，它通过HQL解析，将HQL翻译成Spark上的RDD操作，然后通过Hive的metadata获取数据库里的表信息，实际的HDFS上的数据和文件会有Spark获取并放到Spark上计算。
val sortedByNameEmployees = sqlContext.sql("select * from employeeTable order by name desc")
sortedByNameEmployees.show()
+-----+-------+
|empId| name|
+-----+-------+
| |iteblog|
| | Jason|
| | Abhi|
+-----+-------+
The second method for creating DataFrames is through a programmatic interface that allows you to construct a schema and then apply it to an existing RDD.
a DataFrame can be created programmatically with three steps:
第一步：将RDD（可以是任何类型）转换为RDD[Row]
需要引包：
import org.apache.spark.sql._
演示：
val rdd = sc.textFile("/user/vin/people.txt")
import org.apache.spark.sql._
val rowRdd = rdd.map(line => line.split(", ")).map(x => Row(x(0), x(1).toInt))
Spark基础全解析SPARK全解析
第二步：
创建schema
引包：
import org.apache.spark.sql.types._
val schema = StructType(StructField("name",StringType,true) :: StructField("age",IntegerType,true) :: Nil)
Spark基础全解析SPARK全解析
第三步：Apply the schema to the RDD of Rows
val people_df = sqlContext.createDataFrame(rowRdd, schema)
/**
* 从字面看，是合并
* Merges two aggregation buffers
* and stores the updated buffer values back to `buffer1`.
* * This is called when we merge two partially aggregated data together.
* * @param buffer1
* @param buffer2
*/
override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
// 获取缓冲数据
val salTotal1 = buffer1.getDouble()
val salCount1 = buffer1.getInt()
// 获取缓冲数据
val salTotal2 = buffer2.getDouble()
val salCount2 = buffer2.getInt()
// 更新缓冲数据
buffer1.update(, salTotal1 + salTotal2)
buffer1.update(, salCount1 + salCount2)
}
val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds()) //设置批次时间,测试可以用5秒
//在Spark-shell中可以通过传递sc方式创建StreamingContext：
val ssc = new StreamingContext(sc, Seconds())
/**
- Create a input stream from TCP source hostname:port. Data is received using
- a TCP socket and the receive bytes is interpreted as UTF8 encoded `\n` delimited lines.
- @param hostname Hostname to connect to for receiving data
- @param port Port to connect to for receiving data
- @param storageLevel Storage level to use for storing the received objects
- (default: StorageLevel.MEMORY_AND_DISK_SER_2)
*/
def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String] = withNamedScope("socket text stream") {
socketStream[String](hostname, port, SocketReceiver.bytesToLines, storageLevel)
}
val socketDStream= scc.socketTextStream(“vin01”,9999)
这里的socketDStream是一行一行的数据
- 第二步：基于DStream进行处理数据
// 将行数据分隔成单词
val words = socketDStream.flatMap(_.split(" "))
// 计算这一批次的词频，先将单词转换成元组，再reduce
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)
第三步：输出
wordCounts.print()
最后启动该应用即可：
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate
运行测试：
Spark基础全解析SPARK全解析
输出：
Spark基础全解析SPARK全解析
Spark Streaming 读取HDFS数据
读取HDFS上的数据，需要用到Spark Streaming类中的一个方法：
即：
/**
* Create a input stream that monitors a Hadoop-compatible filesystem
* for new files and reads them as text files (using key as LongWritable, value
* as Text and input format as TextInputFormat). Files must be written to the
* monitored directory by "moving" them from another location within the same
* file system. File names starting with . are ignored.
* @param directory HDFS directory to monitor for new file
*/
def textFileStream(directory: String): DStream[String] = withNamedScope("text file stream") {
fileStream[LongWritable, Text, TextInputFormat](directory).map(_._2.toString)
}
同样，使用wordcount模板，更改的是创建DStream的方式
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
val ssc = new StreamingContext(sc, Seconds())
val socketDStream = ssc.textFileStream("/user/vin/sparkstreaming/hdfsfiles")
val words = socketDStream.flatMap(_.split(" "))
val pairs = words.map(word => (word, ))
val wordCounts = pairs.reduceByKey(_ + _)
wordCounts.print()
ssc.start()
ssc.awaitTermination()
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.flume._
val ssc = new StreamingContext(sc, Seconds())
// Step ：Recevier Data From Where
// Flume: FlumeUtils, Kafka: KafkaUtils
val flumeDStream = FlumeUtils.createStream(ssc, "vin01", ).map(event => new String(event.event.getBody.array()))
// Step : Process Data Base DStream
// DStream[Long]
val words = flumeDStream.flatMap(_.split(" "))
// Count each word in each batch
val pairs = words.map(word => (word, ))
val wordCounts = pairs.reduceByKey(_ + _)
// Step : Output Result
// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate
sc.stop
import kafka.serializer.StringDecoder
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.{SparkConf, SparkContext}
val ssc = new StreamingContext(sc, Seconds(5))
// set checkpoint
ssc.checkpoint("sparkstreaming/kafka/")
// Step ： Recevier Data From Where
val kafkaParams = Map("metadata.broker.list" -> "vin01:9092")
val topics = Set("sparkPullTopic")
// InputDStream[(K, V)]
val socketDStream = KafkaUtils.createDirectStream[
String, String,StringDecoder, StringDecoder](
ssc,
kafkaParams, // Map[String, String]
topics // Set[String]
).map(_._2)
// Step : Process Data Base DStream
// Split each line into words
val words = socketDStream.flatMap(_.split(" "))
// Count each word in each batch
val pairs = words.map(word => (word, ))
/**
* updateFunc: (Seq[V], Option[S]) => Option[S]
*
* DStream[(Key, Value)]
* Seq[V]
* V: 代表的是 DStream中Value的类型,针对WordCount程序来说，V是Int
* Option[S]
* Option\Some\None
* S: 代表的是状态State，存储的是以前分析的结果，针对WordCount程序来说，S是Count，Int
* S可以是任意类型，依据实际需求而定
*
*/
val wordCounts = pairs.updateStateByKey(
(values: Seq[Int], state: Option[Int]) => {
//获取当前的要计算的值
val currentCount = values.sum
// 获取以前状态中的值
val previousCount = state.getOrElse()
// update state and return
Some(currentCount + previousCount)
}
)
// Step : Output Result
// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate
sc.stop
/**
* Return a new "state" DStream where the state for each key is updated by applying
* the given function on the previous state of the key and the new values of each key.
* Hash partitioning is used to generate the RDDs with Spark's default number of partitions.
* @param updateFunc State update function. If `this` function returns None, then
* corresponding state key-value pair will be eliminated.
* @tparam S State type
*/
def updateStateByKey[S: ClassTag](
updateFunc: (Seq[V], Option[S]) => Option[S]
): DStream[(K, S)] = ssc.withScope {
updateStateByKey(updateFunc, defaultPartitioner())
}
val WordCounts = Pairs.updataStateByKey((values:Seq[Int], state:Option[Int]) => {
//获取当前要计算的值
val currentCount = values.sum
//获取以前状态中的值
val previousCount = state.getOrElse(0)
//更新状态，返回
Some(currentCount + prevousCount)
})
完整代码：
import org.apache.spark.streaming._
import org.apache.spark.{SparkConf, SparkContext}
object UpdateStateWordCount {
def main(args: Array[String]) {
// step 0: SparkContext
val sparkConf = new SparkConf()
.setAppName("LogAnalyzer Applicaiton") // name
.setMaster("local[2]") // --master local[2] | spark://xx:7077 | yarn
// Create SparkContext
val sc = new SparkContext(sparkConf)
val ssc = new StreamingContext(sc, Seconds())
// set checkpoint
ssc.checkpoint("sparkstreaming/socketwc/")
// Step 1： Recevier Data From Where
val socketDStream = ssc.socketTextStream("hadoop-senior01.ibeifeng.com", )
// transformFunc: RDD[T] => RDD[U]
// 仅仅针对DStream中的RDD来操作，在实际开发中，RDD的操作更加方便
/**
* WordCount
* . ? @ # $ !
* 如果单词是上述标点符号，统计毫无意义，可以进行过滤
*/
val filterRdd = sc.parallelize(List(".", "?", "@", "#", "!")).map((_, true))
socketDStream.transform(rdd => {
val tupleRdd = rdd.map((_, ))
// join,filter
tupleRdd.leftOuterJoin(filterRdd)
})
// Step 2: Process Data Base DStream
// Split each line into words
val words = socketDStream.flatMap(_.split(" "))
// Count each word in each batch
val pairs = words.map(word => (word, ))
/**
* updateFunc: (Seq[V], Option[S]) => Option[S]
*
* DStream[(Key, Value)]
* Seq[V]
* V: 代表的是 DStream中Value的类型,针对WordCount程序来说，V是Int
* Option[S]
* Option\Some\None
* S: 代表的是状态State，存储的是以前分析的结果，针对WordCount程序来说，S是Count，Int
* S可以是任意类型，依据实际需求而定
*
* 回顾一下：
* reduceByKey()
* =
* reduce(Key, Values)
*
*/
val wordCounts = pairs.updateStateByKey(
(values: Seq[Int], state: Option[Int]) => {
//获取当前的要计算的值
val currentCount = values.sum
// 获取以前状态中的值
val previousCount = state.getOrElse()
// update state and return
Some(currentCount + previousCount)
}
)
val wcDStream = pairs.updateStateByKey(
(values: Seq[Int], state: Option[Int]) => Some(values.sum + state.getOrElse()))
// Step 3: Output Result
// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate
sc.stop
}
}
测试：
打开nc 9999端口，运行程序：
两次在端口输入：
Spark基础全解析SPARK全解析
第一次输出：
Spark基础全解析SPARK全解析
第二次输出：
Spark基础全解析SPARK全解析
实现了累加统计，测试成功
注：在代码中，使用了transform方法，对其进行解析：
DStream的transform方法
transform方法在DStream中，其源码为：
/**
* Return a new DStream in which each RDD is generated by applying a function
* on each RDD of 'this' DStream.
*/
def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U] = ssc.withScope {
// because the DStream is reachable from the outer object here, and because
// DStreams can't be serialized with closures, we can't proactively check
// it for serializability and so we pass the optional false to SparkContext.clean
val cleanedF = context.sparkContext.clean(transformFunc, false)
transform((r: RDD[T], t: Time) => cleanedF(r))
}
// foreachFunc: (RDD[T], Time) => Unit
dstream.foreachRDD(rdd => {
// 将分析的数据存储到JDBC中，MySQL数据中
val connection = createJDBCConnection() // executed at the driver
rdd.foreach { record =>
connection.putStateResult(record) // executed at the worker
}
})
/**
* Return a new DStream by applying `reduceByKey` over a sliding window. This is similar to
* `DStream.reduceByKey()` but applies it over a sliding window. Hash partitioning is used to
* generate the RDDs with Spark's default number of partitions.
* @param reduceFunc associative reduce function
* @param windowDuration width of the window; must be a multiple of this DStream's
* batching interval
* @param slideDuration sliding interval of the window (i.e., the interval after which
* the new DStream will generate RDDs); must be a multiple of this
* DStream's batching interval
*/
def reduceByKeyAndWindow(
reduceFunc: (V, V) => V,
windowDuration: Duration,
slideDuration: Duration
): DStream[(K, V)] = ssc.withScope {
reduceByKeyAndWindow(reduceFunc, windowDuration, slideDuration, defaultPartitioner())
}
