Blake Smith

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

final Path hadoopPath = new Path("file:///tmp/file.parquet");
final Configuration hadoopConfiguration = new Configuration();
final ParquetWriter<T> writer = AvroParquetWriter.<T>builder(hadoopPath)
    .withConf(hadoopConfiguration)
    .withSchema(schema)
    .build();

import org.apache.parquet.conf.ParquetConfiguration;
import org.apache.parquet.conf.PlainParquetConfiguration;
import org.apache.parquet.io.LocalOutputFile;
import java.nio.file.FileSystem;
import java.nio.file.FileSystems;
import java.nio.file.Path;

final FileSystem fs = FileSystems.getFileSystem(new URI("file", null, "/", null, null));
final ParquetConfiguration parquetConf = new PlainParquetConfiguration();
final Path nioOutputPath = fs.getPath("file:///tmp/file.parquet");
final LocalOutputFile outputFile = new LocalOutputFile(nioOutputPath);
final ParquetWriter<T> writer = AvroParquetWriter.<T>builder(outputFile)
    .withConf(parquetConf)
    .withSchema(schema)
    .build();

  <!-- BEGIN LINGERING HADOOP TRANSITIVE DEPENDENCIES
      There are still some lingering classes in parquet-mr that are coupled to Hadoop via their public interfaces.
      Remove this entirely once Hadoop is correctly made optional in parquet-mr for another
      bunch of jar size savings.
      NOTE: None of *our* code explicitly calls into any Hadoop code. However, some of the Parquet classes
      we depend on will import Hadoop dependencies (but not use them in our core code paths). Once parquet-mr
      makes Hadoop optional (probably sometime in 2.0.0, since these would break certain public interfaces),
      we can remove everything between these Hadoop dependencies.
      On wildcard exclusions: This is the bare-minimum jars that must be brought into the classpath to satisfy import dependencies.
      (The import graph from the classes we *do* use, such as ParquetWriter, etc.). We explicitly avoid bringing in any transitive dependencies, and just bring
      in the exact jars we need.
      -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>${hadoop.version}</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.woodstox</groupId>
      <artifactId>woodstox-core</artifactId>
      <version>5.4.0</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.codehaus.woodstox</groupId>
      <artifactId>stax2-api</artifactId>
      <version>4.2.1</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>commons-collections</groupId>
      <artifactId>commons-collections</artifactId>
      <version>3.2.2</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-core</artifactId>
      <version>${hadoop.version}</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop.thirdparty</groupId>
      <artifactId>hadoop-shaded-guava</artifactId>
      <version>1.2.0</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <!-- END LINGERING HADOOP TRANSITIVE DEPENDENCIES
         delete everything between these two blocks once parquet-mr makes Hadoop entirely optional.
    -->

package me.blakesmith.parquet;

import org.apache.parquet.io.InputFile;
import org.apache.parquet.io.SeekableInputStream;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;

/**
 * An {@link org.apache.parquet.io.InputFile} implementation that only uses
 * java.nio.* interfaces for I/O operations. The LocalInputFile implementation in
 * upstream parquet-mr currently falls back to the old-school java file I/O APIs
 * (via Path#toFile) which won't work with nio remote FileSystems such as an S3
 * FileSystem implementation.
 */
public class NioInputFile implements InputFile {
  private final Path path;
  private long length = -1;

  public NioInputFile(Path file) {
    path = file;
  }

  @Override
  public long getLength() throws IOException {
    if (length == -1) {
      length = Files.size(path);
    }
    return length;
  }

  @Override
  public SeekableInputStream newStream() throws IOException {

    return new SeekableInputStream() {

      private final SeekableByteChannel byteChannel = Files.newByteChannel(path);
      private final ByteBuffer singleByteBuffer = ByteBuffer.allocate(1);

      @Override
      public int read() throws IOException {
        // There has to be a better way to do this?
        singleByteBuffer.clear();
        final int numRead = read(singleByteBuffer);
        if (numRead >= 0) {
          int value = (int)singleByteBuffer.get(0) & 0xFF;
          return value;
        } else {
          return -1;
        }
      }

      @Override
      public long getPos() throws IOException {
        return byteChannel.position();
      }

      @Override
      public void seek(long newPos) throws IOException {
        byteChannel.position(newPos);
      }

      @Override
      public void readFully(byte[] bytes) throws IOException {
        readFully(bytes, 0, bytes.length);
      }

      @Override
      public void readFully(byte[] bytes, int start, int len) throws IOException {
        final ByteBuffer buf = ByteBuffer.wrap(bytes);
        buf.position(start);
        readFully(buf);
      }

      @Override
      public int read(ByteBuffer buf) throws IOException {
        return byteChannel.read(buf);
      }

      @Override
      public void readFully(ByteBuffer buf) throws IOException {
        int numRead = 0;
        while (numRead < buf.limit()) {
          final int code = read(buf);
          if (code == -1) {
            return;
          } else {
            numRead += code;
          }
        }
      }

      @Override
      public void close() throws IOException {
        byteChannel.close();
      }
    };
  }
}