大数据hadoop练习___出租车MRWordCount案例

简介: 大数据hadoop练习___出租车MRWordCount案例

WordCountDemo

计算出10月1日这天每小时的载客量

JobMain

package input.mapreduceT1;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class JobMain  {
    //该方法用于指定一个job任务
    public static void main(String[] args) throws Exception {
        Job job = Job.getInstance(new Configuration(), "taxi");
        //设置驱动方法
       job.setJarByClass(JobMain.class);
        //设置map,reduce
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        //设置map
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        //设置reduce
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        //输入输出路径
        FileInputFormat.addInputPath(job,new Path("F:\\Personal\\Desktop\\day05\\代码\\day05_mapreduce\\src\\main\\java\\租车\\1.txt"));
        FileOutputFormat.setOutputPath(job,new Path("F:\\Personal\\Desktop\\day05\\代码\\day05_mapreduce\\src\\main\\java\\output\\Test1.5"));
        boolean b = job.waitForCompletion(true);
        System.exit(b?0:1);
    }
}

WordCountMapper

package input.mapreduceT1;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
/*
  四个泛型解释:
    KEYIN :K1的类型
    VALUEIN: V1的类型
    KEYOUT: K2的类型
    VALUEOUT: V2的类型
 */
public class WordCountMapper  extends Mapper<LongWritable, Text,Text,Text>{
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //214704,4,2,20121001081102,116.5546341,39.9706421,0,128,1
        //车机标识,触发事件,运营状态,GPS时间,GPS经度,GPS纬度,GPS速度,GPS方向,GPS状态
        String[] split = value.toString().split(",");
        if (split.length >= 4){
            //上车
            if ("1".equals(split[1])){
                //10月1日
                String date = split[3].substring(0, 10);
                //判断GPS是否有效
                if (split[8].equals("1")){
                    if (date.substring(0, 8).endsWith("1001")){
                        //key是车牌号+每小时
                        context.write(new Text(date.substring(8,10)),value);                    }
                }
            }
        }
        //key GPS时间_车机标识 value 1
    }
}

WordCountReducer

package input.mapreduceT1;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
/*
  四个泛型解释:
    KEYIN:  K2类型
    VALULEIN: V2类型
    KEYOUT: K3类型
    VALUEOUT:V3类型
 */
public class WordCountReducer extends Reducer<Text,Text,Text,Text>{
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        /**
         * 载客量
         */
        int CarryingCapacity=0;
        HashMap<String, ArrayList<String>> map = new HashMap<>();
        ArrayList<String> list = new ArrayList<>();
        for (Text value : values) {
            String[] split = value.toString().split(",");
            if (split[2].equals("1")){
                CarryingCapacity++;
            }
        }
        context.write(key,new Text(CarryingCapacity+""));
        //每小时每个车载客量
    }
}


相关实践学习
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程:基于MaxCompute的热门话题分析
目录
相关文章
|
存储 分布式计算 Hadoop
从“笨重大象”到“敏捷火花”:Hadoop与Spark的大数据技术进化之路
从“笨重大象”到“敏捷火花”:Hadoop与Spark的大数据技术进化之路
787 79
|
关系型数据库 MySQL 大数据
大数据新视界--大数据大厂之MySQL 数据库课程设计:MySQL 数据库 SQL 语句调优的进阶策略与实际案例(2-2)
本文延续前篇,深入探讨 MySQL 数据库 SQL 语句调优进阶策略。包括优化索引使用,介绍多种索引类型及避免索引失效等;调整数据库参数,如缓冲池、连接数和日志参数;还有分区表、垂直拆分等其他优化方法。通过实际案例分析展示调优效果。回顾与数据库课程设计相关文章,强调全面认识 MySQL 数据库重要性。为读者提供综合调优指导,确保数据库高效运行。
|
存储 分布式计算 大数据
Flume+Hadoop:打造你的大数据处理流水线
本文介绍了如何使用Apache Flume采集日志数据并上传至Hadoop分布式文件系统(HDFS)。Flume是一个高可用、可靠的分布式系统,适用于大规模日志数据的采集和传输。文章详细描述了Flume的安装、配置及启动过程,并通过具体示例展示了如何将本地日志数据实时传输到HDFS中。同时,还提供了验证步骤,确保数据成功上传。最后,补充说明了使用文件模式作为channel以避免数据丢失的方法。
1016 4
|
存储 分布式计算 Hadoop
数据湖技术:Hadoop与Spark在大数据处理中的协同作用
【10月更文挑战第27天】在大数据时代,数据湖技术凭借其灵活性和成本效益成为企业存储和分析大规模异构数据的首选。Hadoop和Spark作为数据湖技术的核心组件,通过HDFS存储数据和Spark进行高效计算,实现了数据处理的优化。本文探讨了Hadoop与Spark的最佳实践,包括数据存储、处理、安全和可视化等方面,展示了它们在实际应用中的协同效应。
770 2
|
存储 分布式计算 Hadoop
数据湖技术:Hadoop与Spark在大数据处理中的协同作用
【10月更文挑战第26天】本文详细探讨了Hadoop与Spark在大数据处理中的协同作用,通过具体案例展示了两者的最佳实践。Hadoop的HDFS和MapReduce负责数据存储和预处理,确保高可靠性和容错性;Spark则凭借其高性能和丰富的API,进行深度分析和机器学习,实现高效的批处理和实时处理。
702 1
|
分布式计算 大数据 Linux
大数据体系知识学习(二):WordCount案例实现及错误总结
这篇文章介绍了如何使用PySpark进行WordCount操作,包括环境配置、代码实现、运行结果和遇到的错误。作者在运行过程中遇到了Py4JJavaError和JAVA_HOME未设置的问题,并通过导入findspark初始化和设置环境变量解决了这些问题。文章还讨论了groupByKey和reduceByKey的区别。
376 1
|
10月前
|
机器学习/深度学习 传感器 分布式计算
数据才是真救命的:聊聊如何用大数据提升灾难预警的精准度
数据才是真救命的:聊聊如何用大数据提升灾难预警的精准度
638 14
|
12月前
|
数据采集 分布式计算 DataWorks
ODPS在某公共数据项目上的实践
本项目基于公共数据定义及ODPS与DataWorks技术,构建一体化智能化数据平台,涵盖数据目录、归集、治理、共享与开放六大目标。通过十大子系统实现全流程管理,强化数据安全与流通,提升业务效率与决策能力,助力数字化改革。
412 4
|
11月前
|
机器学习/深度学习 运维 监控
运维不怕事多,就怕没数据——用大数据喂饱你的运维策略
运维不怕事多,就怕没数据——用大数据喂饱你的运维策略
1074 0