入门Hadoop存储与计算：实现单词统计的分布式文件处理系统

 #添加用户
 sudo useradd hadoop
 
 #设置密码
 sudo passwd hadoop
 
 #切换用户
 su hadoop

#安装
yum install openssh
 
#需要密码校验
ssh localhost
 
#登陆成功后退出 开始配置免密登陆
exit
 
cd  ~/.ssh

#生成密钥 回车几下
ssh-keygen -t rsa

#添加
cat ./id_rsa.pub >> ./authorized_keys
 
#确保有权限
chmod 700 ~/.ssh
chmod 600 ~/.ssh/authorized_keys
 
#再次登陆不需要密码
ssh localhost

#更新包索引
sudo yum update -y
 
#安装JDK
sudo yum install java-1.8.0-openjdk-devel -y

#找到JDK目录  JDK通常在/usr/lib下 
#/usr/lib/jvm
pwd
 
#目录名太长改成jdk8
mv java-1.8.0-openjdk-1.8.0.432.b06-2.oc8.x86_64/ jdk8
 
#配置环境变量 在末尾追加环境变量
vim  ~/.bashrc
export JAVA_HOME=/usr/lib/jvm/jdk8
export PATH=$JAVA_HOME/bin:$PATH
 
#环境变量生效
source ~/.bashrc
 
#查看版本号 判断是否安装成功
java -version

 #解压
 sudo tar -zxf hadoop-3.4.1.tar.gz
 
 #进入目录
 cd hadoop-3.4.1
 
 #查看版本 如果没找到JDK说明 JDK环境变量配的有问题
 ./bin/hadoop version

 #末尾追加 我的Hadoop目录是:/home/lighthouse/hadoop-3.4.1
 vim ~/.bashrc  
 
 export HADOOP_HOME=/home/lighthouse/hadoop-3.4.1
 export HADOOP_MAPRED_HOME=$HADOOP_HOME
 export HADOOP_COMMON_HOME=$HADOOP_HOME
 export HADOOP_HDFS_HOME=$HADOOP_HOME
 export YARN_HOME=$HADOOP_HOME
 export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
 export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib/native"
 
 #这里也有JAVA的环境变量
 export PATH=${JAVA_HOME}/bin:${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

 <configuration>
     <!-- 临时文件 -->
     <property>
         <name>hadoop.tmp.dir</name>
         <value>file:/home/lighthouse/hadoop-data/tmp</value>
         <description>Abase for other temporary directories.</description>
     </property>
     <!--hdfs地址-->
     <property>
         <name>fs.defaultFS</name>
         <value>hdfs://localhost:9000</value>
     </property>
 </configuration>

 <configuration>
         <!-- 副本数量 -->
         <property>
                 <name>dfs.replication</name>
                 <value>3</value>
         </property>
         <!-- namenode 元数据 存储目录 -->
         <property>
                 <name>dfs.namenode.name.dir</name>
                 <value>file:/home/lighthouse/hadoop-data/tmp/dfs/name</value>
         </property>
         <!-- datanode 真正数据 存储目录 -->
         <property>
                 <name>dfs.datanode.data.dir</name>
                 <value>file:/home/lighthouse/hadoop-data/tmp/dfs/data</value>
         </property>
 </configuration>

exportJAVA_HOME=/usr/1ib/jvm/jdk8

<dependencies>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>3.4.1</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>3.4.1</version>
  </dependency>
</dependencies>

	/**
     * 实现Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
     */
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        /**
         * map 转换为 <单词,1>
         *
         * @param key     输入KEY
         * @param value   输入VALUE
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException 		  {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                //输出 <单词,1> 比如 <CaiCai,1> 表示CaiCai出现1次
                context.write(word, one);
            }
        }
    }

	/**
     * 实现Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
     */
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            //统计次数
            int sum = 0;

            //经过Shuffle 会对相同key进行分区 比如<CaiCai,2> <CaiCai,1> <CaiCai,3>在一个分区
            //values就是次数的集合 <2,1,3> 分别出现2、1、3次，累加即可
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            //累加次数后输出 <>
            context.write(key, result);
        }
    }

package com.caicaijava.hadoop;

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.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 统计单词出现的次数
 */
public class WordCount {

    /**
     * 实现Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
     */
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        /**
         * map 转换为 <单词,1>
         *
         * @param key     输入KEY
         * @param value   输入VALUE
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                //输出 <单词,1> 比如 <CaiCai,1> 表示CaiCai出现1次
                context.write(word, one);
            }
        }
    }

    /**
     * 实现Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
     */
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            //统计次数
            int sum = 0;

            //经过Shuffle 会对相同key进行分区 比如<CaiCai,2> <CaiCai,1> <CaiCai,3>在一个分区
            //values就是次数的集合 <2,1,3> 分别出现2、1、3次，累加即可
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            //累加次数后输出 <>
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //第一个参数为输入地址
        FileInputFormat.addInputPath(job, new Path(args[0]));
        //第二个参数为输出地址
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //等待任务执行完成结束
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

#!/bin/bash

# 检查参数数量
if [ "$#" -ne 3 ]; then
    echo "Usage: $0 <word> <number_of_words> <output_file>"
    exit 1
fi

# 获取参数
word=$1
num_words=$2
output_file=$3

# 检查输出文件是否存在
if [ ! -f "$output_file" ]; then
    touch "$output_file"
fi

# 生成指定数量的单词并追加到文件
generated_words=""
for (( i=1; i<=num_words; i++ ))
do
    generated_words+="$word "
done

# 去掉最后一个多余的空格
generated_words=${generated_words% }

# 追加到输出文件
echo "$generated_words" >> "$output_file"

echo "Generated $num_words instances of '$word' and appended to $output_file"

chmod 777 gen.sh 
./gen.sh Hello 999 input.txt
#打印 Generated 999 instances of 'Hello' and appended to input.txt
./gen.sh CaiCai 999 input.txt
#打印 Generated 999 instances of 'CaiCai' and appended to input.txt

hadoop fs -mkdir /test
hadoop fs -mkdir /test/input
hadoop fs -put input.txt /test/input/hellocaicai.txt

hadoop jar HadoopDemo-1.0-SNAPSHOT.jar com.caicaijava.hadoop.WordCount /test/input/hellocaicai.txt /test/output/hellocaicai

hadoop fs -cat /test/output/hellocaicai/part-r-00000