账号密码登录
微信安全登录
微信扫描二维码登录

登录后绑定QQ、微信即可实现信息互通

手机验证码登录
找回密码返回
邮箱找回 手机找回
注册账号返回
其他登录方式
分享
  • 收藏
    X
    hadoop streaming中shuffle过程问题
    51
    0

    hadoop streaming map reduce 输入输出疑问

        hadoop 中在map和reduce中间会有一个shuffle过程。所以正常来说map输出reduce的输入应该如下:

    # map输出
    2018-02-11    1
    2018-02-11    0
    2018-02-11    4
    2018-02-11    0
    2018-02-11    0
    
    # reduce输入
    2018-02-11 [1,0,4,0,0]
    但是实际过程中发现,map的输入和reduce的输出是相同的。

    hadoop使用如下:

    ${hadoop_bin} streaming \
        -input ${hdfs_input} \
        -output ${hdfs_output} \
        -mapper "lib/python/bin/pythonmapper.py" \
        -reducer "cat" \
        -file ${WORK_DIR}/mapper.py \
        -cacheArchive ${python_lib}\#lib \
        -jobconf mapred.job.name="${jobname}" \
        -inputformat org.apache.hadoop.mapred.TextInputFormat \
        -outputformat org.apache.hadoop.mapred.TextOutputFormat \

    请问:

    • hadoop streaming中的shuffle过程 是否真的没有merge的过程?
    • 有merge的shuffle过程,是否只在非streaming的hadoop中存在?
    • hadoop streaming中shuffle过程到底是做什么?

       希望有人能帮我解答一下,多谢

    0
    打赏
    收藏
    点击回答
        全部回答
    • 0
    • 泼熄这一秒 普通会员 1楼

      在Hadoop Streaming中,shuffle过程主要是将输入数据文件按照某种方式(如随机顺序、键值对等)分组,然后将这些分组的数据进行处理。

      当使用Hadoop Streaming进行数据处理时,通常需要将数据文件加载到Hadoop的MapReduce集群上,然后使用MapReduce进行数据处理。在这个过程中,shuffle过程是非常重要的一步。

      以下是一个简单的示例,说明了如何在Hadoop Streaming中使用shuffle过程:

      ```java import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; 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; import org.apache.hadoop.streaming.api.datastream.DataStream; import org.apache.hadoop.streaming.api.environment.StreamExecutionEnvironment;

      public class WordCount {

      public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
      
          private final static IntWritable one = new IntWritable(1);
          private Text word = new Text();
      
          public void map(LongWritable key, Text value, Context context) throws Exception {
              StringTokenizer itr = new StringTokenizer(value.toString());
              while (itr.hasMoreTokens()) {
                  word.set(itr.nextToken());
                  context.write(word, one);
              }
          }
      }
      
      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 Exception {
              int sum = 0;
              for (IntWritable value : values) {
                  sum += value.get();
              }
              result.set(sum);
              context.write(key, result);
          }
      }
      
      public static void main(String[] args) throws Exception {
          // 创建Hadoop环境
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      
          // 设置输入和输出文件路径
          env.setJarByClass(WordCount.class);
          env.setMapperClass(TokenizerMapper.class);
          env.setCombinerClass(IntSumReducer.class);
          env.setReducerClass(IntSumReducer.class);
      
          // 设置输入文件路径
          env.addInputPath("input.txt", "data");
      
          // 设置输出文件路径
          env.setOutputPath("output.txt", "output");
      
          // 执行MapReduce任务
          env.execute("wordCount");
      }
      

      } ```

      在这个示例中,我们首先定义了两个Mapper类,分别用于处理输入数据和计算结果。然后,我们定义了一个Reducer类,用于将结果进行合并。

      在main方法中,我们首先创建了一个Hadoop环境,然后设置了输入和输出文件路径,然后设置了Mapper和Reducer的类,最后执行了MapReduce任务。

    更多回答
    扫一扫访问手机版
    • 回到顶部
    • 回到顶部