前往小程序，Get更优阅读体验！
立即前往
文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
首页
学习
活动
专区
圈层
工具
MCP广场
返回腾讯云官网
社区首页 >专栏 >Hadoop和spark基础使用

Hadoop和spark基础使用

作者头像
栖西
发布于 2023-10-17 00:20:11
发布于 2023-10-17 00:20:11
31400
代码可运行
举报
文章被收录于专栏：栖西栖西
运行总次数：0
代码可运行
关联问题
换一批

前言：以下所有知识仅限于入门，对自己代码的一个记录，有不详细之处留到日后补录。

Hadoop

map的输入固定是LongWritable和Text，可理解为偏移量和String类型的数据。 核心：map的输出的key和value是reduce的输入的key和value

1、求和

主类

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
 public static void main(String[] args) throws Exception{

        // 1、初始化配置，告诉取得程序
        Configuration configuration = new Configuration();
        // 2、初始化任务
        Job job = Job.getInstance(configuration);
        job.setJarByClass(PriceSumMain.class);
        // 3、开始配置map 所有的map都在做切割文件
        job.setMapperClass(PriceSumMapper.class);
        // 设置map阶段的输入和输出阶段
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        // 4、开始配置reduce
        job.setReducerClass(PriceSumReduce.class);
        // 设置reduce的输出
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        // 5、设置输入输出结果集
        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        // 6、提交任务
        System.exit(job.waitForCompletion(true)?0:1);

    }

mapper

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
// 求价格总和
// 偏移量  数据  输出的key  输出的value
public class PriceSumMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        // 取得每一行数据
        String line=value.toString();
        // 对每一行数据进行切割
        String[] items = line.split(",");
        // 取得第一列的数据
        int num = Integer.parseInt(items[1]);

        context.write(new Text("num"), new LongWritable(num));
    }

}

reduce

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
public class PriceSumReduce extends Reducer<Text, LongWritable, Text, LongWritable> {

    protected void reduce(Text key, Iterable<LongWritable> values,Context context) throws IOException, InterruptedException {

        long sum =0L;
        for(LongWritable a:values){
            sum +=a.get();
        }

        context.write(new Text("最后求出的数据："), new LongWritable(sum));

    }

}
2、排序

排序分为俩种，数字排序和字母排序。排序是对key的排序 根据数字排序，默认升序。

主类

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
    public static void main(String[] args) throws Exception{

        Configuration configuration = new Configuration();

        Job job = Job.getInstance(configuration);
        job.setJarByClass(SortSalaryMain.class);

        // 根据数字排序
        job.setMapperClass(SortSalaryByNumMapper.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        // 数字排序无需reduce

        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        System.exit(job.waitForCompletion(true)?0:1);


    }

mapper

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
// 员工薪资排序
public class SortSalaryByNumMapper extends Mapper<LongWritable, Text, IntWritable, Text> {
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        // 获取到每一行数据
        String line = value.toString();
        // 切割数据
        String[] emps = line.split(",");
        // 获取员工姓名
        String name=emps[1];
        // 获取员工薪资
        int salary =Integer.parseInt(emps[5]);

        context.write( new IntWritable(salary),new Text(name));
    }
}

根据字母排序

主类

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
 public static void main(String[] args) throws Exception{

        Configuration configuration = new Configuration();

        Job job = Job.getInstance(configuration);
        job.setJarByClass(SortSalaryByCharMain.class);

        job.setMapperClass(SortSalaryByNumMapper.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        // 配置其他 根据字母排序
        job.setSortComparatorClass(SortSalaryComparator.class);

        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        System.exit(job.waitForCompletion(true)?0:1);


    }

排序

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
// WritableComparator
public class SortSalaryComparator extends WritableComparator {
//public class SortSalaryComparator implements RawComparator {

    // 构造器调用父类方法，告诉Comparator(比较器) 需要比较的数据类型是什么
    public SortSalaryComparator(){
        super(IntWritable.class,true);
    }

    // 重写比较方法，默认升序，实现降序 自定义比较规则

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        // 向下转型
        IntWritable xa = (IntWritable) a;
        IntWritable xb = (IntWritable) b;

        // 也执行成功
        return -super.compare(a, b);

        // 可行
//        return -xa.compareTo(xb);
    }
}
3、去重

主类

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
 public static void main(String[] args) throws Exception {

        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        job.setJarByClass(DistinctJobMain.class);

        job.setMapperClass(DistinctJobMapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        job.setReducerClass(DistinctJobReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        System.exit(job.waitForCompletion(true)?0:1);

    }

mapper

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
public class DistinctJobMapper extends Mapper <LongWritable, Text, Text, NullWritable> {

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line=value.toString();

        String[] items = line.split(",");
        // 获取员工的职位
        String job = items[2];
        // 因为原语，所以将所有的key都设置为相同的
        context.write(new Text(job), NullWritable.get());
    }

}

reduce

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
public class DistinctJobReduce extends Reducer <Text, NullWritable, Text, NullWritable> {
    protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        // 利用原语，相同的key为一组，调用一个或一次reduce，过滤重复的工作
            context.write(key, NullWritable.get());

    }
}
4、分区

需求：根据薪资进行分区 （0,1500，5000，10000，以上）

主类

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
 // 根据员工薪资分区
    public static void main(String[] args) throws Exception{
        // 1、初始化配置，告诉取得程序
        Configuration configuration = new Configuration();
        // 2、初始化任务
        Job job = Job.getInstance(configuration);
        job.setJarByClass(PartitionBySalaryMain.class);
        // 3、开始配置map 所有的map都在做切割文件
        job.setMapperClass(PartitionMapper.class);
        // 设置map阶段的输入和输出阶段
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 4、开始配置reduce
        job.setReducerClass(PartitionReduce.class);
        // 设置reduce的输出
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置分区
        job.setPartitionerClass(EmpPartitionBySalary.class);
        // 设置分区数量  按工资分区0,1500,5000,10000
        job.setNumReduceTasks(4);


        // 5、设置输入输出结果集
        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        // 6、提交任务
        System.exit(job.waitForCompletion(true)?0:1);
    }

mapper

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
public class PartitionMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();

        String[] items = line.split(",");

        // 获取员工的薪资
        int salary = Integer.parseInt(items[5]);
        // 获取员工的姓名
        String name = items[1];

        context.write(new Text(name), new IntWritable(salary));
    }

}

reduce

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
public class PartitionReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        for (IntWritable value:values){
            context.write(key,value);
        }


    }
}

分区类

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
public class EmpPartitionBySalary extends Partitioner<Text, IntWritable> {
    @Override
    public int getPartition(Text name, IntWritable salary, int numPartition) {
//        根据薪资进行分区  （0,1500，5000，10000,以上）
        int sal = salary.get();

        if(sal<1500){
            return 1% numPartition;
        }else if(sal < 5000){
            return 2%numPartition;
        }else if(sal <10000){
            return 3%numPartition;
        }

        return 4%numPartition;
    }
}
5、合并

主类

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
/**
 * 合并员工信息
 */
public class JoinInfoMain {
    public static void main(String[] args) throws Exception{

        Configuration configuration = new Configuration();

        Job job = Job.getInstance(configuration);
        job.setJarByClass(JoinInfoMain.class);

        job.setMapperClass(JoinInfoMapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setReducerClass(JoinInfoReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        System.exit(job.waitForCompletion(true)?0:1);


    }
}

mapper

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
//在做多个文档联合数据分析的时候，一定要注意你join的点在哪地方
public class JoinInfoMapper extends Mapper<LongWritable, Text, Text, Text> {
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();

        String[] items = line.split(",");

        // 根据每一行的长度来判断是哪一个文档，根据不同的文档来取值
        if(items.length==8){
            String name = items[1];
            String job = items[2];
            context.write(new Text(job),new Text(name));
        }else if(items.length==3){
            String job = items[0];
            String desc=items[2];  //  获取岗位的描述
            context.write(new Text(job), new Text("--"+desc));
        }

        // 打包，去分析，现在输出的是什么，然后在分析reduce


// 合并的话，要知道那个字段相同，他们两个数据集的key相同，原语，去重，合并
//   context.write(new Text(job),new Text(name));   第一个数据集
//  context.write(new Text(job), new Text("--"+desc));  第二个数据集


    }
}

reduce

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
public class JoinInfoReduce extends Reducer<Text, Text, Text, Text> {
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        // 创建一个变量，用于存储该岗位下的所有员工姓名
        StringBuffer names = new StringBuffer();
        // 岗位信息
        String jobInfo = "";
        for(Text value:values){

            // 取得mapper传过来的所有数据
            String msg = value.toString();

            // 判断是否包含--  是的话，则为岗位描述
            boolean flag = msg.contains("--");
            if(flag){
                jobInfo=msg;
            }else {
                // 将所有的values组合起来
                names.append(msg+"、");
            }

        }

        // 输出join之后的数据集
        context.write(new Text(jobInfo),new Text(names.toString()));

    }
}
6、topN

待后期完善

导入依赖
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
   <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.1.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.1.2</version>
        </dependency>
    </dependencies>
使用
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
Hadoop节点的顺序：NameNode---DataNode---SecondaryNameNode
1、将文件从Windows本地传到Linux本地，使用winscp即可
2、将文件从Linux本地上传到hdfs上
hdfs dfs -put 本地文件  hdfs上的路径
3、执行
hadoop jar jar的路径 Java类的包名+主类名  数据集在hdfs的位置 hdfs的输出路径
例如：hadoop jar /usr/mydata/restuemp-1.0.0.jar com.mypartition.PartitionBySalaryMain /data/newemp.csv /out1754

可能会遇到的问题，解决方案
Hadoop离开安全模式
hadoop dfsadmin -safemode leave

Hive

1、常用命令
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
（-e）不进入hive，执行命令：hive -e "select id from student;"
（-f）执行脚本SQL语句：touch hivef.sql
	 执行文件中的SQL  hive -f hivef.sql
	 执行文件中的SQL，并将结果写入到文件中
	 				hive -f hivejs.sq > result-hive.txt
可通过hive查看hdfs文件
dfs -ls /
退出：quit;(直接退出)或者exit;(先隐型提交数据，在退出)

在hive中查看本地文件：
! ls /root/tools

在~目录下，输入 cat .hivehistory 可查看所有hive的历史命令

经典语句：row format delimited fields terminated by ‘\t’

2、数据操作
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
创建分区表
注意：分区字段不能是表中已经存在的数据，可以将分区字段看作表的伪列。
create table dept_partition(
deptno int, dname string, loc string
)
partitioned by (month string)
row format delimited fields terminated by '\t';

加载数据到分区表
注意：分区表加载数据时，必须指定分区
load data local inpath '/root/tools/dept.txt' into table dept_partition partition(month='20210722');
查询分区数据
select * from dept_partition where month='20210722';

多个分区联合查询
select * from dept_partition where month='20210721'
union
select * from dept_partition where month='20210722'
union
select * from dept_partition where month='20210723';

创建二级分区表
create table dept_partition2(
deptno int, dname string, loc string
)
partitioned by (month string, day string)
row format delimited fields terminated by '\t';
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
加载本地文件到hive表里面  
load data local inpath '/root/emp.txt' into table emp;

加载HDFS文件到hive里面---加载数据覆盖本地已有数据
load data inpath '/dara/student.txt' overwrite into table student;

通过查询语句向表中插入数据
将结果文件写入本地	去掉local 则为写到hdfs路径下
insert overwrite local directory '/usr/local/distribute-result' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' select * from emp distribute by deptno sort by idno desc;

import数据到指定hive表中
import table student2 partition(month='201709') from
 '/user/hive/warehouse/export/student';
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
  数据导出
 insert导出  导出到hdfs 下没有local
 insert overwrite local directory '/opt/module/datas/export/student1'
 ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'            select * from student;
 
 hadoop命令，从hdfs下载到本地  get的使用
 hdfs dfs -get resource_path_name target_path_name
 
 hive shell导出
 hive -e 'select * from default.student;' > /data/student4.txt;
 
 export 导出到hdfs上
 export table default.student to '/data/student';
 export和import主要用于两个Hadoop平台集群之间Hive表迁移。
3、表操作
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
创建表时，location指定加载数据路径
create external table if not exists student5(
id int, name string
)row format delimited fields terminated by '\t'
location '/student';
重命名表
ALTER TABLE table_name RENAME TO new_table_name
删除表
drop table table_name
清除表中数据（Truncate）:Truncate只能删除管理表，不能删除外部表中数据
truncate table student;
增加单个分区
alter table dept_partition add partition(month='201706') ;
同时创建多个分区
alter table dept_partition add partition(month='201705') partition(month='201704');
删除分区  删除多个分区逗号隔开
alter table dept_partition drop partition (month='201704');
查看分区表有多少分区
show partitions dept_partition;
查看分区表结构
desc formatted dept_partition;

ADD是代表新增一字段，字段位置在所有列后面(partition列前)，REPLACE则是表示替换表中所有字段。
添加字段
 alter table dept_partition add columns(deptdesc string);
 替换字段
 alter table dept_partition replace columns(deptno string, dname
 string, loc string);
 更新列
 alter table dept_partition change column deptdesc desc int;

sqoop

Sqoop是一款开源的工具，主要用于在Hadoop(Hive)与传统的数据库(mysql、postgresql…)间进行数据的传递，可以将一个关系型数据库（例如 ： MySQL ,Oracle ,Postgres等）中的数据导进到Hadoop的HDFS中，也可以将HDFS的数据导进到关系型数据库中。

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
从hive写到MySQL  即从hdfs-->mysql   我的文件名是course

sqoop export \
--connect jdbc:mysql://192.168.233.133:3306/company \
--username root \
--password Su#2021Gjj  \
--table deptsumsalary \
--num-mappers 1 \
--export-dir /output/emp/course  \
--input-fields-terminated-by ","
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
MySQL到hdfs

sqoop import \
--connect jdbc:mysql://192.168.233.133:3306/xyxy \
--username root \
--password Su#2021Gjj \
--table tbl_student \
--target-dir /output/tbl_student \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by ","
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
mysql到hive   默认写到default数据库  其他数据的话，前面加数据库名.

sqoop import \
--connect jdbc:mysql://192.168.233.133:3306/xyxy \
--username root \
--password Su#2021Gjj \
--table tbl_student \
--num-mappers 1 \
--hive-import \
--fields-terminated-by "," \
--hive-overwrite \
--hive-table tbl_student_hive

逗号和制表符都是同样的效果

scala

spark也是基于scala开发的

spark

sparkSql
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
import java.text.SimpleDateFormat

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

// 创建样例类  将数据集和样例类做一个映射
case class userview(userid:String,productid:String,ptypeid:String,beh:String,time:String)

/**
  * 利用spark SQL 分析用户行为
  */
object SparkDemo {
  def main(args: Array[String]): Unit = {
    val appName = "demo" // 给当前任务取一个名字
    val master = "local[*]"  //初始化核数 ：*代表根据具体的电脑CPU来模拟核数  有几个核心就起几个线程来处理当前任务
    // 初始化spark配置环境
    val conf = new SparkConf().setAppName(appName).setMaster(master)
    // 初始化spark session  
    val spark = SparkSession.builder().config(conf).getOrCreate()
    // 初始化spark context
    val sc = spark.sparkContext

    // 1、读取数据
    val data = sc.textFile("E:\\java\\workplace2018\\studyknowlege\\sparkdemo\\src\\main\\resources\\UserBehavior.csv")

    // 导入隐式转换  就可以使用toDF()了
    import spark.implicits._

    // 2、将数据集和样例类进行映射
    val userDF = data.map{
      line =>
        // 将行数据用"," 切割，形成数组
        val array = line.split(",")
        // 返回一个样例对象：对象使用数组作为数据填充
        userview(array(0),array(1),array(2),array(3),array(4))
    }.toDF()

    // 3、统计用户id为1的所有用户的行为
    userDF.createOrReplaceTempView("uv")
//    spark.sql("select * from uv where userid = 1").show()
//		show方法是打印到控制台
//    spark.sql("select count(distinct(userid)) as alltotal from uv").show()

//        spark.sql("select beh count(beh) from uv group by beh ").show()

    // 统计某一个月用户的行为是什么样的


    // 自定义一个函数：用于处理时间数据
    def formateDate(timestamp:String):String = {
      val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        simpleDateFormat.format(timestamp.toLong * 1000).split(" ")(0)
    }

    // 将自定义函数注册 _
    spark.udf.register("formateDate",formateDate _)


    // 将自定义匿名(处理时间)函数注册到spark当中去  可用
//    spark.udf.register("formateDate",(timestamp: String) => {
//      val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
//      simpleDateFormat.format(timestamp.toLong * 1000).split(" ")(0)
//    })



    // 重新通过formatDate函数处理数据集
    val datanew = spark.sql("select userid,productid,ptypeid,beh,formateDate(time) from uv where userid = 1").show()

    // 问题：使用group by 和 order by 统计每日用户活跃量


    // 如何将统计好的数据存储到mysql中
    val userbehDF = spark.sql("select beh,count(beh) from uv group by beh")
    userbehDF.write.format("jdbc")  // 告诉spark 存储的时候格式化数据的方式
      .option("url","jdbc:mysql://192.168.233.133:3306/company") // 告诉spark存储的路径和数据库
      .option("dbtable","userbeh")  // 告诉spark 要操作的表（无需在数据库创建，直接创建并插入值）
      .option("user","root")
      .option("password","Su#2021Gjj")
      .mode("Overwrite")  // Overwrite覆盖、Append追加还是其他
      .save()   // 开始存储

    sc.stop()
    spark.stop()

  }

}
sparkStream
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
package com.sparkStream

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

/**
  * sprakStreaming
  */
object SparkStreamDemo {
  def main(args: Array[String]): Unit = {
    // 1、告诉spark 我的配置 核心数 本次运行驱动的名字
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    // 2、初始化StreamingContext 告诉spark 每五秒读取一次数据
    val ssc = new StreamingContext(conf, Seconds(10))   // ，每五秒读取一次
    // 3、监控端口，开始读取数据
//    val lines = ssc.socketTextStream("192.168.233.133", 9999)   // Linux启动9999端口  nc -lk 9999
    val lines = ssc.textFileStream("hdfs://192.168.233.133:9000/sparkStream/")
    //  可以合一句话
//    val words = lines.flatMap(_.split(" "))
//    val pairs = words.map(word => (word, 1))
//    val wordCounts = pairs.reduceByKey(_ + _)

    // 4、将读到的数据可以做Wordcount的处理  可按照RDD的处理算子来进行调用
    val wordCounts = lines.flatMap(_.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)
//    val wordCounts = lines.flatMap(_.split(" ")).map(word => (word,1)).reduceByKey( (a,b) => a+b)
    // 5、输出数据集
    wordCounts.print()
    // 6、将sparkstream启动  开始监控
    ssc.start()
    ssc.awaitTermination()

  }
}

附录

操作完数据之后，进行数据渲染

jQuery关键性代码
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
<script src="./static/jquery.min.js" charset="utf-8"></script>
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
   $(function () {
            $.get("./servletdemo",function(data,status){
                var result = JSON.parse(data);
                var x = new Array();
                var y = new Array();
                for(var i = 0; i< result.length; i++){
                    x.push(result[i].deptno);
                    y.push(result[i].sumsalary);
                }
                console.log(x);
                console.log(y);
        });
     
另一个案例
// 组成这样的形式{value: 60, name: '访问'}, 即 records =[{...},{},{}]
var records = Array();  
for(var i=0;i<result.length;i++){
records.push({"value":result[i].counts,"name":result[i].beh})
}
console.log("新的data记录集：",records)
echarts关键代码

可参考菜鸟教程地址：学习echarts

代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
<div id="main1" style="width: 600px;height:400px;"></div>
<script src="./static/echarts.min.js" charset="utf-8"></script>
代码语言：javascript
代码运行次数：0
运行
AI代码解释
复制
   $(function () {
            $.get("./servletdemo",function(data,status){
                var result = JSON.parse(data);
                var x = new Array();
                var y = new Array();
                for(var i = 0; i< result.length; i++){
                    x.push(result[i].deptno);
                    y.push(result[i].sumsalary);
                }
                console.log(x);
                console.log(y);
                // 基于准备好的dom，初始化echarts实例
                var myChart1 = echarts.init(document.getElementById('main1'));

                // 指定图表的配置项和数据
                var option1 = {
                    title: {
                        text: '部门员工薪资总和'
                    },
                    tooltip: {},
                    legend: {
                        data:['薪资总和']
                    },
                    xAxis: {
                        data: x
                    },
                    yAxis: {},
                    series: [{
                        name: '销量',
                        type: 'bar',
                        data: y
                    }]
                };

                // 使用刚指定的配置项和数据显示图表。
                myChart1.setOption(option1);


            });
        });
本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
原始发表：2023-10-13，如有侵权请联系 cloudcommunity@tencent.com 删除
spark
hadoop
基础
排序
数据

本文分享自 作者个人站点/博客 前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ，欢迎热爱写作的你一起参与！

spark
hadoop
基础
排序
数据
评论
登录后参与评论
暂无评论
登录 后参与评论
推荐阅读
大数据：Hadoop-MapReduce练习
mapreducehadoop
第9、10、11、12字符为年份，第20、21、22、23字符代表温度，求每年的最高温度
许喜朝
2020/10/26
4160
MapReduce操作实例
mapreducejavaapachehadoop存储
要点：有一个combiner方法，可以在执行完map时调用，从而对数据进行先一步的处理，降低Reduce的IO压力。
曼路
2018/10/18
1.6K0
Hadoop: MapReduce2的几个基本示例
hadoopmapreduce
1) WordCount  这个就不多说了,满大街都是,网上有几篇对WordCount的详细分析 http://www.sxt.cn/u/235/blog/5809 http://www.cnblogs.com/zhanghuijunjava/archive/2013/04/27/3036549.html 这二篇都写得不错, 特别几张图画得很清晰 2) 去重处理(Distinct) 类似于db中的select distinct(x) from table , 去重处理甚至比WordCount还要简单,假如我
菩提树下的杨过
2018/01/19
1.1K0
Hadoop基础教程-第7章 MapReduce进阶（7.6 MapReduce 二次排序）
node.jsmapreducelinux大数据oracle
MapReduce框架对处理结果的输出会根据key值进行默认的排序，这个默认排序可以满足一部分需求，但是也是十分有限的。在我们实际的需求当中，往往有要对reduce输出结果进行二次排序的需求。所谓二次排序，先按第1个字段进行排序，然后对第1个字段相同的数据，再按第2个字段进行排序。
程裕强
2022/05/06
2290
几个关于MapReduce的小例子
mapreduce文件存储打包hadoopjar
上一篇文章通过写一个WordCount学习了MapReduce的入门操作，那么这篇文章继续通过多一些例子来学习MapReduce。下面介绍几种比较常见的操作：排序，去重，求和，求平均数，TopK查询(查询排名前K名的记录)
java技术爱好者
2022/03/15
4770
几个关于MapReduce的小例子
hadoop_入门1
hadoop
该文介绍了如何使用 Hadoop 进行单词计数和排序。首先介绍了 Hadoop 的整体架构，然后详细讲解了如何使用 Hadoop 进行单词计数和排序。最后，通过一个实例展示了如何使用 Hadoop 进行排序操作。
用户1147754
2018/01/03
6340
hadoop_入门1
Hadoop（十五）MapReduce程序实例
hadoopmapreduce数据库
一、统计好友对数（去重） 1.1、数据准备 joe, jon joe , kia joe, bob joe ,ali kia, joe kia ,jim kia, dee dee ,kia dee, ali ali ,dee ali, jim ali ,bob ali, joe ali ,jon jon, joe jon ,ali bob, joe bob ,ali b
用户1195962
2018/01/18
1.3K0
Hadoop（十五）MapReduce程序实例
MapReduce
mapreduce排序配置数据序列化
(4) ReduceTask进程对每-组相同k的<K，V>组调用一次reduce()方法
CBeann
2023/12/25
4380
MapReduce
MapReduce词频统计【自定义复杂类型、自定义Partitioner、NullWritable使用介绍】
大数据mapreduce
InputFormat读数据，通过Split将数据切片成InputSplit，通过RecordReader读取记录，再交给map处理，处理后输出一个临时的<k,v>键值对，再将结果交给shuffle处理，最终在reduce中将最后处理后的<k,v>键值对结果通过OutputFormat重新写回到HDFS中。
Java架构师必看
2021/05/14
9150
MapReduce词频统计【自定义复杂类型、自定义Partitioner、NullWritable使用介绍】
使用Eclipse编译运行MapReduce程序
javamapreduce
mapreduce也是比较久学的，详细的内容和操作可以看下面的文档。 点击下载 链接：https://pan.baidu.com/s/1BIBpClKy2xcqAJtxUJoYVA 提取码：ctca
不愿意做鱼的小鲸鱼
2022/09/24
8640
使用Eclipse编译运行MapReduce程序
hadoop系列之MR的经典代码案例一
hadoop
七、MapReduce经典案例 1、网站分析案例 1）分析 省份访问 procinceId --> Key 1 -->Value <procinceId,lis
Spark学习技巧
2018/01/31
2.2K0
hadoop系列之MR的经典代码案例一
《快学BigData》--Hadoop总结（G）（40）
java编程算法文件存储mapreducehadoop
Hadoop总结 - - - - - - - - - - - - - - - - - - - - - - - - - - - - 210
小徐
2019/08/05
4250
《快学BigData》--Hadoop总结（G）（40）
Hadoop学习：深入解析MapReduce的大数据魔力（二）
排序数据mapreducehadoop大数据
1、问题引出 要求将统计结果按照条件输出到不同文件中（分区）。比如：将统计结果按照手机 归属地不同省份输出到不同文件中（分区） 2、默认Partitioner分区
老虎也淘气
2024/01/30
1540
Hadoop学习：深入解析MapReduce的大数据魔力（二）
Hadoop学习：深入解析MapReduce的大数据魔力之数据压缩（四）
压缩apachemapreducehadoop大数据
压缩方式选择时重点考虑：压缩/解压缩速度、压缩率（压缩后存储大小）、压缩后是否 可以支持切片。
老虎也淘气
2024/01/30
2400
Hadoop学习：深入解析MapReduce的大数据魔力之数据压缩（四）
Hadoop Partitioner使用教程
mapreduce
partitioner在处理输入数据集时就像条件表达式(condition)一样工作。分区阶段发生在Map阶段之后，Reduce阶段之前。partitioner的个数等于reducer的个数(The number of partitioners is equal to the number of reducers)。这就意味着一个partitioner将根据reducer的个数来划分数据(That means a partitioner will divide the data according to the number of reducers)。因此，从一个单独partitioner传递过来的数据将会交由一个单独的reducer处理(the data passed from a single partitioner is processed by a single Reducer)。
smartsi
2019/08/07
7700
Java编写Hadoop第一个MapReduce任务
hadoopjavamapreduce大数据json
因为在做网站案例的时候，想学习如何处理海量数据，所以想接触大数据相关的知识，之前对于大数据的了解，仅仅是停留在知道Hadoop，Hive，HBase，Spark的层面上，所以如何学习确实对我造成了困扰，所幸我所在公司部门存在大数据开发，于是不断的请教大佬，大佬说第一步是学一点Linux和Java，然后Hadoop，再然后......。再然后就先不说了，对于Linux和Java我这里可以直接跨过了，然后就是学Hadoop。这周利用散碎的时间，学会了Hadoop的安装使用，使用Java写一个Hadoop任务。安装这里我就不说了，大家可以去网上搜索，或者来我的网站文章模块看我如何安装(Mac): 网址:www.study-java.cn来看一下(需要打开微信小程序：每天学Java进行扫码登录)
每天学Java
2020/06/01
1.5K0
大数据技术之_05_Hadoop学习_04_MapReduce_Hadoop企业优化(重中之重)+HDFS小文件优化方法+MapReduce扩展案例+倒排索引案例(多job串联)+TopN案例+找博客
大数据mapreducewindowshadoopjava
  MapReduce优化方法主要从六个方面考虑：数据输入、Map阶段、Reduce阶段、IO传输、数据倾斜问题和常用的调优参数。
黑泽君
2019/03/08
6150
hadoop案例实现之WordCount （计算单词出现的频数）
打包
一、编写java代码，实现map函数以及reduce函数 package com.paic.elis.test; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apa
MickyInvQ
2020/09/27
7800
hadoop案例实现之WordCount （计算单词出现的频数）
2021年大数据Hadoop（十九）：​​​​​​​MapReduce分区
mapreduce
在 MapReduce 中, 通过我们指定分区, 会将同一个分区的数据发送到同一个Reduce当中进行处理。例如: 为了数据的统计, 可以把一批类似的数据发送到同一个 Reduce 当中, 在同一个 Reduce 当中统计相同类型的数据, 就可以实现类似的数据分区和统计等
Lansonli
2021/10/11
5970
MapReduce工作原理
javascriptmapreducehadoop
这篇文章是我之前在自学 MapReduce 的笔记，想着最近再回顾一下 MapReduce 的原理，于是就整理了一下。
create17
2019/07/19
4.2K0
推荐阅读
编辑精选文章
万字详解高可用架构设计Go 开发者必备：Protocol Buffers 入门指南10分钟带你彻底搞懂分布式链路跟踪
相关讨论
基础网络和私有网络区别？Hadoop是什么？怎么使用呢？hadoop 运行hadoop version失败 ！！急求！！？？？
相关课程
轻量应用构建训练营Python教程-Django框架快速入门到实战TDSQL for PG训练营
大数据：Hadoop-MapReduce练习
4160
MapReduce操作实例
1.6K0
Hadoop: MapReduce2的几个基本示例
1.1K0
Hadoop基础教程-第7章 MapReduce进阶（7.6 MapReduce 二次排序）
2290
几个关于MapReduce的小例子
4770
hadoop_入门1
6340
Hadoop（十五）MapReduce程序实例
1.3K0
MapReduce
4380
MapReduce词频统计【自定义复杂类型、自定义Partitioner、NullWritable使用介绍】
9150
使用Eclipse编译运行MapReduce程序
8640
hadoop系列之MR的经典代码案例一
2.2K0
《快学BigData》--Hadoop总结（G）（40）
4250
Hadoop学习：深入解析MapReduce的大数据魔力（二）
1540
Hadoop学习：深入解析MapReduce的大数据魔力之数据压缩（四）
2400
Hadoop Partitioner使用教程
7700
Java编写Hadoop第一个MapReduce任务
1.5K0
大数据技术之_05_Hadoop学习_04_MapReduce_Hadoop企业优化(重中之重)+HDFS小文件优化方法+MapReduce扩展案例+倒排索引案例(多job串联)+TopN案例+找博客
6150
hadoop案例实现之WordCount （计算单词出现的频数）
7800
2021年大数据Hadoop（十九）：​​​​​​​MapReduce分区
5970
MapReduce工作原理
4.2K0
相关推荐
大数据：Hadoop-MapReduce练习
更多 >
栖西0
LV.0
这个人很懒，什么都没有留下～
文章
21
获赞
72
专栏
1
目录
  • Hadoop
    • 1、求和
    • 2、排序
    • 3、去重
    • 4、分区
    • 5、合并
    • 6、topN
    • 导入依赖
    • 使用
  • Hive
    • 1、常用命令
    • 2、数据操作
    • 3、表操作
  • sqoop
  • scala
  • spark
    • sparkSql
    • sparkStream
  • 附录
    • jQuery关键性代码
    • echarts关键代码
加入讨论
的问答专区 >
程序员晚枫
1宣传大使擅长3个领域
相关课程
一站式学习中心 >
轻量应用构建训练营
2933人在学
云服务器
轻量应用服务器
云计算
Python教程-Django框架快速入门到实战
959人在学
对象存储
python
开源框架
TDSQL for PG训练营
183人在学
分布式数据库 TDSQL
数据库
云数据库 PostgreSQL
领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2025 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 深公网安备号 44030502008569

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号 | 京公网安备号11010802020287

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

Copyright © 2013 - 2025 Tencent Cloud.

All Rights Reserved. 腾讯云 版权所有

登录 后参与评论
5
目录