Hadoop

Spark

<?xml version="1.0" encoding="UTF-8"?>

  <project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

  

    <groupId>cn.itcast.spark</groupId>

    <artifactId>spark-mvn</artifactId>

    <version>1.0-SNAPSHOT</version>

  

    <properties>

        <maven.compiler.source>1.7</maven.compiler.source>

        <maven.compiler.target>1.7</maven.compiler.target>

        <encoding>UTF-8</encoding>

        <scala.version>2.10.6</scala.version>

        <scala.compat.version>2.10</scala.compat.version>

    </properties>

  

    <dependencies>

        <dependency>

            <groupId>org.scala-lang</groupId>

            <artifactId>scala-library</artifactId>

            <version>${scala.version}</version>

        </dependency>

  

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-core_2.10</artifactId>

            <version>1.5.2</version>

        </dependency>

  

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-streaming_2.10</artifactId>

            <version>1.5.2</version>

        </dependency>

  

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-client</artifactId>

            <version>2.6.2</version>

        </dependency>

    </dependencies>

  

    <build>

        <sourceDirectory>src/main/scala</sourceDirectory>

        <testSourceDirectory>src/test/scala</testSourceDirectory>

        <plugins>

            <plugin>

                <groupId>net.alchim31.maven</groupId>

                <artifactId>scala-maven-plugin</artifactId>

                <version>3.2.0</version>

                <executions>

                    <execution>

                        <goals>

                            <goal>compile</goal>

                            <goal>testCompile</goal>

                        </goals>

                        <configuration>

                            <args>

                                <arg>-make:transitive</arg>

                                <arg>-dependencyfile</arg>

                                <arg>${project.build.directory}/.scala_dependencies</arg>

                            </args>

                        </configuration>

                    </execution>

                </executions>

            </plugin>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-surefire-plugin</artifactId>

                <version>2.18.1</version>

                <configuration>

                    <useFile>false</useFile>

                    <disableXmlReport>true</disableXmlReport>

                    <includes>

                        <include>**/*Test.*</include>

                        <include>**/*Suite.*</include>

                    </includes>

                </configuration>

            </plugin>

  

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-shade-plugin</artifactId>

                <version>2.3</version>

                <executions>

                    <execution>

                        <phase>package</phase>

                        <goals>

                            <goal>shade</goal>

                        </goals>

                        <configuration>

                            <filters>

                                <filter>

                                    <artifact>*:*</artifact>

                                    <excludes>

                                        <exclude>META-INF/*.SF</exclude>

                                        <exclude>META-INF/*.DSA</exclude>

                                        <exclude>META-INF/*.RSA</exclude>

                                    </excludes>

                                </filter>

                            </filters>

                            <transformers>

                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">

                                    <mainClass>cn.itcast.spark.WordCount</mainClass>

                                </transformer>

                            </transformers>

                        </configuration>

                    </execution>

                </executions>

            </plugin>

        </plugins>

    </build>

  </project>

package cn.itcast.spark

  

  import org.apache.spark.{SparkContext, SparkConf}

  

  object WordCount {

  def main(args: Array[String]) {

    //创建SparkConf()并设置App名称

    val conf = new SparkConf().setAppName("WC")

    //创建SparkContext，该对象是提交spark App的入口

    val sc = new SparkContext(conf)

    //使用sc创建RDD并执行相应的transformation和action

    sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_, 1).sortBy(_._2, false).saveAsTextFile(args(1))

    //停止sc，结束该任务

    sc.stop()

  }

}

转换

含义

map(func)

返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成

filter(func)

返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成

flatMap(func)

类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）

mapPartitions(func)

类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]

mapPartitionsWithIndex(func)

类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是

(Int, Interator[T]) => Iterator[U]

sample(withReplacement, fraction, seed)

根据fraction指定的比例对数据进行采样，可以选择是否使用随机数进行替换，seed用于指定随机数生成器种子

union(otherDataset)

对源RDD和参数RDD求并集后返回一个新的RDD

intersection(otherDataset)

对源RDD和参数RDD求交集后返回一个新的RDD

distinct([numTasks]))

对源RDD进行去重后返回一个新的RDD

groupByKey([numTasks]) 

在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD

reduceByKey(func, [numTasks])

在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置

aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

sortByKey([ascending], [numTasks])

在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD

sortBy(func,[ascending], [numTasks])

与sortByKey类似，但是更灵活

join(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

cogroup(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

cartesian(otherDataset)

笛卡尔积

pipe(command, [envVars])

coalesce(numPartitions)  

repartition(numPartitions)

repartitionAndSortWithinPartitions(partitioner)

动作

含义

reduce(func)

通过func函数聚集RDD中的所有元素，这个功能必须是课交换且可并联的

collect()

在驱动程序中，以数组的形式返回数据集的所有元素

count()

返回RDD的元素个数

first()

返回RDD的第一个元素（类似于take(1)）

take(n)

返回一个由数据集的前n个元素组成的数组

takeSample(withReplacement,num, [seed])

返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子

takeOrdered(n, [ordering])

saveAsTextFile(path)

将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本

saveAsSequenceFile(path) 

将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。

saveAsObjectFile(path) 

countByKey()

针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。

foreach(func)

在数据集的每一个元素上，运行函数func进行更新。

<dependency>

    <groupId>org.apache.spark</groupId>

    <artifactId>spark-sql_2.10</artifactId>

    <version>1.5.2</version>

  </dependency>

package cn.itcast.spark.sql

  

  import org.apache.spark.{SparkConf, SparkContext}

  import org.apache.spark.sql.SQLContext

  

  object InferringSchema {

  def main(args: Array[String]) {

  

    //创建SparkConf()并设置App名称

    val conf = new SparkConf().setAppName("SQL-1")

    //SQLContext要依赖SparkContext

    val sc = new SparkContext(conf)

    //创建SQLContext

    val sqlContext = new SQLContext(sc)

  

    //从指定的地址创建RDD

    val lineRDD = sc.textFile(args(0)).map(_.split(" "))

  

    //创建case class

    //将RDD和case class关联

    val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))

    //导入隐式转换，如果不到人无法将RDD转换成DataFrame

    //将RDD转换成DataFrame

    import sqlContext.implicits._

    val personDF = personRDD.toDF

    //注册表

    personDF.registerTempTable("t_person")

    //传入SQL

    val df = sqlContext.sql("select * from t_person order by age desc limit 2")

    //将结果以JSON的方式存储到指定位置

    df.write.json(args(1))

    //停止Spark Context

    sc.stop()

  }

}

  //case class一定要放到外面

  case class Person(id: Int, name: String, age: Int)

package cn.itcast.spark.sql

  

  import org.apache.spark.sql.{Row, SQLContext}

  import org.apache.spark.sql.types._

  import org.apache.spark.{SparkContext, SparkConf}

  

  /**

  * Created by ZX on 2015/12/11.

  */

  object SpecifyingSchema {

  def main(args: Array[String]) {

    //创建SparkConf()并设置App名称

    val conf = new SparkConf().setAppName("SQL-2")

    //SQLContext要依赖SparkContext

    val sc = new SparkContext(conf)

    //创建SQLContext

    val sqlContext = new SQLContext(sc)

    //从指定的地址创建RDD

    val personRDD = sc.textFile(args(0)).map(_.split(" "))

    //通过StructType直接指定每个字段的schema

    val schema = StructType(

      List(

        StructField("id", IntegerType, true),

        StructField("name", StringType, true),

        StructField("age", IntegerType, true)

      )

    )

    //将RDD映射到rowRDD

    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))

    //将schema信息应用到rowRDD上

    val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)

    //注册表

    personDataFrame.registerTempTable("t_person")

    //执行SQL

    val df = sqlContext.sql("select * from t_person order by age desc limit 4")

    //将结果以JSON的方式存储到指定位置

    df.write.json(args(1))

    //停止Spark Context

    sc.stop()

  }

}

package cn.itcast.spark.sql

  import java.util.Properties

  import org.apache.spark.sql.{SQLContext, Row}

  import org.apache.spark.sql.types.{StringType, IntegerType, StructField, StructType}

  import org.apache.spark.{SparkConf, SparkContext}

  

  object JdbcRDD {

  def main(args: Array[String]) {

    val conf = new SparkConf().setAppName("MySQL-Demo")

    val sc = new SparkContext(conf)

    val sqlContext = new SQLContext(sc)

    //通过并行化创建RDD

    val personRDD = sc.parallelize(Array("1 tom 5", "2 jerry 3", "3 kitty 6")).map(_.split(" "))

    //通过StructType直接指定每个字段的schema

    val schema = StructType(

      List(

        StructField("id", IntegerType, true),

        StructField("name", StringType, true),

        StructField("age", IntegerType, true)

      )

    )

    //将RDD映射到rowRDD

    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))

    //将schema信息应用到rowRDD上

    val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)

    //创建Properties存储数据库相关属性

    val prop = new Properties()

    prop.put("user", "root")

    prop.put("password", "123456")

    //将数据追加到数据库

    personDataFrame.write.mode("append").jdbc("jdbc:mysql://192.168.10.1:3306/bigdata", "bigdata.person", prop)

    //停止SparkContext

    sc.stop()

  }

}

Spark

Storm

开发语言：Scala

开发语言：Clojure

编程模型：DStream

编程模型：Spout/Bolt

Transformation

Meaning

map(func)

Return a new DStream by passing each element of the source DStream through a function func.

flatMap(func)

Similar to map, but each input item can be mapped to 0 or more output items.

filter(func)

Return a new DStream by selecting only the records of the source DStream on which func returns true.

repartition(numPartitions)

Changes the level of parallelism in this DStream by creating more or fewer partitions.

union(otherStream)

Return a new DStream that contains the union of the elements in the source DStream and otherDStream.

count()

Return a new DStream of single-element RDDs by counting the number of elements in each RDD of the source DStream.

reduce(func)

Return a new DStream of single-element RDDs by aggregating the elements in each RDD of the source DStream using a function func (which takes two arguments and returns one). The function should be associative so that it can be computed in parallel.

countByValue()

When called on a DStream of elements of type K, return a new DStream of (K, Long) pairs where the value of each key is its frequency in each RDD of the source DStream.

reduceByKey(func, [numTasks])   

When called on a DStream of (K, V) pairs, return a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. Note: By default, this uses Spark’s default number of parallel tasks (2 for local mode, and in cluster mode the number is determined by the config property spark.default.parallelism) to do the grouping. You can pass an optional numTasks argument to set a different number of tasks.

join(otherStream, [numTasks])

When called on two DStreams of (K, V) and (K, W) pairs, return a new DStream of (K, (V, W)) pairs with all pairs of elements for each key.

cogroup(otherStream, [numTasks])

When called on a DStream of (K, V) and (K, W) pairs, return a new DStream of (K, Seq[V], Seq[W]) tuples.

transform(func)     

Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream. This can be used to do arbitrary RDD operations on the DStream.

updateStateByKey(func)

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 for the key. This can be used to maintain arbitrary state data for each key.

Output Operation

Meaning

print()

Prints the first ten elements of every batch of data in a DStream on the driver node running the streaming application. This is useful for development and debugging.

saveAsTextFiles(prefix, [suffix])

Save this DStream’s contents as text files. The file name at each batch interval is generated based on prefix and suffix: “prefix-TIME_IN_MS[.suffix]”.

saveAsObjectFiles(prefix, [suffix])

Save this DStream’s contents as SequenceFiles of serialized Java objects. The file name at each batch interval is generated based on prefix and suffix: “prefix-TIME_IN_MS[.suffix]”.

saveAsHadoopFiles(prefix, [suffix])

Save this DStream’s contents as Hadoop files. The file name at each batch interval is generated based on prefix and suffix: “prefix-TIME_IN_MS[.suffix]”.

foreachRDD(func)

The most generic output operator that applies a function, func, to each RDD generated from the stream. This function should push the data in each RDD to an external system, such as saving the RDD to files, or writing it over the network to a database. Note that the function func is executed in the driver process running the streaming application, and will usually have RDD actions in it that will force the computation of the streaming RDDs.

package cn.itcast.spark.streaming

  

  import cn.itcast.spark.util.LoggerLevel

  import org.apache.spark.SparkConf

  import org.apache.spark.streaming.{Seconds, StreamingContext}

  

  object NetworkWordCount {

  def main(args: Array[String]) {

    //设置日志级别

    LoggerLevel.setStreamingLogLevels()

    //创建SparkConf并设置为本地模式运行

    //注意local[2]代表开两个线程

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")

    //设置DStream批次时间间隔为2秒

    val ssc = new StreamingContext(conf, Seconds(2))

    //通过网络读取数据

    val lines = ssc.socketTextStream("192.168.10.101", 9999)

    //将读到的数据用空格切成单词

    val words = lines.flatMap(_.split(" "))

    //将单词和1组成一个pair

    val pairs = words.map(word => (word, 1))

    //按单词进行分组求相同单词出现的次数

    val wordCounts = pairs.reduceByKey(_ + _)

    //打印结果到控制台

    wordCounts.print()

    //开始计算

    ssc.start()

    //等待停止

    ssc.awaitTermination()

  }

}

package cn.itcast.spark.streaming

  

  import cn.itcast.spark.util.LoggerLevel

  import org.apache.spark.{HashPartitioner, SparkConf}

  import org.apache.spark.streaming.{StreamingContext, Seconds}

  

  object NetworkUpdateStateWordCount {

  /**

    * String : 单词 hello

    * Seq[Int] ：单词在当前批次出现的次数

    * Option[Int] ： 历史结果

    */

  val updateFunc = (iter: Iterator[(String, Seq[Int], Option[Int])]) => {

    //iter.flatMap(it=>Some(it._2.sum + it._3.getOrElse(0)).map(x=>(it._1,x)))

    iter.flatMap{case(x,y,z)=>Some(y.sum + z.getOrElse(0)).map(m=>(x, m))}

  }

  

  def main(args: Array[String]) {

    LoggerLevel.setStreamingLogLevels()

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkUpdateStateWordCount")

    val ssc = new StreamingContext(conf, Seconds(5))

    //做checkpoint 写入共享存储中

    ssc.checkpoint("c://aaa")

    val lines = ssc.socketTextStream("192.168.10.100", 9999)

    //reduceByKey 结果不累加

    //val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

    //updateStateByKey结果可以累加但是需要传入一个自定义的累加函数：updateFunc

    val results = lines.flatMap(_.split(" ")).map((_,1)).updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)

    results.print()

    ssc.start()

    ssc.awaitTermination()

  }

}

package cn.itcast.spark.streaming

  

  package cn.itcast.spark

  

  import org.apache.spark.{HashPartitioner, SparkConf}

  import org.apache.spark.storage.StorageLevel

  import org.apache.spark.streaming.kafka.KafkaUtils

  import org.apache.spark.streaming.{Seconds, StreamingContext}

  

  object UrlCount {

  val updateFunc = (iterator: Iterator[(String, Seq[Int], Option[Int])]) => {

    iterator.flatMap{case(x,y,z)=> Some(y.sum + z.getOrElse(0)).map(n=>(x, n))}

  }

  

  def main(args: Array[String]) {

    //接收命令行中的参数

    val Array(zkQuorum, groupId, topics, numThreads, hdfs) = args

    //创建SparkConf并设置AppName

    val conf = new SparkConf().setAppName("UrlCount")

    //创建StreamingContext

    val ssc = new StreamingContext(conf, Seconds(2))

    //设置检查点

    ssc.checkpoint(hdfs)

    //设置topic信息

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

    //重Kafka中拉取数据创建DStream

    val lines = KafkaUtils.createStream(ssc, zkQuorum ,groupId, topicMap, StorageLevel.MEMORY_AND_DISK).map(_._2)

    //切分数据，截取用户点击的url

    val urls = lines.map(x=>(x.split(" ")(6), 1))

    //统计URL点击量

    val result = urls.updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)

    //将结果打印到控制台

    result.print()

    ssc.start()

    ssc.awaitTermination()

  }

}