现在的位置: 首页 > 搜索技术 > 黄专家专栏 > 正文

Hadoop I/O 上 SequenceFile 类在 Pipes 上的应用

2014年10月30日 搜索技术, 黄专家专栏 ⁄ 共 3830字 ⁄ 字号 评论关闭

秒速赛车公式 www.l19l7.cn 有的时候,我们在 hadoop 上的输入可能不是一些基于行的文本,是希望自定义一些结构化的数据。这种情况,一般会选用工具将结构化的数据序列化成字节流,存储在磁盘上。然后在 maper 中读取进来,反序列化即可得到原来的数据。

我们使用 google protobuf 作为这种结构化的信息传递的工具。

首先可以先定义 person.proto 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
option java_package = "com.hackerlight.proto";
option java_outer_classname = "ProtobufMessage";

message Person {
  required string name = 1;
  required int32 id = 2;
  optional string email = 3;

  enum PhoneType {
    MOBILE = 0;
    HOME = 1;
    WORK = 2;
  }

  message PhoneNumber {
    required string number = 1;
    optional PhoneType type = 2 [default = HOME];
  }

  repeated PhoneNumber phone = 4;

  message CountryInfo {
          required string name = 1;
          required string code = 2;
          optional int32 number = 3;
  }
}

生成 protobuf 的 java 代码

1
protoc --java_out=./ test.proto

java 代码会生成在 当前目录下的 com/hackerlight/proto 下.

生成 SequenceFile 结构的代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
package com.hackerlight.writer;

import java.io.FileOutputStream;
import java.net.URI;

import com.hackerlight.proto.ProtobufMessage;
import com.hackerlight.proto.ProtobufMessage.Person;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;

public class Writer {
public static void main(String[] args) throws Exception {
  String uri = args[0];

  org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
  FileSystem fs = FileSystem.get(URI.create(uri), conf);
  Path path = new Path(uri);

  Person.Builder builder = Person.newBuilder();
  builder.setEmail("[email protected]");
  builder.setId(1024);
  builder.setName("Moon");

  Person.PhoneNumber.Builder p = Person.PhoneNumber.newBuilder();
  p.setNumber("18900000000");
  builder.addPhone(p);

  byte[] serialize_string = builder.build().toByteArray();
  IntWritable key = new IntWritable();
  BytesWritable value = new BytesWritable();
  SequenceFile.Writer write = null;
  try {
    write = SequenceFile.createWriter(fs, conf, path, key.getClass(), value.getClass());
    for (int i = 0; i < 100; ++i) {
      key.set(i);
      value.set(serialize_string, 0, serialize_string.length);
      write.append(key, value);
    }
  } finally {
    IOUtils.closeStream(write);
  }
}
}

可以写一个读取的代码反序列化写入的结构数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
package com.hackerlight.reader;

import java.io.FileOutputStream;
import java.net.URI;

import com.hackerlight.proto.ProtobufMessage;
import com.hackerlight.proto.ProtobufMessage.Person;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;

public class Reader {
public static void main(String[] args) throws Exception {
  String uri = args[0];
  org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
  FileSystem fs = FileSystem.get(URI.create(uri), conf);
  Path path = new Path(uri);
  SequenceFile.Reader reader = null;
  try {
    reader = new SequenceFile.Reader(fs, path, conf);
    Writable key = (Writable)ReflectionUtils.newInstance(reader.getKeyClass(), conf);
    BytesWritable value = (BytesWritable)ReflectionUtils.newInstance(reader.getValueClass(), conf);
    while (reader.next(key, value)) {
      System.out.println("key : " + key);
      Person new_person = Person.PARSER.parseFrom(value.getBytes(), 0, value.getLength());
      System.out.println("name  : " + new_person.getName());
      System.out.println("email : " + new_person.getEmail());
    }
  } finally {
    IOUtils.closeStream(reader);
  }
}
}

在 Pipes 上可以这样读取

1
2
3
4
5
6
7
8
9
10
class MyMap: public HadoopPipes::Mapper {
 public:
   WordCountMap(HadoopPipes::TaskContext& context) {}
   void map(HadoopPipes::MapContext& context) {
     const std::string& key = context.getInputKey();
     const std::string& value = context.getInputValue();
     Person person;
     assert(person.ParseFromString(value));
   }
};

执行 hadoop 命令的时候一定要记得加上参数 -inputformat org.apache.hadoop.mapred.SequenceFileInputFormat

抱歉!评论已关闭.

  • 倒着走能治腰颈椎痛?假的! 2019-04-19
  • 长效机制加速推进 楼市下半年或持续降温 2019-04-19
  • 树立文化自信 创新节庆模式 2019-04-19
  • 朝韩将军级会谈时隔11年后在板门店重启 2019-04-19
  • 经济日报多媒体数字报刊 2019-04-18
  • 搞好公有制就是好,故得出结论:计划经济好。 2019-04-18
  • 云南理发店老板涉嫌杀害女演员因办卡纠纷起杀心 2019-04-18
  • 南海网-海南新闻网-权威媒体 海南门户 2019-04-17
  • 海底捞回应侵犯音乐人林海著作权:已停止播放 2019-04-17
  • 自然型社会和规则性社会,是会随着科技的改变而发生改变的,当然只有规矩也就是制度才能规范人的行为,所以国家是不会灭亡的,但国家的形式是会发生改变的。 2019-04-17
  • 惊艳!上外学子英译60首热门中文歌  让世界倾听中国 2019-04-16
  • 西安,给盲人朋友留一条路吧…无障碍设施盲道-编辑整合 2019-04-16
  • 的确如此。报刊亭取消的确是短视行为。把报刊亭设计的现代化一些,与城市绿化衔接起来,相得益彰,成为文化一景多好。 2019-04-16
  • 让更多企业和劳动者尝到协商的“甜头” 2019-04-16
  • 2014金家岭财富论坛嘉宾云集(二) 2019-04-15
  • 181| 289| 305| 401| 396| 66| 30| 408| 58| 171|