Spark之RDD实战篇3

1
2
3

val rdd =sc.parallelize(Array(("panda",0),("pink",3),("pirate",3),("panda",1),("pink",4)),2)
val result =rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
result.collect

1
2
3
4
5
6
7

val rdd =sc.parallelize(Array(("panda",0),("pink",3),("pirate",3),("panda",1),("pink",4)),2)
val result = rdd.combineByKey(
  (v) => (v, 1),
  (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
  (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
)
result.collect

1
2
3
4
5
6
7
8
9
10
11
12

var rddl = sc.makeRDD(Array(("A",0), ("A",2), ("B",1), ("B",2), ("Cn",1)))
rdd1.groupByKey().collect
//使用reduceByKey操作将RDD[K,V]中每个K对应的V值根据映射函数来运算                 
var rdd2 = rdd1.reduceByKey((x,y) => x + y)
//对rddl使用reduceByKey操作进行重新分区
var rdd2 = rdd1.reduceByKey (new org.apache.spark.HashPartitioner(2),(x, y) => x + y)
rdd2.collect

var rdd2 = sc.makeRDD(Array(("A","a"),("C","c"),("D","d")),2)
var rdd3 = sc.makeRDD(Array(("A","A"),("E","E")),2)
var rdd4 = rddl.cogroup(rdd2,rdd3)
rdd4.collect

1
2
3
4
5
6
7
8
9

var rdd1 = sc.makeRDD (Array(("A", "1"), ("B","2") , ("C","3")), 2)
var rdd2 = sc.makeRDD (Array(("A", "a"), ("C","c") , ("D","d")), 2)
//进行内连接操作
rdd1.join(rdd2).collect
//进行左连接操作
rdd1.leftOuterJoin(rdd2).collect
rdd1.leftOuterJoin(rdd2).collect                      
//进行右连接操作
rdd1.rightOuterJoin(rdd2).collect

1
2
3

··val rdd = sc.parallelize(Array((3,"hadoop"),(6,"hohblog"),(2,"flink"),(1,"spark")))
rdd.sortByKey(true).collect()
rdd.sortByKey(false).collect()

1
2
3
4

val pairs =sc.parallelize(List((1,1),(2,2),(3,3)))
pairs.partitioner
val partitioned = pairs.partitionBy(new org.apache.spark.HashPartitioner(2))
partitioned.partitioner

1
2
3
4
5
6
7

val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)
nopar.partitioner
nopar.mapPartitionsWithIndex((index,iter)=>{ Iterator(index.toString+" : "+iter.mkString("|")) }).collect
val hashpar = nopar.partitionBy(new org.apache.spark.HashPartitioner(7))
hashpar.count
hashpar.partitioner
hashpar.mapPartitions(iter => Iterator(iter.length)).collect()

1
2
3
4
5
6
7

val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)
nopar.partitioner
nopar.mapPartitionsWithIndex((index,iter)=>{ Iterator(index.toString+" : "+iter.mkString("|")) }).collect
val rangepar = nopar.partitionBy(new org.apache.spark.RangePartitioner(2,nopar))
rangepar.count
rangepar.partitioner
rangepar.mapPartitions(iter => Iterator(iter.length)).collect()

1
2
3

val data=sc.parallelize(List("aa.2","bb.2","cc.3","dd.3","ee.5").zipWithIndex,2)
data.collect
data.mapPartitionsWithIndex((index,iter)=>Iterator(index.toString +" : "+ iter.mkString("|"))).collect

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import org.apache.spark.{Partitioner, SparkConf, SparkContext}
class CustomerPartitioner(numParts:Int) extends org.apache.spark.Partitioner{

  //覆盖分区数
  override def numPartitions: Int = numParts

  //覆盖分区号获取函数
  override def getPartition(key: Any): Int = {
    val ckey: String = key.toString
    ckey.substring(ckey.length-1).toInt%numParts
  }
}

val result = data.partitionBy(new CustomerPartitioner(4))
result.mapPartitionsWithIndex((index,iter)=>Iterator(index.toString +" : "+ iter.mkString("|"))).collect

1
2
3
4

val rdd = sc.textFile("/input/test.txt")
rdd.collect
val rdd = sc.textFile("/input/*")
rdd.collect

1
2
3

{"name":"Hadoop","age":13}
{"name":"Spark", "age":11}
{"name":"Flink", "age":3}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

package com.hph

import org.apache.spark.{SparkConf, SparkContext}
import org.json4s.ShortTypeHints
import org.json4s.jackson.JsonMethods._
import org.json4s.jackson.Serialization

object TestJson {

  case class BigData(name:String,year:Int)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("JSON")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")
    implicit val formats = Serialization.formats(ShortTypeHints(List()))
    val input = sc.textFile("D:\\input\\people.json")

    input.collect().foreach(x => {var c = parse(x).extract[BigData];println(c.name + "," + c.year)})

  }

}

1
2
3
4

val data=sc.parallelize(List((1,"hphblog"),(2,"Spark"),(3,"Flink"),(4,"SpringBoot"),(5,"SpringCloud")))
data.saveAsObjectFile("hdfs://datanode1:9000/objfile")
val objrdd:org.apache.spark.rdd.RDD[(Int,String)] = sc.objectFile[(Int,String)]("hdfs://datanode1:9000/objfile/p*")
objrdd.collect()

1
2

val data = sc.parallelize(Array((1,"Hadoop"), (2,"Spark"), (3,"Flink")))
data.saveAsHadoopFile("hdfs://datanode1:9000/output/hdfs_spark",classOf[Text],classOf[IntWritable],classOf[TextOutputFormat[Text,IntWritable]])

1
2

val data = sc.parallelize(Array(("Hadoop",1), ("Spark",2), ("Flink",3)))
data.saveAsNewAPIHadoopFile("hdfs://datanod1:9000/output/NewAPI/",classOf[Text],classOf[IntWritable] , classOf[org.apache.hadoop.mapreduce.OutputFormat[Text,IntWritable]])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

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

/**
  * Created by 清风笑丶 Cotter on 2019/5/30.
  */
object JDBCRdd {
  def main (args: Array[String] ) {
    val sparkConf = new SparkConf ().setMaster ("local[*]").setAppName ("JdbcApp")
    val sc = new SparkContext (sparkConf)
    val rdd = new org.apache.spark.rdd.JdbcRDD (
      sc,
      () => {
        Class.forName ("com.mysql.jdbc.Driver").newInstance()
        java.sql.DriverManager.getConnection ("jdbc:mysql://datanode1:3306/rdd", "root", "123456")
      },
      "select * from rddtable where id >= ? and id <= ?;",  //SQL
      1,   // 下界
      10, //上界
      1, //分区数
      r => (r.getInt(1), r.getString(2)))

    println (rdd.count () )
    rdd.foreach (println (_) )
    sc.stop ()
  }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

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

/**
  * Created by 清风笑丶 Cotter on 2019/5/30.
  */
object JDBCRDD2MySQL {
  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("HBaseApp")
    val sc = new SparkContext(sparkConf)
    val data = sc.parallelize(List("JDBC2Mysql", "JDBCSaveToMysql","RDD2Mysql"))

    data.foreachPartition(insertData)
  }

  def insertData(iterator: Iterator[String]): Unit = {
    Class.forName ("com.mysql.jdbc.Driver").newInstance()
    val conn = java.sql.DriverManager.getConnection("jdbc:mysql://datanode1:3306/rdd", "root", "hive")
    iterator.foreach(data => {
      val ps = conn.prepareStatement("insert into rddtable(name) values (?)")
      ps.setString(1, data)
      ps.executeUpdate()
    })
  }

}

1
2
3
4
5

create 'fruit','info'
put 'fruit','1001','info:name','Apple'
put 'fruit','1001','info:color','Read'
put 'fruit','1002','info:name','Banana'
put 'fruit','1002','info:color','Yelow'

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by 清风笑丶 Cotter on 2019/5/30.
  */
object ReadHBase {
  def main(args: Array[String]) {


    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("HBaseApp")
    val sc = new SparkContext(sparkConf)

    sc.setLogLevel("ERROR")
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.1.101");
    conf.set("hbase.zookeeper.property.clientPort", "2181")
    //HBase中的表名
    conf.set(TableInputFormat.INPUT_TABLE, "fruit")

    val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result])

    val count = hBaseRDD.count()
    println("hBaseRDD RDD Count:" + count)
    hBaseRDD.cache()
    hBaseRDD.foreach {
      case (_, result) =>
        val key = Bytes.toString(result.getRow)
        val name = Bytes.toString(result.getValue("info".getBytes, "name".getBytes))
        val color = Bytes.toString(result.getValue("info".getBytes, "color".getBytes))
        println("Row key:" + key + " Name:" + name + " Color:" + color)
    }
    sc.stop()
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

import org.apache.hadoop.hbase.client.{HBaseAdmin, Put}
import org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by 清风笑丶 Cotter on 2019/5/30.
  */
object Write2Hbase {
  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("HBaseApp")
    val sc = new SparkContext(sparkConf)

    sc.setLogLevel("ERROR")
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.1.101");
    conf.set("hbase.zookeeper.property.clientPort", "2181")

    val jobConf = new JobConf(conf)
    jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE, "fruit_spark")

    val fruitTable = TableName.valueOf("fruit_spark")
    val tableDescr = new HTableDescriptor(fruitTable)
    tableDescr.addFamily(new HColumnDescriptor("info".getBytes))

    val admin = new HBaseAdmin(conf)
    if (admin.tableExists(fruitTable)) {
      admin.disableTable(fruitTable)
      admin.deleteTable(fruitTable)
    }
    admin.createTable(tableDescr)

    def convert(triple: (Int, String, Int)) = {
      val put = new Put(Bytes.toBytes(triple._1))
      put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(triple._2))
      put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("price"), Bytes.toBytes(triple._3))
      (new ImmutableBytesWritable, put)
    }
    val initialRDD = sc.parallelize(List((1,"apple",11), (2,"banana",12), (3,"pear",13)))
    val localData = initialRDD.map(convert)

    localData.saveAsHadoopDataset(jobConf)
  }

}

1
2
3
4
5
6
7
8
9
10

val notice = sc.textFile("file:///opt/module/spark/README.md")
 val blanklines = sc.accumulator(0)
val tmp = notice.flatMap(line => {
         if (line == "") {
            blanklines += 1
         }
         line.split(" ")
      })
tmp.count()
blanklines.value

1
2
3
4
5
6

@deprecated("use AccumulatorV2", "2.0.0")
def accumulator[T](initialValue: T)(implicit param: AccumulatorParam[T]): Accumulator[T] = {
  val acc = new Accumulator(initialValue, param)
  cleaner.foreach(_.registerAccumulatorForCleanup(acc.newAcc))
  acc
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

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

import scala.collection.JavaConversions._

class LogAccumulator extends org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] {
  private val _logArray: java.util.Set[String] = new java.util.HashSet[String]()

  override def isZero: Boolean = {
    _logArray.isEmpty
  }

  override def reset(): Unit = {
    _logArray.clear()
  }

  override def add(v: String): Unit = {
    _logArray.add(v)
  }

  override def merge(other: org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]]): Unit = {
    other match {
      case o: LogAccumulator => _logArray.addAll(o.value)
    }

  }

  override def value: java.util.Set[String] = {
    java.util.Collections.unmodifiableSet(_logArray)
  }

  override def copy():org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] = {
    val newAcc = new LogAccumulator()
    _logArray.synchronized{
      newAcc._logArray.addAll(_logArray)
    }
    newAcc
  }
}

// 过滤掉带字母的
object LogAccumulator {
  def main(args: Array[String]) {
    val conf=new SparkConf().setAppName("LogAccumulator").setMaster("local[*]")
    val sc=new SparkContext(conf)

    val accum = new LogAccumulator
    sc.register(accum, "logAccum")
    val sum = sc.parallelize(Array("1", "2a", "3", "4b", "5", "6", "7cd", "8", "9"), 2).filter(line => {
      val pattern = """^-?(\d+)"""
      val flag = line.matches(pattern)
      if (!flag) {
        accum.add(line)
      }
      flag
    }).map(_.toInt).reduce(_ + _)  //1+3+5+6+7+8+9 =32

    println("sum: " + sum)
    for (v <- accum.value) print(v + "")
    println()
    sc.stop()
  }
}

1
2
3

val broadcastVar = sc.broadcast(Array(1,2,3,4,5))
broadcastVar.value
sc.parallelize(Array(1,2,3,4,5,6,7,8)).flatMap(x => (broadcastVar.value)).collect

1
2
3
4

val broadcas = sc.textFile("file:///opt/module/spark/README.md")
val broadcasRDD = broadcas.collect
val c = sc.broadcast(broadcasRDD)
c.value