package com.company.sparkcore

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

object CountPVByGroup {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName(CountPVByGroup.getClass.getSimpleName)
      .setMaster("local")
    Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
    Logger.getLogger("org.apache.hadoop").setLevel(Level.OFF)
    val sc = new SparkContext(conf)
    val lines = sc.textFile("file:///e:/pv_users.txt")
    //拼接成（1_25,1）的形式
    val newKeyValue = lines.map(_.split(",")).map(pvdata => ((pvdata(0)+ "_" + pvdata(1)),1))
   //对上述KV进行统计
    val pvcount = newKeyValue.reduceByKey(_ + _)
    //将拼接符号去掉，组合成形如(1,25,1)的形式
    val pvid_age_count = pvcount.map(newkv => (newkv._1.split("_")(0),newkv._1.split("_")(1),newkv._2))
    //结果输出
// (1,25,1)
// (2,25,2)
// (1,32,1)
    pvid_age_count.collect().foreach(println)
  }

}

单机安装伪hadoop集群
见：https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/SingleCluster.html
注：在Mac中安装遇到了一些问题 但是google一下就能解决 恕不一一道来

思考题解答步骤
cat pv_users
1,25
2,25
1,32
2,25

# 导入该文件到dfs中
bin/hdfs dfs -put pv_users pv_users

# 因为每一行只有pageid, age并且中间没有空格 可以直接利用hadoop自带的wordcount程序
# 读取dfs中的pv_user文件 进行统计 然后将结果输出到pv_users_count中
bin/yarn jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.9.1.jar wordcount pv_users pv_users_count
# 读取统计结果
bin/hdfs dfs -cat pv_users_count/part-r-00000
1,25    1
1,32    1
2,25    2

不能光想，必须动手实践：
## Mapper
public class PageMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String data = value.toString();
        String[] words = data.split("\n");
        for (String word : words) {
            context.write(new Text(word), new IntWritable(1));
        }
    }
}
## Reducer
public class PageReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int total=0;
        for (IntWritable value : values) {
            total=total+value.get();
        }
        context.write(key, new IntWritable(total));
    }
}
## Main
public class PageMain {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance();
        job.setJarByClass(PageMain.class);

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

        job.setReducerClass(PageReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

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

        job.waitForCompletion(true);


    }
}

## 命令行
hadoop jar page-1.0-SNAPSHOT.jar PageMain /input/page /output5
ps：第一次运行报错了~~（不练不知道）
错误：Initialization of all the collectors failed. Error in last collector was :interface javax.xml.soap.
原因：编写Main的时候，Text的引用import错了，习惯了弹出提示直接确定~应该导入`import org.apache.hadoop.io.Text;`