【spark 算子案例】
2024-08-30 00:59:21
package spark_example01; import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Random; /**
*/
public class PeopleInfoFileGenerator {
public static void main(String[] args){
File file = new File("/Users/xls/Desktop/code/bigdata/data/PeopleInfo.txt"); try {
Random random = new Random();//生成随机数
FileWriter fileWriter = new FileWriter(file);//新建一个文件
for (long i=1;i<=100000000;i++){ //生成1000万个数字
int height = random.nextInt(220);
if (height < 50) {
height = height + 50;
}
String gender = getRandomGender(); //性别方法
if (height < 100 && gender == "M") {
height = height + 100;
}
if (height < 100 && gender == "F") {
height = height + 40;
}
fileWriter.write( i + " " + getRandomGender() + " " + height); //文件格式:ID 性别 身高
fileWriter.write(System.getProperty("line.separator"));
}
fileWriter.flush();
fileWriter.close();
System.out.println("People Information File generated successfully.");
}catch (IOException e){
e.printStackTrace();
}
} public static String getRandomGender(){ //构建一个随机生成性别方法
Random random = new Random();
int randomNum = random.nextInt(2) + 1;
if( randomNum % 2 == 0){
return "M";
}else{
return "F";
}
}
}
package spark_example01; 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 java.util.Arrays;
/**
* Created by Administrator on 2017/11/17.
*/
public class PeopleInfoCalculator {
public static void main(String[] args){ SparkConf sparkConf = new SparkConf().setAppName("PeopleInfoCalculator").setMaster("local[3]"); JavaSparkContext sc = new JavaSparkContext(sparkConf); JavaRDD<String> dataFile = sc.textFile("/Users/xls/Desktop/code/bigdata/data/PeopleInfo.txt"); // step1:过滤出性别为M的数据
JavaRDD<String> maleFilterData = dataFile.filter(new Function<String, Boolean>() { public Boolean call(String s) throws Exception {
return s.contains("M");
}
}); //step2:过滤出性别为F的数据
JavaRDD<String> femaleFilterData = dataFile.filter(new Function<String, Boolean>() { public Boolean call(String s) throws Exception {
return s.contains("F");
}
}); //setp3:得到性别为M的身高数据---对每一行的字符串进行切分,最后取[2]这个字符串---flatMap常用于字符串切分
JavaRDD<String> maleHeightData = maleFilterData.flatMap(new FlatMapFunction<String, String>() {
@Override
public java.util.Iterator<String> call(String s) throws Exception {
return Arrays.asList(s.split(" ")[2]).iterator();
}
}); // step4:得到性别为F的身高数据---对每一行的字符串进行切分,最后取[2]这个字符串---flatMap常用于字符串切分
JavaRDD<String> femaleHeightData = femaleFilterData.flatMap(new FlatMapFunction<String, String>() {
@Override
public java.util.Iterator<String> call(String s) throws Exception {
return Arrays.asList(s.split(" ")[2]).iterator();
}
}); // step5:将男性身高的字符串格式转化为整型格式
JavaRDD<Integer> maleHeightDataInt = maleHeightData.map(new Function<String, Integer>() {//
@Override
public Integer call(String s) throws Exception {
return Integer.parseInt(String.valueOf(s));
}
}); //step6:将女性身高的字符串格式转化为整型格式
JavaRDD<Integer> femaleHeightDataInt = femaleHeightData.map(new Function<String, Integer>() {//将字符串格式转化为整型格式
@Override
public Integer call(String s) throws Exception {
return Integer.parseInt(String.valueOf(s));
}
}); //sortBy(<T>,ascending,numPartitions) 解释:
//第一个参数是一个函数,该函数的也有一个带T泛型的参数,返回类型和RDD中元素的类型是一致的;
//第二个参数是ascending,这参数决定排序后RDD中的元素是升序还是降序,默认是true,也就是升序;
//第三个参数是numPartitions,该参数决定排序后的RDD的分区个数,默认排序后的分区个数和排序之前的个数相等,即为this.partitions.size。 //step7:对男性身高按着从低到高进行排序---参数true表示默认排序,为升序排序,从低到高排
JavaRDD<Integer> maleHeightLowSort = maleHeightDataInt.sortBy(new Function<Integer,Integer>(){
public Integer call(Integer s) throws Exception {
return s;
}
},true,3); //step8:对女性身高按着从低到高进行排序---参数true表示默认排序,为升序排序,从低到高排
JavaRDD<Integer> femaleHeightLowSort = femaleHeightDataInt.sortBy(new Function<Integer,Integer>(){
public Integer call(Integer s) throws Exception {
return s;
}
},true,3); //step9:对男性身高按着从高到低进行排序---false表示为降序排序,从高到低
JavaRDD<Integer> maleHeightHightSort = maleHeightDataInt.sortBy(new Function<Integer,Integer>(){
public Integer call(Integer s) throws Exception {
return s;
}
},false,3); //step10:对女性身高按着从高到低进行排序---false表示为降序排序,从高到低
JavaRDD<Integer> femaleHeightHightSort = femaleHeightDataInt.sortBy(new Function<Integer,Integer>(){
public Integer call(Integer s) throws Exception {
return s;
}
},false,3); Integer lowestMale = maleHeightLowSort.first(); //求出升序的第一个数,即最小值
Integer lowestFemale = femaleHeightLowSort.first();//求出升序的第一个数,即最小值
Integer highestMale = maleHeightHightSort.first();//求出降序的第一个数,即最大值
Integer highestFemale = femaleHeightHightSort.first();//求出降序的第一个数,即最大值 System.out.println("Number of Female Peole:" + femaleHeightData.count());//求出女性的总个数
System.out.println("Number of Male Peole:" + maleHeightData.count());//求出男性的总个数
System.out.println("Lowest Male:" + lowestMale);//求出男性最矮身高
System.out.println("Lowest Female:" + lowestFemale);//求出女性最矮身高
System.out.println("Highest Male:" + highestMale);//求出男性最高身高
System.out.println("Highest Female:" + highestFemale);//求出女性最高身高 }
} /*
* a. 案例描述
本案例假设我们需要对某个省的人口 (10万) 性别还有身高进行统计,需要计算出男女人数,男性中的最高和最低身高,以及女性中的最高和最低身高。
本案例中用到的源文件有以下格式, 三列分别是 ID,性别,身高 (cm),格式如下:
b.人口数据的生成
利用Java语言随机生成一组人口数据,包括序列ID,性别M/F,身高cm,代码如下:
c. 实例过程分析
对于这个案例,我们要分别统计男女的信息,那么很自然的想到首先需要对于男女信息从源文件的对应的 RDD 中进行分离,这样会产生两个新的 RDD,分别包含男女信息;
其次是分别对男女信息对应的 RDD 的数据进行进一步映射,使其只包含身高数据,这样我们又得到两个 RDD,分别对应男性身高和女性身高;
最后需要对这两个 RDD 进行排序,进而得到最高和最低的男性或女性身高。
第一步,先分离男女信息,使用 filter 算子过滤条件包含”M” 的行是男性,包含”F”的行是女性;
第二步我们需要使用 map 算子把男女各自的身高数据从 RDD 中分离出来;
第三步我们需要使用 sortBy 算子对男女身高数据进行排序。
特别注意:RDD 转化的过程中需要把身高数据转换成整数,否则 sortBy 算子会把它视为字符串,那么排序结果就会受到影响,
例如 身高数据如果是:123,110,84,72,100,那么升序排序结果将会是 100,110,123,72,84,显然这是不对的。
d.求出身高统计代码实现:
* */
最新文章
- 双日历时间段选择控件—daterangepicker(汉化版)
- 《微信小程序七日谈》- 第三天:玩转Page组件的生命周期
- Java学习-012-文件删除实例及源代码
- Python requests模块在Windows下安装
- javaweb之Java基础加强
- Process.StandardInput属性
- css 嵌套 元素所属类别
- ueditor .net设置步骤
- 芝麻HTTP:Python爬虫实战之爬取百度贴吧帖子
- mysql物理备份基本操作
- RESTful API接口文档规范小坑
- Oracle RMAN备份与还原注意事项
- H5新属性FileReader实现选择图片后立即显示在页面上
- zabbix3.x自动发现主机
- IOS开发之--iPhone XR,iPhone XS Max适配
- bzoj4278
- oracle上机实验内容
- Python中__init__()方法注意点
- 20155227 2016-2017-2 《Java程序设计》第十周学习总结
- Angular中form表单中input自动响应回车事件无效