import parquet.example.data.Group; //导入依赖的package包/类
public PFileWriter build(){
    try {
        this.parquetWriter = new ParquetWriter<Group>(
                file,
                gws,
                CompressionCodecName.SNAPPY,
                1024,
                1024,
                512,
                true,
                false,
                ParquetProperties.WriterVersion.PARQUET_1_0,
                conf);
    }catch (IOException ioe){
        LOG.error(ioe.toString());
    }
    return this;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
public static ParquetWriter<Group> initWriter(String fileName, Map<String, String> metas)
        throws IOException{


    GroupWriteSupport.setSchema(schema, conf);


    ParquetWriter<Group> writer = new ParquetWriter<Group>(
            initFile(fileName),
            new GroupWriteSupport(metas),
            CompressionCodecName.SNAPPY,
            1024,
            1024,
            512,
            true,
            false,
            ParquetProperties.WriterVersion.PARQUET_1_0,
            conf);

    return writer;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
/**
 * transform data in group into cells(List<cell> - > {@link org.apache.hadoop.hbase.client.Result}</>)
 * @param group
 * @return
 */
public static List<Cell> groupToCells(Group group){

    List<Cell> cells = new LinkedList<>();
    if(group != null){
        cells = new LinkedList<>();
        GroupType groupType = group.getType();
        List<Type> types = groupType.getFields();
        byte [] rowKey = group.getBinary(HConstants.ROW_KEY, 0).getBytes();

        long timestamp = group.getLong(HConstants.TIME_STAMP, 0);

        for(Type t : types){
            if(! t.getName().equals(HConstants.ROW_KEY) && ! t.getName().equals(HConstants.TIME_STAMP)){
                String name = t.getName();
                String [] names = name.split(":");
                if(names.length == 2) {
                    byte[] value = group.getBinary(name, 0).getBytes();
                    Cell cell = new KeyValue(rowKey, names[0].getBytes(), names[1].getBytes(), timestamp, value);
                    cells.add(cell);
                }
            }
        }
    }
    return cells;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
public static List<Group> readFile(File f, Filter filter) throws IOException {
    Configuration conf = new Configuration();
    GroupWriteSupport.setSchema(schema, conf);

    ParquetReader<Group> reader =
            ParquetReader.builder(new GroupReadSupport(), new Path(f.getAbsolutePath()))
                    .withConf(conf)
                    .withFilter(filter)
                    .build();

    Group current;
    List<Group> users = new ArrayList<Group>();

    current = reader.read();
    while (current != null) {
        users.add(current);
        current = reader.read();
    }

    return users;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
@Test
public void testUserDefinedByInstance() throws Exception {
    LongColumn name = longColumn("id");

    final HashSet<Long> h = new HashSet<Long>();
    h.add(20L);
    h.add(27L);
    h.add(28L);

    FilterPredicate pred = userDefined(name, new SetInFilter(h));

    List<Group> found = PhoneBookWriter.readFile(phonebookFile, FilterCompat.get(pred));

    assertFilter(found, new UserFilter() {

        public boolean keep(User u) {
            return u != null && h.contains(u.getId());
        }
    });
}
 

import parquet.example.data.Group; //导入依赖的package包/类
/**
 * transform mutation to group
 * assume that sfg is not null
 * @param gf
 * @return
 */

public Group asGroup(GroupFactory gf){

    Group group = gf.newGroup().append(ROW_KEY, Bytes.toString(row));
    CellScanner cellScanner = cellScanner();
    try {
        while (cellScanner.advance()){
            Cell cell = cellScanner.current();
            group.append(
                    Bytes.toString(cell.getFamily()) +":"+ Bytes.toString(cell.getQualifier()),
                    Bytes.toString(cell.getValue())
            );
        }
        group.append("timestamp", System.currentTimeMillis());
    }catch (IOException io){
        LOG.error(io);
    }
    return group;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
public String toString(String indent) {
  StringBuilder result = new StringBuilder();
  int i = 0;
  for (Type field : this.schema.getFields()) {
    String name = field.getName();
    List<Object> values = this.data[i];
    for (Object value : values) {
      result.append(indent).append(name);
      if (value == null) {
        result.append(": NULL\n");
      } else if (value instanceof Group) {
        result.append("\n").append(((ParquetGroup) value).toString(indent + "  "));
      } else {
        result.append(": ").append(value.toString()).append("\n");
      }
    }
    i++;
  }
  return result.toString();
}
 

import parquet.example.data.Group; //导入依赖的package包/类
/**
 * Build a {@link ParquetWriter<Group>} for given file path with a block size.
 * @param blockSize
 * @param stagingFile
 * @return
 * @throws IOException
 */
public ParquetWriter<Group> getWriter(int blockSize, Path stagingFile)
    throws IOException {
  State state = this.destination.getProperties();
  int pageSize = state.getPropAsInt(getProperty(WRITER_PARQUET_PAGE_SIZE), DEFAULT_PAGE_SIZE);
  int dictPageSize = state.getPropAsInt(getProperty(WRITER_PARQUET_DICTIONARY_PAGE_SIZE), DEFAULT_BLOCK_SIZE);
  boolean enableDictionary =
      state.getPropAsBoolean(getProperty(WRITER_PARQUET_DICTIONARY), DEFAULT_IS_DICTIONARY_ENABLED);
  boolean validate = state.getPropAsBoolean(getProperty(WRITER_PARQUET_VALIDATE), DEFAULT_IS_VALIDATING_ENABLED);
  String rootURI = state.getProp(WRITER_FILE_SYSTEM_URI, LOCAL_FS_URI);
  Path absoluteStagingFile = new Path(rootURI, stagingFile);
  CompressionCodecName codec = getCodecFromConfig();
  GroupWriteSupport support = new GroupWriteSupport();
  Configuration conf = new Configuration();
  GroupWriteSupport.setSchema(this.schema, conf);
  ParquetProperties.WriterVersion writerVersion = getWriterVersion();
  return new ParquetWriter<>(absoluteStagingFile, support, codec, blockSize, pageSize, dictPageSize, enableDictionary,
      validate, writerVersion, conf);
}
 

import parquet.example.data.Group; //导入依赖的package包/类
private List<Group> readParquetFiles(File outputFile)
    throws IOException {
  ParquetReader<Group> reader = null;
  List<Group> records = new ArrayList<>();
  try {
    reader = new ParquetReader<>(new Path(outputFile.toString()), new SimpleReadSupport());
    for (Group value = reader.read(); value != null; value = reader.read()) {
      records.add(value);
    }
  } finally {
    if (reader != null) {
      try {
        reader.close();
      } catch (Exception ex) {
        System.out.println(ex.getMessage());
      }
    }
  }
  return records;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
@Override
protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
  Mean mean = new Mean();
  for (DoubleWritable val : values) {
    mean.increment(val.get());
  }
  Group group = factory.newGroup()
      .append("symbol", key.toString())
      .append("avg", mean.getResult());
  context.write(null, group);
}
 

import parquet.example.data.Group; //导入依赖的package包/类
@Override
public void map(Void key,
                Group value,
                Context context) throws IOException, InterruptedException {
  context.write(new Text(value.getString("symbol", 0)),
      new DoubleWritable(Double.valueOf(value.getValueToString(2, 0))));
}
 

import parquet.example.data.Group; //导入依赖的package包/类
public void append(Group group){
    try {
        parquetWriter.write(group);
    }catch (IOException ioe){
        LOG.error(ioe.toString());
    }
}
 

import parquet.example.data.Group; //导入依赖的package包/类
/**
 * read a row
 */
@Override
public Group readGroup() {
    Group group = null;
    try {
        group = reader.read();
    }catch (IOException ioe){
     LOG.error(ioe);
    }
    return group;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
/**
 * read value from parquet as cell
 *
 * @return
 */
@Override
public List<Cell> readCells() {
    List<Cell> cells = new LinkedList<>();
    Group group = readGroup();
    if(group == null)
        return null;
    else {
        List<ColumnDescriptor> columns = schema.getColumns();
        for (ColumnDescriptor column : columns){

        }
        return cells;
    }
}
 

import parquet.example.data.Group; //导入依赖的package包/类
/**
 * @return the next {@link Group}
 */
public Group nextRow() {
    Group rs = curr;
    curr = next;
    next = reader.readGroup();
    return rs;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
/**
 * return the record
 */
@Override
public List<Cell> next() {
    List<Cell> record = new LinkedList<>();
    Group group = nextRow();
    if(group != null){
        record.addAll(groupToCells(group));
    }
    return record;
}
 

import parquet.example.data.Group; //导入依赖的package包/类
public static void main(String []args) throws IOException{


        int fileNum = 10;   //num of files constructed
        int fileRecordNum = 50; //record num of each file
        int rowKey = 0;
        for(int i = 0; i < fileNum; ++ i ) {

            Map<String, String> metas = new HashMap<>();
            metas.put(HConstants.START_KEY, genRowKey("%10d", rowKey + 1));
            metas.put(HConstants.END_KEY, genRowKey("%10d", rowKey + fileRecordNum));

            ParquetWriter<Group> writer = initWriter("pfile/scanner_test_file" + i, metas);

            for (int j = 0;  j < fileRecordNum; ++j) {
                rowKey ++;
                Group group = sfg.newGroup().append("rowkey", genRowKey("%10d", rowKey))
                        .append("cf:name", "wangxiaoyi" + rowKey)
                        .append("cf:age", String.format("%10d", rowKey))
                        .append("cf:job", "student")
                        .append("timestamp", System.currentTimeMillis());
                writer.write(group);
            }

            writer.close();
        }
    }
 

import parquet.example.data.Group; //导入依赖的package包/类
public static void main(String[] args) throws IOException {


        Path root = new Path("hdfs://10.214.208.11:9000/parquet/");//文件夹路径

        Configuration configuration = new Configuration();

        MessageType schema = MessageTypeParser.parseMessageType( //parquet文件模式

                " message people { " +

                        "required binary rowkey;" +
                        "required binary cf:name;" +
                        "required binary cf:age;" +
                        "required int64 timestamp;"+
                 " }");

        GroupWriteSupport.setSchema(schema, configuration);

        SimpleGroupFactory sfg = new SimpleGroupFactory(schema);
        Path file = new Path(root, "people002.parquet");

        Map<String, String> meta = new HashMap<String, String>();
        meta.put("startkey", "1");
        meta.put("endkey", "2");


        ParquetWriter<Group> writer = new ParquetWriter<Group>(
                file,
                new GroupWriteSupport(meta),
                CompressionCodecName.UNCOMPRESSED,
                1024,
                1024,
                512,
                true,
                false,
                ParquetProperties.WriterVersion.PARQUET_1_0,
                configuration);

        Group group = sfg.newGroup().append("rowkey", "1")
                      .append("cf:name", "wangxiaoyi")
                      .append("cf:age", "24")
                      .append("timestamp", System.currentTimeMillis());


        for (int i = 0; i < 10000; ++i) {
            writer.write(
                    sfg.newGroup()
                            .append("name", "wangxiaoyi" + i)
                            .append("age", i));
        }
        writer.close();


    }
 

import parquet.example.data.Group; //导入依赖的package包/类
@Test
public void testWriteReadStatisticsAllNulls() throws Exception {

    File testFile = new File("target/test/TestParquetFileWriter/testParquetFile").getAbsoluteFile();
    testFile.delete();

    writeSchema = "message example {\n" +
            "required binary content;\n" +
            "}";

    Path path = new Path(testFile.toURI());

    MessageType schema = MessageTypeParser.parseMessageType(writeSchema);
    Configuration configuration = new Configuration();
    GroupWriteSupport.setSchema(schema, configuration);

    ParquetWriter<Group> writer = new ParquetWriter<Group>(path, configuration, new GroupWriteSupport(null));

    Group r1 = new SimpleGroup(schema);
    writer.write(r1);
    writer.close();

    ParquetMetadata readFooter = ParquetFileReader.readFooter(configuration, path);

    // assert the statistics object is not empty
    assertTrue((readFooter.getBlocks().get(0).getColumns().get(0).getStatistics().isEmpty()) == false);
    // assert the number of nulls are correct for the first block
    assertEquals(1, (readFooter.getBlocks().get(0).getColumns().get(0).getStatistics().getNumNulls()));
}
 

import parquet.example.data.Group; //导入依赖的package包/类
public static SimpleGroup groupFromUser(User user) {
    SimpleGroup root = new SimpleGroup(schema);
    root.append("id", user.getId());

    if (user.getName() != null) {
        root.append("name", user.getName());
    }

    if (user.getPhoneNumbers() != null) {
        Group phoneNumbers = root.addGroup("phoneNumbers");
        for (PhoneNumber number : user.getPhoneNumbers()) {
            Group phone = phoneNumbers.addGroup("phone");
            phone.append("number", number.getNumber());
            if (number.getKind() != null) {
                phone.append("kind", number.getKind());
            }
        }
    }

    if (user.getLocation() != null) {
        Group location = root.addGroup("location");
        if (user.getLocation().getLon() != null) {
            location.append("lon", user.getLocation().getLon());
        }
        if (user.getLocation().getLat() != null) {
            location.append("lat", user.getLocation().getLat());
        }
    }
    return root;
}