12、MapReduce框架原理(上)

MapReduce工作流程

流程图如下


mapreduce工作流程.png

解释
上面的流程是整个mapreduce最全工作流程，但是shuffle过程只是从第7步开始到第16步结束，具体shuffle过程详解，如下：

  1. maptask收集我们的map()方法输出的kv对，放到内存缓冲区中
  2. 从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件
  3. 多个溢出文件会被合并成大的溢出文件
  4. 在溢出过程中，及合并的过程中，都要调用partitioner进行分区和针对key进行排序
  5. reducetask根据自己的分区号，去各个maptask机器上取相应的结果分区数据
  6. reducetask会取到同一个分区的来自不同maptask的结果文件，reducetask会将这些文件再进行合并（归并排序）
  7. 合并成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程（从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法）

注意
Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。
缓冲区的大小可以通过参数调整，参数：io.sort.mb 默认100M

InputFormat数据输入

Job提交流程源码

main方法调用job.waitForCompletion(true)方法，waitForCompletion方法又调用了submit方法，提交过程开始了，submit部分代码如下

    //主要是获取客户端和RM进行RPC通信时使用的代理对象
    connect();
    //获取JobSubmitter对象
    final JobSubmitter submitter = 
        getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
    status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
      public JobStatus run() throws IOException, InterruptedException, 
      ClassNotFoundException {
        // 调用JobSubmitter的submitJobInternal方法
        return submitter.submitJobInternal(Job.this, cluster);
      }
    });

接下来看看connect()方法

private synchronized void connect()
          throws IOException, InterruptedException, ClassNotFoundException {
    if (cluster == null) {
      cluster = 
        ugi.doAs(new PrivilegedExceptionAction<Cluster>() {
                   public Cluster run()
                          throws IOException, InterruptedException, 
                                 ClassNotFoundException {
                     return new Cluster(getConfiguration());
                   }
                 });
    }
  }
在这个方法创建提交job的代理，看看具体是怎么工作的
  public Cluster(Configuration conf) throws IOException {
    this(null, conf);
  }

  public Cluster(InetSocketAddress jobTrackAddr, Configuration conf) 
      throws IOException {
    this.conf = conf;
    this.ugi = UserGroupInformation.getCurrentUser();
    initialize(jobTrackAddr, conf);
  }
  
  private void initialize(InetSocketAddress jobTrackAddr, Configuration conf)
      throws IOException {
    /*
      List<ClientProtocolProvider> localProviderList =
              new ArrayList<ClientProtocolProvider>();
      for (ClientProtocolProvider provider : frameworkLoader) {
              localProviderList.add(provider);
            }
    */
    //initProviderList方法执行了上面注释的代码，frameworkLoader中封装了LocalClientProtocolProvider和YarnClientProtocolProvider对象
    initProviderList();
    //providerList中封装了LocalClientProtocolProvider和YarnClientProtocolProvider对象，对应的表示MR程序在本地运行和在YARN上运行
    for (ClientProtocolProvider provider : providerList) {
      ClientProtocol clientProtocol = null;
        if (jobTrackAddr == null) {
          //这里拿到RPC通信使用的本地代理对象
          clientProtocol = provider.create(conf);
        } else {
          //这里拿到RPC通信使用的Yarn代理对象
          clientProtocol = provider.create(jobTrackAddr, conf);
        }

        if (clientProtocol != null) {
          clientProtocolProvider = provider;
          //保存到Cluster中的client成员变量
          client = clientProtocol;
          LOG.debug("Picked " + provider.getClass().getName()
              + " as the ClientProtocolProvider");
          break;
        } 
    }

接下来看看submitter.submitJobInternal方法

  JobStatus submitJobInternal(Job job, Cluster cluster) 
  throws ClassNotFoundException, InterruptedException, IOException {

    //检查作业输出路径是否存在，若存在抛出异常。
    checkSpecs(job);
    // 创建给集群提交数据的Stag路径
    Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
    // 获取jobid ，并创建job路径
    JobID jobId = submitClient.getNewJobID();
    job.setJobID(jobId);
    Path submitJobDir = new Path(jobStagingArea, jobId.toString());
    // 创建job路径，拷贝jar包到集群
    copyAndConfigureFiles(job, submitJobDir);
    Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
    // 计算切片，生成切片规划文件
    int maps = writeSplits(job, submitJobDir);
    conf.setInt(MRJobConfig.NUM_MAPS, maps);
    // 向Stag路径写xml配置文件
    writeConf(conf, submitJobFile);
    //客户端准备就绪，请求RM运行作业
   status = submitClient.submitJob(
          jobId, submitJobDir.toString(), job.getCredentials());
      if (status != null) {
        return status;
      } else {
        throw new IOException("Could not launch job");
      }
    } finally {
      if (status == null) {
        if (jtFs != null && submitJobDir != null)
          // 删除相关文件
          jtFs.delete(submitJobDir, true);
      }
    }
  }

TextInputFormat切片源码

接下来看看int maps = writeSplits(job, submitJobDir)是怎么切片的，调用了writeNewSplits(job, jobSubmitDir)方法

private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,
      Path jobSubmitDir) throws IOException,
      InterruptedException, ClassNotFoundException {
    JobConf jConf = (JobConf)job.getConfiguration();
    int maps;
    if (jConf.getUseNewMapper()) {
      maps = writeNewSplits(job, jobSubmitDir);
    } else {
      maps = writeOldSplits(jConf, jobSubmitDir);
    }
    return maps;
  }

看看writeNewSplits方法，调用了getSplits方法

private <T extends InputSplit>
  int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
      InterruptedException, ClassNotFoundException {
    Configuration conf = job.getConfiguration();
    // 通过反射获取创建InputFormat对象
    InputFormat<?, ?> input =
      ReflectionUtils.newInstance(job.getInputFormatClass(), conf);
    // 调用InputFormat的getSplits方法，默认的InputFormat是
    List<InputSplit> splits = input.getSplits(job);
    T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);
    Arrays.sort(array, new SplitComparator());
    // 将切片信息写到一个切片规划文件中
    JobSplitWriter.createSplitFiles(jobSubmitDir, conf, 
        jobSubmitDir.getFileSystem(conf), array);
    return array.length;
  }

看看getSplits方法

public List<InputSplit> getSplits(JobContext job) throws IOException {
    StopWatch sw = new StopWatch().start();
    /*
    protected long getFormatMinSplitSize() {
      return 1;
    }
    public static long getMinSplitSize(JobContext job) {
       return job.getConfiguration().getLong(SPLIT_MINSIZE, 1L);
    }
    */
    // 最小切片数，getFormatMinSplitSize()值是1，getMinSplitSize(job)值是1
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
    /*
      public static long getMaxSplitSize(JobContext context) {
        return context.getConfiguration().getLong(SPLIT_MAXSIZE, 
                                              Long.MAX_VALUE);
      }
      */
    // 最大切片数，Long的最大值
    long maxSize = getMaxSplitSize(job);

    List<InputSplit> splits = new ArrayList<InputSplit>();
    List<FileStatus> files = listStatus(job);
    // 遍历目录下所有文件
    for (FileStatus file: files) {
      Path path = file.getPath();
      // 获取文件大小
      long length = file.getLen();
      if (length != 0) {
        // 是可以分片的
        if (isSplitable(job, path)) {
          // blockSize 本地模式是32M，集群是128M
          long blockSize = file.getBlockSize();
          /*
            protected long computeSplitSize(long blockSize, long minSize,
                                  long maxSize) {
              return Math.max(minSize, Math.min(maxSize, blockSize));
          }
          */
          // 确定分片大小，默认分片大小和blocksize相等
          long splitSize = computeSplitSize(blockSize, minSize, maxSize);

          long bytesRemaining = length;
          // private static final double SPLIT_SLOP = 1.1;
          // 每次切片时，都要判断切完剩下的部分是否大于块的1.1倍，不大于1.1倍就划分一块切片
          while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
            int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
            splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                        blkLocations[blkIndex].getHosts(),
                        blkLocations[blkIndex].getCachedHosts()));
            bytesRemaining -= splitSize;
          }

          if (bytesRemaining != 0) {
            int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
            splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
                       blkLocations[blkIndex].getHosts(),
                       blkLocations[blkIndex].getCachedHosts()));
          }
        } else { // not splitable
          splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
                      blkLocations[0].getCachedHosts()));
        }
      } else { 
        //Create empty hosts array for zero length files
        splits.add(makeSplit(path, 0, length, new String[0]));
      }
    }
    // Save the number of input files for metrics/loadgen
    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
    sw.stop();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Total # of splits generated by getSplits: " + splits.size()
          + ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
    }
    return splits;
  }

TextInputFormat切片机制

  1. 简单地按照文件的内容长度进行切片
  2. 切片大小，默认等于block大小
  3. 切片时不考虑数据集整体，而是逐个针对每一个文件单独切片
  4. 切片大小确定
    在FileInputFormat中，计算切片大小的逻辑：Math.max(minSize, Math.min(maxSize, blockSize));
    切片主要由这几个值来运算决定
    mapreduce.input.fileinputformat.split.minsize=1 默认值为1
    mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue
    因此，默认情况下，切片大小=blocksize。
    maxsize（切片最大值）：参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值。
    minsize（切片最小值）：参数调的比blockSize大，则可以让切片变得比blocksize还大。
  5. 数据切片只是在逻辑上对输入数据进行分片，并不会再磁盘上将其切分成分片进行存储。InputSplit只记录了分片的元数据信息，比如起始位置、长度以及所在的节点列表等
  6. 提交切片规划文件到yarn上，yarn上的MrAppMaster就可以根据切片规划文件计算开启maptask个数
  7. 获取切片信息API
    // 根据文件类型获取切片信息
    FileSplit inputSplit = (FileSplit) context.getInputSplit();
    // 获取切片的文件名称
    String name = inputSplit.getPath().getName();

CombineTextInputFormat切片机制

关于大量小文件的优化策略
默认情况下TextInputformat对任务的切片机制是按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个maptask，这样如果有大量小文件，就会产生大量的maptask，处理效率极其低下。

  1. 优化策略
    1). 最好的办法，在数据处理系统的最前端（预处理/采集），将小文件先合并成大文件，再上传到HDFS做后续分析。
    2). 补救措施：如果已经是大量小文件在HDFS中了，可以使用另一种InputFormat来做切片（CombineTextInputFormat），它的切片逻辑跟TextFileInputFormat不同：它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个maptask。
    优先满足最小切片大小，不超过最大切片大小
    CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
    CombineTextInputFormat.setMinInputSplitSize(job, 2097152);// 2m
    举例：0.5m+1m+0.3m+5m=2m + 4.8m=2m + 4m + 0.8m

2．具体实现步骤
// 如果不设置InputFormat,它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class)
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
CombineTextInputFormat.setMinInputSplitSize(job, 2097152);// 2m

CombineTextInputFormat使用

使用TextInputFormat运行上篇文章的统计每一个手机号耗费的总上行流量、下行流量、总流量的程序，观察有多少个maptask


输入数据.png

16个maptask.png

修改Driver程序

public class CombineTextFlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(FlowDriver.class);

        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);

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

        job.setInputFormatClass(CombineTextInputFormat.class);
        CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);
        CombineTextInputFormat.setMinInputSplitSize(job, 2097152);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

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

        boolean success = job.waitForCompletion(true);

        System.exit(success ? 0 : 1);
    }
}
1个maptask.png

InputFormat接口实现类

MapReduce任务的输入文件一般是存储在HDFS里面。输入的文件格式包括：基于行的日志文件、二进制格式文件等。这些文件一般会很大，达到数十GB，甚至更大。那么MapReduce是如何读取这些数据的呢？下面我们首先学习InputFormat接口。
InputFormat常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。
下图是InputFormat接口的实现类


InputFormat实现类.png
  1. TextInputFormat是默认的InputFormat。每条记录是一行输入。键是LongWritable类型，存储该行在整个文件中的起始字节偏移量。值是这行的内容，不包括任何行终止符（换行符和回车符）。
  2. 每一行均为一条记录，被分隔符分割为key，value。可以通过在驱动类中设置conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");来设定分隔符。默认分隔符是tab（\t）
  3. 如果使用NlineInputFormat，代表每个map进程处理的InputSplit不再按block块去划分，而是按NlineInputFormat指定的行数N来划分。即输入文件的总行数/N=切片数，如果不整除，切片数=商+1。

KeyValueTextInputFormat使用

需求

统计输入文件中每一行的第一个单词相同的行数。

输入文件
java hello
python hello
java hi
python hi
java hadoop
java hdfs
java mapreduce maptask reducetask
输出文件
java 5
python 2
编码实现

Mapper

public class KeyValueMapper extends Mapper<Text, Text, Text, IntWritable> {
    private IntWritable v = new IntWritable(1);
    @Override
    protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
        context.write(key, v);
    }
}

Reducer

public class KeyValueReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for(IntWritable value: values) {
            sum += value.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

Driver

public class KeyValueDriver {
    public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {
        Configuration configuration = new Configuration();
        configuration.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");
        Job job = Job.getInstance(configuration);

        job.setJarByClass(KeyValueDriver.class);

        job.setMapperClass(KeyValueMapper.class);
        job.setReducerClass(KeyValueReducer.class);

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

        job.setInputFormatClass(KeyValueTextInputFormat.class);

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

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

        boolean success = job.waitForCompletion(true);

        System.exit(success ? 0 : 1);
    }
}

NLineInputFormat使用

需求

对每个单词进行个数统计，要求根据每个输入文件的行数来规定输出多少个切片。此案例要求每五行放入一个切片中。

输入数据
hello java
hello scala
hello hadoop
hello yarn
hello mapreduce
hello hive
hello hbase
hello java
hello scala
hello hadoop
hello yarn
hello mapreduce
hello hive
hello hbase
hello java
hello scala
hello hadoop
hello yarn
hello mapreduce
hello hive
hello hbase
hello java
hello scala
hello hadoop
hello yarn
hello mapreduce
hello hive
hello hbase
代码实现

Maper和Reduce和第10篇文章WordCount代码相同
Driver端做下修改即可

public class WordCountNLineDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        job.setInputFormatClass(NLineInputFormat.class);
        NLineInputFormat.setNumLinesPerSplit(job, 5);

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

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

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

        boolean success = job.waitForCompletion(true);

        System.exit(success ? 0 : 1);
    }
}

结果看起了多少个maptask


nline结果.png

自定义InputFormat

需求

将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value。

分析

  1. 自定义一个类继承FileInputFormat
    1）. 重写isSplitable()方法，返回false不可切割
    2）. 重写createRecordReader()，创建自定义的RecordReader对象，并初始化
  2. 改写RecordReader，实现一次读取一个完整文件封装为KV
  3. 在输出时使用SequenceFileOutPutFormat输出合并文件

实现

  1. 自定义InputFromat
public class TotalInputFormat extends FileInputFormat<NullWritable, BytesWritable> {

    /**
     * 是否可以切片
     * @param context
     * @param filename
     * @return
     */
    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        return false;
    }

    @Override
    public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {

        TotalRecordReader totalRecordReader = new TotalRecordReader();
        totalRecordReader.initialize(split, context);
        return totalRecordReader;
    }
}
  1. 自定义RecordReader
public class TotalRecordReader extends RecordReader<NullWritable, BytesWritable> {

    BytesWritable value = new BytesWritable();

    private Configuration configuration;
    private FileSplit split;

    private boolean processed = false;


    /**
     * 初始化
     * @param split
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {

        this.split = (FileSplit) split;
        this.configuration = context.getConfiguration();
    }

    /**
     * 业务逻辑处理
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {

        if (!processed) {
            // 1 定义缓存区
            byte[] contents = new byte[(int) split.getLength()];

            Path path = split.getPath();
            FileSystem fileSystem = FileSystem.get(configuration);
            FSDataInputStream inputStream = fileSystem.open(path);

            IOUtils.readFully(inputStream, contents, 0, contents.length);
            value.set(contents, 0, contents.length);
            processed = true;
            inputStream.close();
            return true;
        }

        return false;
    }

    @Override
    public NullWritable getCurrentKey() throws IOException, InterruptedException {
        return NullWritable.get();
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return value;
    }

    /**
     * 获取进度
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public float getProgress() throws IOException, InterruptedException {
        return processed ? 0 : 1;
    }

    /**
     * 关闭资源
     * @throws IOException
     */
    @Override
    public void close() throws IOException {

    }
}
  1. Mapper
public class TotalMapper extends Mapper<NullWritable, BytesWritable, Text, BytesWritable> {

    private Text text = new Text();
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
        Path path = fileSplit.getPath();
        text.set(path.toString());
    }

    @Override
    protected void map(NullWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {
        context.write(text, value);
    }
}
  1. Reducer
public class TotalReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> {

    @Override
    protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {
        for(BytesWritable bytesWritable: values) {
            context.write(key, bytesWritable);
        }
    }
}
  1. Driver
public class TotalDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(TotalDriver.class);
        job.setInputFormatClass(TotalInputFormat.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

        job.setMapperClass(TotalMapper.class);
        job.setReducerClass(TotalReducer.class);

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

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BytesWritable.class);

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

        boolean success = job.waitForCompletion(true);

        System.exit(success ? 0 : 1);
    }
}

MapTask工作机制

maptask工作机制.png
  1. Read阶段：MapTask通过用户编写的RecordReader，从输入InputSplit中解析出一个个key/value。
  2. Map阶段：该节点主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。
  3. Collect收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。
  4. Spill阶段：即“溢写”，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。
    溢写阶段详情：
    1）：利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号Partition进行排序，然后按照key进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。
    2）：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。
    3）：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中。
  5. Combine阶段：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。
    当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。
    在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并io.sort.factor（默认10）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。
    让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。

ReduceTask工作机制

ReduceTask工作机制

ReduceTask工作流程.png
  1. Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。
  2. Merge阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。
  3. Sort阶段：按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。
  4. Reduce阶段：reduce()函数将计算结果写到HDFS上。

设置ReduceTask并行度（个数）

ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置

// 默认值是1，手动设置为5
job.setNumReduceTasks(5);

注意事项

  1. ReduceTask=0，表示没有Redce阶段，输出文件个数和Map个数一致
  2. ReduceTask默认值就是1，所以输出文件个数为1个
  3. 如果数据分布不均匀，就有可能在Reduce阶段产生数据倾斜
  4. ReduceTask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有一个ReduceTask
  5. 具体多少个ReduceTask，需要根据集群性能决定
  6. 如果分区数不是1，但是ReduceTask为1，不执行分区过程，因为在MapTask的源码中，执行分区的前提是先判断ReduceNum个数是否大于1，不大于1肯定不执行分区过程。
collector = createSortingCollector(job, reporter);
      partitions = jobContext.getNumReduceTasks();
      if (partitions > 1) {
        partitioner = (org.apache.hadoop.mapreduce.Partitioner<K,V>)
          ReflectionUtils.newInstance(jobContext.getPartitionerClass(), job);
      } else {
        partitioner = new org.apache.hadoop.mapreduce.Partitioner<K,V>() {
          @Override
          public int getPartition(K key, V value, int numPartitions) {
            return partitions - 1;
          }
        };
      }

Shuffle机制

Shuffle机制

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle


shuffle过程.png

Partition分区

问题引出：要求将统计结果按照条件输出到不同文件中（分区）。比如：将统计结果按照手机归属地不同省份输出到不同文件中（分区）

  1. 默认partition分区
public class HashPartitioner<K, V> extends Partitioner<K, V> {
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }
}

默认分区是根据key的hashCode对reduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区。

  1. 自定义Partitioner步骤
    1）自定义类继承Partitioner，重写getPartition()方法
    2）在job驱动中，设置自定义partitioner
    3）自定义partition后，要根据自定义partitioner的逻辑设置相应数量的reduce task
  2. 注意
    如果reduceTask的数量> getPartition的结果数，则会多产生几个空的输出文件part-r-000xx；
    如果1<reduceTask的数量<getPartition的结果数，则有一部分分区数据无处安放，会Exception；
    如果reduceTask的数量=1，则不管mapTask端输出多少个分区文件，最终结果都交给这一个reduceTask，最终也就只会产生一个结果文件 part-r-00000；

Partition分区案例

需求

将统计结果按照手机归属地不同省份输出到不同文件中（分区）

输入数据

1363157985066   13726230503 00-FD-07-A4-72-B8:CMCC  120.196.100.82  i02.c.aliimg.com        24  27  2481    24681   200
1363157995052   13826544101 5C-0E-8B-C7-F1-E0:CMCC  120.197.40.4            4   0   264 0   200
1363157991076   13926435656 20-10-7A-28-CC-0A:CMCC  120.196.100.99          2   4   132 1512    200
1363154400022   13926251106 5C-0E-8B-8B-B1-50:CMCC  120.197.40.4            4   0   240 0   200
1363157993044   18211575961 94-71-AC-CD-E6-18:CMCC-EASY 120.196.100.99  iface.qiyi.com  视频网站    15  12  1527    2106    200
1363157995074   84138413    5C-0E-8B-8C-E8-20:7DaysInn  120.197.40.4    122.72.52.12        20  16  4116    1432    200
1363157993055   13560439658 C4-17-FE-BA-DE-D9:CMCC  120.196.100.99          18  15  1116    954 200
1363157995033   15920133257 5C-0E-8B-C7-BA-20:CMCC  120.197.40.4    sug.so.360.cn   信息安全    20  20  3156    2936    200
1363157983019   13719199419 68-A1-B7-03-07-B1:CMCC-EASY 120.196.100.82          4   0   240 0   200
1363157984041   13660577991 5C-0E-8B-92-5C-20:CMCC-EASY 120.197.40.4    s19.cnzz.com    站点统计    24  9   6960    690 200
1363157973098   15013685858 5C-0E-8B-C7-F7-90:CMCC  120.197.40.4    rank.ie.sogou.com   搜索引擎    28  27  3659    3538    200
1363157986029   15989002119 E8-99-C4-4E-93-E0:CMCC-EASY 120.196.100.99  www.umeng.com   站点统计    3   3   1938    180 200
1363157992093   13560439658 C4-17-FE-BA-DE-D9:CMCC  120.196.100.99          15  9   918 4938    200
1363157986041   13480253104 5C-0E-8B-C7-FC-80:CMCC-EASY 120.197.40.4            3   3   180 180 200
1363157984040   13602846565 5C-0E-8B-8B-B6-00:CMCC  120.197.40.4    2052.flash2-http.qq.com 综合门户    15  12  1938    2910    200
1363157995093   13922314466 00-FD-07-A2-EC-BA:CMCC  120.196.100.82  img.qfc.cn      12  12  3008    3720    200
1363157982040   13502468823 5C-0A-5B-6A-0B-D4:CMCC-EASY 120.196.100.99  y0.ifengimg.com 综合门户    57  102 7335    110349  200
1363157986072   18320173382 84-25-DB-4F-10-1A:CMCC-EASY 120.196.100.99  input.shouji.sogou.com  搜索引擎    21  18  9531    2412    200
1363157990043   13925057413 00-1F-64-E1-E6-9A:CMCC  120.196.100.55  t3.baidu.com    搜索引擎    69  63  11058   48243   200
1363157988072   13760778710 00-FD-07-A4-7B-08:CMCC  120.196.100.82          2   2   120 120 200
1363157985066   13726238888 00-FD-07-A4-72-B8:CMCC  120.196.100.82  i02.c.aliimg.com        24  27  2481    24681   200
1363157993055   13560436666 C4-17-FE-BA-DE-D9:CMCC  120.196.100.99          18  15  1116    954 200

分析

  1. Mapreduce中会将map输出的kv对，按照相同key分组，然后分发给不同的reducetask。默认的分发规则为：根据key的hashcode%reducetask数来分发
  2. 如果要按照我们自己的需求进行分组，则需要改写数据分发（分组）组件Partitioner
    自定义一个CustomPartitioner继承抽象类：Partitioner
  3. 在job驱动中，设置自定义partitioner： job.setPartitionerClass(CustomPartitioner.class)

代码实现

FlowBean类

public class FlowBean implements WritableComparable<FlowBean> {
    private int up;
    private int down;
    private int total;


    public FlowBean() {
    }

    public FlowBean(int up, int down) {
        this.up = up;
        this.down = down;
        this.total = up + down;
    }

    public int getUp() {
        return up;
    }

    public void setUp(int up) {
        this.up = up;
    }

    public int getDown() {
        return down;
    }

    public void setDown(int down) {
        this.down = down;
    }

    public int getTotal() {
        return total;
    }

    public void setTotal(int total) {
        this.total = total;
    }

    public void set(int up, int down) {
        this.up = up;
        this.down = down;
        this.total = up + down;
    }

    @Override
    public String toString() {
        return this.up + "\t" + this.down + "\t" + this.total;
    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeInt(up);
        dataOutput.writeInt(down);
        dataOutput.writeInt(total);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.up = dataInput.readInt();
        this.down = dataInput.readInt();
        this.total = dataInput.readInt();
    }
}

Mapper类

public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {

    private Text ke = new Text();
    FlowBean flowBean = new FlowBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] strings = value.toString().split("\t");
        ke.set(strings[1]);
        flowBean.set(Integer.parseInt(strings[8]), Integer.parseInt(strings[9]));
        context.write(ke, flowBean);
    }
}

Reducer类

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        int totalUp = 0;
        int totalDown = 0;
        for(FlowBean flowBean:values) {
            totalUp += flowBean.getUp();
            totalDown += flowBean.getDown();
        }
        context.write(key, new FlowBean(totalUp, totalDown));
    }
}

分区类

public class FlowPartiton extends Partitioner<Text, FlowBean> {

    @Override
    public int getPartition(Text text, FlowBean flowBean, int numPartitions) {
        if(text.toString().startsWith("135")){
            return 0;
        }else if(text.toString().startsWith("136")){
            return 1;
        }else if(text.toString().startsWith("137")){
            return 2;
        }else if(text.toString().startsWith("138")){
            return 3;
        }else{
            return 4;
        }
    }
}

Driver类

public class UsePartitionFlowDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(FlowDriver.class);

        job.setPartitionerClass(FlowPartiton.class);
        job.setNumReduceTasks(5);

        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);

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

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

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

        boolean success = job.waitForCompletion(true);

        System.exit(success ? 0 : 1);
    }
}

WritableComparable排序

排序是MapReduce框架中最重要的操作之一。Map Task和Reduce Task均会对数据（按照key）进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。
对于Map Task，它会将处理的结果暂时放到一个缓冲区中，当缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次排序，并将这些有序数据写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行一次合并，以将这些文件合并成一个大的有序文件。
对于Reduce Task，它从每个Map Task上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则放到磁盘上，否则放到内存中。如果磁盘上文件数目达到一定阈值，则进行一次合并以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据写到磁盘上。当所有数据拷贝完毕后，Reduce Task统一对内存和磁盘上的所有数据进行一次合并。

排序的分类

  1. 部分排序
    区内排序，环形缓冲区，MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部排序。
  2. 全排序
    最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构
  3. 辅助排序
    GroupingComparator分组，在Reduce端对key进行分组，应用于，在接收key为bean对象时，想让一个或几个字段相同（全部字段比较不相同）的key进入到同一个reduyce方法时，可以采用辅助排序
  4. 二次排序
    在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。bean对象实现WritableComparable接口重写compareTo方法，就可以实现自定义排序

WritableComparable排序案例

需求

对上面分区的案例上，对总流量进行排序

分析

  1. 把程序分两步走，第一步正常统计总流量，第二步再把结果进行排序
  2. context.write(总流量，手机号)
  3. FlowBean实现WritableComparable接口重写compareTo方法

代码实现

修改FlowBean类

public class FlowBean implements WritableComparable<FlowBean> {
    private int up;
    private int down;
    private int total;


    public FlowBean() {
    }

    public FlowBean(int up, int down) {
        this.up = up;
        this.down = down;
        this.total = up + down;
    }

    public int getUp() {
        return up;
    }

    public void setUp(int up) {
        this.up = up;
    }

    public int getDown() {
        return down;
    }

    public void setDown(int down) {
        this.down = down;
    }

    public int getTotal() {
        return total;
    }

    public void setTotal(int total) {
        this.total = total;
    }

    public void set(int up, int down) {
        this.up = up;
        this.down = down;
        this.total = up + down;
    }

    @Override
    public String toString() {
        return this.up + "\t" + this.down + "\t" + this.total;
    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeInt(up);
        dataOutput.writeInt(down);
        dataOutput.writeInt(total);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.up = dataInput.readInt();
        this.down = dataInput.readInt();
        this.total = dataInput.readInt();
    }

    @Override
    public int compareTo(FlowBean o) {
        if(total < o.getTotal()) {
            return  -1;
        }else if(total > o.getTotal()) {
            return 1;
        }else {
            return 0;
        }
    }
}

Combine合并

  1. combiner是MR程序中Mapper和Reducer之外的一种组件。
  2. combiner组件的父类就是Reducer。
  3. combiner和reducer的区别在于运行的位置：
    Combiner是在每一个maptask所在的节点运行;
    Reducer是接收全局所有Mapper的输出结果；
  4. combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量。
  5. combiner能够应用的前提是不能影响最终的业务逻辑，而且，combiner的输出kv应该跟reducer的输入kv类型要对应起来。
    例如求平均值如果使用combine的话
    mapper端
    3 5 7 => (3 + 5 + 7) / 3 = 5
    2 4 => (2 + 4) / 2 = 3
    reduce端
    [(5 + 3) / 2 = 4] != [(3 + 5 + 7 + 2 + 4) / 5 = 4.2]
  6. 自定义Combiner实现步骤
    1）自定义一个combiner继承Reducer，重写reduce方法
    2）在job驱动类中设置

Combine合并案例

需求

wordcount，统计过程中对每一个maptask的输出进行局部汇总，以减小网络传输量即采用Combiner功能

分析

方案一：

  1. 新增加一个WordCountCombine类继承Reducer类
  2. 在WordCountCombine中， 统计单词汇总，将统计结果输出
    方案二：
    将WordCountReducer作为combine在WordCount驱动类中指定

代码实现

WordCountReducer

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int count = 0;
        for(IntWritable value : values) {
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

WordCountMapper

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private Text ke = new Text();
    private IntWritable valu = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] strings = value.toString().split(" ");
        for(String string : strings) {
            ke.set(string);
            context.write(ke, valu);
        }
    }
}

方案一实现
新增WordCountCombine

public class WordCountCombine extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for(IntWritable value : values) {
            sum += value.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

Driver

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        job.setCombinerClass(WordCountCombine.class);

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


        job.setNumReduceTasks(3);

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

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

        boolean success = job.waitForCompletion(true);

        System.exit(success ? 0 : 1);
    }
}

方案二实现
Driver

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(WordCountDriver.class);

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

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


        job.setNumReduceTasks(3);

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

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

        boolean success = job.waitForCompletion(true);

        System.exit(success ? 0 : 1);
    }
}

结果

方案一和方案二运行结果如下


combine未使用前.png

combine使用后.png

方案二比方案一简单

GroupingComparator分组

辅助排序，对reduce阶段的数据根据某一个或几个字段进行分组。

GroupingComparator分组案例

需求

统计每一个订单中最贵的商品

输入数据

0000001 Pdt_01  222.8
0000002 Pdt_06  722.4
0000001 Pdt_05  25.8
0000003 Pdt_01  222.8
0000003 Pdt_01  33.8
0000002 Pdt_03  522.8
0000002 Pdt_04  122.4

分析

  1. 利用“订单id和成交金额”作为key，可以将map阶段读取到的所有订单数据按照id分区，按照金额排序，发送到reduce。
  2. 在reduce端利用groupingcomparator将订单id相同的kv聚合成组，然后取第一个即是最大值

代码实现

OrderBean

public class OrderBean implements WritableComparable<OrderBean> {

    private String orderId;
    private String productId;
    private Double price;


    public OrderBean() {
    }

    public void set(String orderId, String productId, Double price) {
        this.orderId = orderId;
        this.productId = productId;
        this.price = price;
    }

    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public String getProductId() {
        return productId;
    }

    public void setProductId(String productId) {
        this.productId = productId;
    }

    public Double getPrice() {
        return price;
    }

    public void setPrice(Double price) {
        this.price = price;
    }

    @Override
    public int compareTo(OrderBean o) {
        if(orderId.compareTo(o.getOrderId()) == 0) {
            return -price.compareTo(o.getPrice());
        }else {
            return orderId.compareTo(o.getOrderId());
        }
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(this.orderId);
        out.writeUTF(this.productId);
        out.writeDouble(this.price);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.orderId = in.readUTF();
        this.productId = in.readUTF();
        this.price = in.readDouble();
    }

    @Override
    public String toString() {
        return "orderId='" + orderId + '\'' +
                ", productId='" + productId + '\'' +
                ", price=" + price;
    }
}

OrderMapper

public class OrderMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {

    private OrderBean orderBean = new OrderBean();
    private NullWritable val = NullWritable.get();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] strings = value.toString().split("\t");
        orderBean.set(strings[0], strings[1], Double.parseDouble(strings[2]));
        context.write(orderBean, val);

    }
}

OrderReducer

public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> {

    @Override
    protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        for(NullWritable value: values) {
            context.write(key, NullWritable.get());
        }
    }
}

OrderPartitioner

public class OrderPartitioner extends Partitioner<OrderBean, NullWritable> {
    @Override
    public int getPartition(OrderBean orderBean, NullWritable nullWritable, int numPartitions) {
        return (orderBean.getOrderId().hashCode() & Integer.MAX_VALUE) % numPartitions;
    }
}

OrderGrouping

public class OrderGrouping extends WritableComparator {
    
    protected OrderGrouping() {
        super(OrderBean.class, true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        OrderBean oa = (OrderBean) a;
        OrderBean ob = (OrderBean) b;
        return oa.getOrderId().compareTo(ob.getOrderId());
    }
}

OrderDriver

public class OrderDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Job job = Job.getInstance(new Configuration());

        job.setJarByClass(OrderDriver.class);

        job.setReducerClass(OrderReducer.class);
        job.setMapperClass(OrderMapper.class);

        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(NullWritable.class);
        
        job.setOutputKeyClass(OrderBean.class);
        job.setOutputValueClass(NullWritable.class);

        job.setPartitionerClass(OrderPartitioner.class);
        job.setGroupingComparatorClass(OrderGrouping.class);
        job.setNumReduceTasks(2);

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

        boolean isSuccess = job.waitForCompletion(true);
    }
}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 203,324评论 5赞 476
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,303评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,192评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,555评论 1赞 273
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,569评论 5赞 365
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,566评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,927评论 3赞 395
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,583评论 0赞 257
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,827评论 1赞 297
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,590评论 2赞 320
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,669评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,365评论 4赞 318
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,941评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,928评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,159评论 1赞 259
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,880评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,399评论 2赞 342

推荐阅读更多精彩内容