Hadoop: word count,并将reduce结果写入ES

一、依赖，其中ES版本为7.6.2

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>TestHadoop3</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.2.1</version>
        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.3</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.47</version>
        </dependency>

        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.6</version>
        </dependency>


        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.12.3</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.6.2</version>
        </dependency>
    </dependencies>
    
</project>

二、定义表示word count 结果的实体类

package cn.edu.tju;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.lib.db.DBWritable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

public class MyDBWritable implements DBWritable, Writable {
    private String word;
    private int count;

    public String getWord() {
        return word;
    }

    public void setWord(String word) {
        this.word = word;
    }

    public int getCount() {
        return count;
    }

    public void setCount(int count) {
        this.count = count;
    }

    @Override
    public void write(PreparedStatement statement) throws SQLException {

        statement.setString(1, word);
        statement.setInt(2, count);
    }

    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        word = resultSet.getString(1);
        count = resultSet.getInt(2);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(word);
        out.writeInt(count);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        word = in.readUTF();
        count = in.readInt();
    }
}

三、定义mapper和reducer

package cn.edu.tju;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private IntWritable one = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String valueString = value.toString();
        String[] values = valueString.split(" ");
        for (String val : values) {
            context.write(new Text(val), one);
        }
    }
}


package cn.edu.tju;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        Iterator<IntWritable> iterator = values.iterator();
        int count = 0;
        while (iterator.hasNext()) {
            count += iterator.next().get();
        }

        context.write(key, new IntWritable(count));
    }
}

四、自定义类，实现接口OutputFormat

package cn.edu.tju;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;


import java.io.IOException;

public class MyOutputFormat extends OutputFormat<Text, IntWritable> {

    @Override
    public RecordWriter<Text, IntWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new MyRecordWriter();
    }

    @Override
    public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {

    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
        NullOutputFormat<NullWritable, NullWritable> format = new NullOutputFormat<>();

        OutputCommitter outputCommitter = format.getOutputCommitter(context);
        return outputCommitter;

    }
}

其中getRecordWriter方法返回了一个RecordWriter类的对象。自定义的RecordWriter类如下：

package cn.edu.tju;

import com.google.gson.Gson;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;

import java.io.IOException;
import java.util.Random;
import java.util.UUID;

public class MyRecordWriter extends RecordWriter<Text, IntWritable> {
    private RestHighLevelClient client;
    public MyRecordWriter() {
        client = new RestHighLevelClient(
                RestClient.builder(HttpHost.create("xx.xx.xx.xx:9200")));
    }

    @Override
    public void write(Text key, IntWritable value) throws IOException, InterruptedException {
        IndexRequest indexRequest = new IndexRequest("demo6");
        indexRequest.id(UUID.randomUUID().toString());

        MyDBWritable myDBWritable = new MyDBWritable();
        myDBWritable.setWord(key.toString());
        myDBWritable.setCount(value.get());
        Gson gson = new Gson();
        String s = gson.toJson(myDBWritable);

        indexRequest.source(s, XContentType.JSON);
        IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
        System.out.println(indexResponse);

    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        client.close();
    }
}

其中构造方法创建了一个ES 客户端对象。write方法用来把数据写入ES.

五、定义主类

package cn.edu.tju;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class MyWordCount6 {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration(true);

        //hdfs路径
        configuration.set("fs.defaultFS", "hdfs://xx.xx.xx.xx:9000/");
        //local 运行
        configuration.set("mapreduce.framework.name", "local");
        configuration.set("yarn.resourcemanager.hostname","xx.xx.xx.xx");
        //configuration.set("mapreduce.job.jar","target\\TestHadoop3-1.0-SNAPSHOT.jar");
        //job 创建
        Job job = Job.getInstance(configuration);
        //
        job.setJarByClass(MyWordCount.class);
        //job name
        job.setJobName("wordcount-" + System.currentTimeMillis());

        //输入数据路径
        FileInputFormat.setInputPaths(job, new Path("/user/root/tju/"));

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setMapperClass(WordCountMapper.class);
        //job.setCombinerClass(MyCombiner.class);
        job.setReducerClass(WordCountReducer.class);

        job.setOutputFormatClass(MyOutputFormat.class);




        //等待任务执行完成
        job.waitForCompletion(true);
    }
}

其中 job.setOutputFormatClass(MyOutputFormat.class);用来指定reduce的结果写到哪里。输入数据来自hdfs
六、任务执行后在ES查询数据
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/516563.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

[StartingPoint][Tier0]Mongod

[StartingPoint][Tier0]Mongod

Task 1 How many TCP ports are open on the machine? (机器上打开了多少个 TCP 端口&#xff1f;) Example: $ sudo nmap -sS -T4 10.129.222.112 -p 27017,22 2 Task 2 Which service is running on port 27017 of the remote host? (哪个服务正在远程主机的端口 270…
阅读更多...
Hippo4j线程池实现技术

Hippo4j线程池实现技术

文章目录 &#x1f50a;博主介绍&#x1f964;本文内容部署运行模式集成线程池监控配置参数默认配置 &#x1f4e2;文章总结&#x1f4e5;博主目标 &#x1f50a;博主介绍 &#x1f31f;我是廖志伟&#xff0c;一名Java开发工程师、Java领域优质创作者、CSDN博客专家、51CTO专家…
阅读更多...
基于Java课程选课系统设计与实现（源码+部署文档）

基于Java课程选课系统设计与实现（源码+部署文档）

博主介绍&#xff1a; ✌至今服务客户已经1000、专注于Java技术领域、项目定制、技术答疑、开发工具、毕业项目实战 ✌ &#x1f345; 文末获取源码联系 &#x1f345; &#x1f447;&#x1f3fb; 精彩专栏 推荐订阅 &#x1f447;&#x1f3fb; 不然下次找不到 Java项目精品实…
阅读更多...
KIl5：Stm32L071下载出现flash download faild “cortex-m0+“的解决方法

KIl5：Stm32L071下载出现flash download faild “cortex-m0+“的解决方法

首先看看有没有芯片&#xff0c;没有芯片下载一下 下载并在device选择对应的芯片 选择调试器 选择flash
阅读更多...
gpt4.0中文版

gpt4.0中文版

我愿把这个网站成为全球最强AI网站&#xff01;弄100多个AI伺候你&#xff1f;&#xff1f; 家人们&#xff0c;你们猜我发现了什么牛逼的AI网站&#xff1f;&#xff1f; 直接上图&#xff1a; 编辑 这个网站&#xff0c;聚合了国内外100多个顶尖的AI&#xff0c;包括了Op…
阅读更多...
入门用Hive构建数据仓库

入门用Hive构建数据仓库

在当今数据爆炸的时代&#xff0c;构建高效的数据仓库是企业实现数据驱动决策的关键。Apache Hive 是一个基于 Hadoop 的数据仓库工具&#xff0c;可以轻松地进行数据存储、查询和分析。本文将介绍什么是 Hive、为什么选择 Hive 构建数据仓库、如何搭建 Hive 环境以及如何在 Hi…
阅读更多...
我认识的Git-史上最强的版本控制系统

我认识的Git-史上最强的版本控制系统

大家好&#xff01; 欢迎大家来一起交流Git使用心得&#xff0c;相信很多同事对Git都很熟悉了&#xff0c;如果下面说的有错误的“知识点”&#xff0c;欢迎批评指正。 初识Git 我认识Git已经很多年了&#xff08;我在有道云笔记里面“Git”文件夹的创建时间是&#xff1a; …
阅读更多...
sky06笔记下

sky06笔记下

1.边沿检测 检测输入信号din的上升沿&#xff0c;并输出pulse module edge_check ( clk, rstn, din, pulse ); input wire clk,rstn; input wire din; output reg pulse;wire din_dly;always (posedge clk or negedge rstn)beginif(!rstn)din_dly < 1b0;elsedin_dly < d…
阅读更多...
JS-11A/11时间继电器 板前接线 JOSEF约瑟

JS-11A/11时间继电器 板前接线 JOSEF约瑟

系列型号&#xff1a; JS-11A/11集成电路时间继电器&#xff1b;JS-11A/12集成电路时间继电器&#xff1b; JS-11A/13集成电路时间继电器&#xff1b;JS-11A/136集成电路时间继电器&#xff1b; JS-11A/137集成电路时间继电器&#xff1b;JS-11A/22集成电路时间继电器&#…
阅读更多...
【C语言】_文件内容操作：随机读写

【C语言】_文件内容操作：随机读写

目录 1. fseek 1.1 随机读文件 1.2 随机写文件 2. ftell 3. rewind 当以读方式打开一个存在且存有内容的文件时&#xff0c;文件指针会默认指向第一个元素。以在test4.txt文件中存储abcdef为例&#xff1a; int main() {//打开文件FILE* pf fopen("E:\\C_文件操作…
阅读更多...
数学知识--（质数，约数）

数学知识--（质数，约数）

本文用于个人算法竞赛学习&#xff0c;仅供参考 目录 一.质数的判定 二.分解质因数 三.质数筛 1.朴素筛法 2.埃氏筛法 3.线性筛法 四.约数 1.求一个数的所有约数 2.约数个数和约数之和 3.欧几里得算法&#xff08;辗转相除法&#xff09;-- 求最大公约数 一.质数的判定 …
阅读更多...
JWT--study

JWT--study

JWT 1、简介 2、结构 头部 载荷 签证 应用 <dependency><groupId>io.jsonwebtoken</groupId><artifactId>jjwt</artifactId><version>0.9.1</version> </dependency>Token 生成 解析token package com.wang.utils;import io.…
阅读更多...
【QT+QGIS跨平台编译】056：【pdal_lepcc+Qt跨平台编译】（一套代码、一套框架，跨平台编译）

【QT+QGIS跨平台编译】056：【pdal_lepcc+Qt跨平台编译】（一套代码、一套框架，跨平台编译）

点击查看专栏目录 文章目录 一、pdal_lepcc介绍二、pdal下载三、文件分析四、pro文件五、编译实践一、pdal_lepcc介绍 pdal_lepcc 是 PDAL（Point Data Abstraction Library）的一个插件，用于点云数据的压缩。它基于 EPCC（Entwine Point Cloud Compression）算法，提供了对点…
阅读更多...
Linux-4 gcc和makefile

Linux-4 gcc和makefile

Linux编译器-gcc/g使用 1.设计样例 c语言&#xff1a;linux中用的stdc99版本--可能会出现其他问题 c&#xff1a;Linux中用的stdc11--使用c11版本 Linux没有文件格式的区分&#xff0c;但是编译器区分 gcc编译器的文件格式是filename.c g编译器的文件格式是filename.cc或者fil…
阅读更多...
利用Spark将Kafka数据流写入HDFS

利用Spark将Kafka数据流写入HDFS

利用Spark将Kafka数据流写入HDFS 在当今的大数据时代&#xff0c;实时数据处理和分析变得越来越重要。Apache Kafka作为一个分布式流处理平台&#xff0c;已经成为处理实时数据的事实标准。而Apache Spark则是一个强大的大数据处理框架&#xff0c;它提供了对数据进行复杂处理…
阅读更多...
可行性研究报告模板（套用）

可行性研究报告模板（套用）

1业务需求可行性分析 2技术可行性分析 2.1规范化原则 2.2高度的兼容性和可移植性 2.3人性化、适用性 2.4标准化统一设计原则 2.5先进安全可扩展性原则 3开发周期可行性分析 4人力资源可行性分析 5成本分析 6收益分析 7结论 所有资料获取进主页或本文末个人名片直接…
阅读更多...
【OSTEP】并发：线程与多线程

【OSTEP】并发：线程与多线程

" A flow of control within a process that consists of a PC, a register set and a stack space" 本章将介绍为单个运行进程提供的新抽象 —— 线程 (thread) 线程是 调度的一个基本单位&#xff08;basic unit of CPU scheduling&#xff09;一个单独的线程至…
阅读更多...
RUST语言函数的定义与调用

RUST语言函数的定义与调用

1.定义函数 定义一个RUST函数使用fn关键字 函数定义语法: fn 函数名(参数名:参数类型,参数名:参数类型) -> 返回类型 { //函数体 } 定义一个没有参数,没有返回类型的参数 fn add() {println!("调用了add函数!"); } 定义有一个参数的函数 fn add(a:u32)…
阅读更多...
力扣热题100_链表_21_合并两个有序链表

力扣热题100_链表_21_合并两个有序链表

文章目录 题目链接解题思路解题代码 题目链接 21. 合并两个有序链表 将两个升序链表合并为一个新的 升序 链表并返回。新链表是通过拼接给定的两个链表的所有节点组成的。 示例 1&#xff1a; 输入&#xff1a;l1 [1,2,4], l2 [1,3,4] 输出&#xff1a;[1,1,2,3,4,4] 示例…
阅读更多...
InternLM2-lesson2作业

InternLM2-lesson2作业

书生浦语大模型趣味 Demo 视频连接&#xff1a;https://www.bilibili.com/video/BV1AH4y1H78d/?vd_source902e3124d4683c41b103f1d1322401fa 目录 书生浦语大模型趣味 Demo一、基础作业二、进阶作业 一、基础作业 第一次执行&#xff1a; 第二次执行&#xff1a; 第一次执…
阅读更多...
最新文章

Copyright @ 2022~2023