初识Spark

偏执的太偏执、 2021-12-10 05:25 204阅读 0赞

# Spark基础组件 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70][]

Spark SQL可以操作数据库，比如hive等等。可以用于批量操作

Spark Streaming主要是对流进行处理，也就是实时处理。一般是搭配kafka进行配置

ML是机器学习相关的

so on

# 简单Spark整合Java，进行统计数据 #

上代码

package mysparkkafkalikes;
 
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.api.java.function.FlatMapFunction;
 import org.apache.spark.api.java.function.Function;
 import org.apache.spark.api.java.function.Function2;
 
 import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Random;
 
 /**
 * Created on 2019/7/15.
 *
 * @author yangsen
 */
 public class a {
 
 public static void main(String[] args){
 File file = new File("C:\\Users\\Max.Yang\\Desktop\\bb.txt");
 try {
 FileWriter fileWriter = new FileWriter(file);
 Random rand = new Random();
 for (int i=1;i<=100000;i++){
 fileWriter.write(i +" " + (rand.nextInt(100)+1));
 fileWriter.write(System.getProperty("line.separator"));
 }
 fileWriter.flush();
 fileWriter.close();
 
 }catch(IOException e){
 e.printStackTrace();
 }
 
 
 SparkConf sparkConf = new SparkConf().setAppName("AvgAgeCalculator").setMaster("local[3]");
 JavaSparkContext sc = new JavaSparkContext(sparkConf);
 //读取文件
 JavaRDD<String> dataFile = sc.textFile("C:\\Users\\Max.Yang\\Desktop\\bb.txt");
 //数据分片并取第二个数
 JavaRDD<String> ageData = dataFile.flatMap(new FlatMapFunction<String, String>() {
 @Override
 public Iterable<String> call(String s) throws Exception {
 return Arrays.asList(s.split(" ")[1]);
 }
 });
 //求出所有年龄个数。
 long count = ageData.count();
 //转换数据类型
 JavaRDD<Integer> ageDataInt = ageData.map(new Function<String, Integer>() {
 @Override
 public Integer call(String s) throws Exception {
 return Integer.parseInt(String.valueOf(s));
 }
 });
 //求出年龄的和
 Integer totalAge = ageDataInt.reduce(new Function2<Integer, Integer, Integer>() {
 @Override
 public Integer call(Integer x, Integer y) throws Exception {
 return x+y;
 }
 });
 //平均值结果为double类型
 Double avgAge = totalAge.doubleValue()/count;
 /*System.out.println(ageData.collect());
 System.out.println(count);*/
 System.out.println("Total Age:" + totalAge + "; Number of People:" + count );
 System.out.println("Average Age is " + avgAge);
 
 }
 
 }

Spark也可以操作hdfs上的文件，文件路径修改为hdfs://host:port

RDD是Spark的弹性分布式数据集，抽象的数据集。将查询的数据放在缓存中，后续的工作可以重用工作集，减少查询

reduce是当存在相同的key时做的合并操作。

# RDD操作 #

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> 转换 </td> 
 <td style="vertical-align:top;"> 含义 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> map(func) </td> 
 <td style="vertical-align:top;"> 返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> filter(func) </td> 
 <td style="vertical-align:top;"> 返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> flatMap(func) </td> 
 <td style="vertical-align:top;"> 类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素） </td> 
 </tr> 
 </tbody> 
</table>

基本使用，类似lambda map元素映射

# Action操作 #

一般一个spark都会有一个action操作

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> 动作 </td> 
 <td style="vertical-align:top;"> 含义 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> reduce(func) </td> 
 <td style="vertical-align:top;"> 通过func函数聚集RDD中的所有元素，这个功能必须是课交换且可并联的 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> collect() </td> 
 <td style="vertical-align:top;"> 在驱动程序中，以数组的形式返回数据集的所有元素 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> count() </td> 
 <td style="vertical-align:top;"> 返回RDD的元素个数 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> first() </td> 
 <td style="vertical-align:top;"> 返回RDD的第一个元素（类似于take(1)） </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> take(n) </td> 
 <td style="vertical-align:top;"> 返回一个由数据集的前n个元素组成的数组 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> takeSample(withReplacement,num, [seed]) </td> 
 <td style="vertical-align:top;"> 返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> takeOrdered(n,&nbsp;[ordering]) </td> 
 <td style="vertical-align:top;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> saveAsTextFile(path) </td> 
 <td style="vertical-align:top;"> 将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> saveAsSequenceFile(path)&nbsp; </td> 
 <td style="vertical-align:top;"> 将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> saveAsObjectFile(path)&nbsp; </td> 
 <td style="vertical-align:top;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> countByKey() </td> 
 <td style="vertical-align:top;"> 针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> foreach(func) </td> 
 <td style="vertical-align:top;"> 在数据集的每一个元素上，运行函数func进行更新。 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> aggregate </td> 
 <td style="vertical-align:top;"> 先对分区进行操作，在总体操作 </td> 
 </tr> 
 </tbody> 
</table>

上面很多摘自[https://www.cnblogs.com/qingyunzong/p/8899715.html][https_www.cnblogs.com_qingyunzong_p_8899715.html]

# Spark Streaming #

对实时流进行处理

SparkConf conf = new SparkConf()
 .setAppName("mySparkLikes")
 .setMaster("local[*]")
 .set("spark.default.parallelism", "15")
 .set("spark.streaming.concurrentJobs", "5")
 .set("spark.executor.memory", "1G")
 .set("spark.cores.max", "3")
 .set("spark.local.dir", "/tmp/mySparkLikes")
 .set("spark.streaming.kafka.maxRatePerPartition", "5");
 
 Set<String> topics = Collections.singleton(topic);
 
 Map<String, String> kafkaParams = new HashMap<>();
 kafkaParams.put("metadata.broker.list", "localhost:9092");
 
 JavaStreamingContext jsc = new JavaStreamingContext(
 new JavaSparkContext(conf),
 Durations.seconds(3));
 jsc.checkpoint("checkpoint"); //保证元数据恢复，就是Driver端挂了之后数据仍然可以恢复
 
 // 得到数据流
 final JavaPairInputDStream<String, String> stream = KafkaUtils.createDirectStream(
 jsc,
 String.class,
 String.class,
 StringDecoder.class,
 StringDecoder.class,
 kafkaParams,
 topics
 );
 System.out.println("stream started!");
 stream.print();
 
 
 
 JavaPairDStream<Integer, Integer> countDStream = stream
 .transformToPair(new Function<JavaPairRDD<String, String>, JavaPairRDD<Integer, Integer>>() {
 
 @Override
 public JavaPairRDD<Integer, Integer> call(JavaPairRDD<String, String> stringStringJavaPairRDD) throws Exception {
 return stringStringJavaPairRDD.mapToPair(new PairFunction<Tuple2<String, String>, Integer, Integer>() {
 
 @Override
 public Tuple2<Integer, Integer> call(Tuple2<String, String> stringStringTuple2) throws Exception {
 return new Tuple2<>(new Integer(stringStringTuple2._1), new Integer(stringStringTuple2._2));
 }
 });
 }
 })
 .reduceByKey(Integer::sum);
 
 countDStream.foreachRDD(v -> {
 v.foreach(record -> {
 String sql = String.format("UPDATE `post` SET likes = likes + %s WHERE id=%d", record._2, record._1);
 System.out.println(sql);
 });
 log.info("一批次数据流处理完： {}", v);
 });
 jsc.start();

checkpoint是Spark中的容错机制，会将数据保存在磁盘中，即使RDD出错，也可以恢复，而不用重新查

Receiver Spark默认有Receiver，我们也可以自定义Receiver

._1,._2可以表示两个参数

可以去查阅下DStream操作

Spark SQL与ML暂时不总结

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]: /images/20211209/04c80993d4b64768a9432b94fe3a950e.png
[https_www.cnblogs.com_qingyunzong_p_8899715.html]: https://www.cnblogs.com/qingyunzong/p/8899715.html