如何使用 AvroParquetWriter 并通过 AmazonS3 api 写入 S3？

Question

我目前正在使用下面的代码通过 Avro 编写镶木地板。此代码将其写入文件系统，但我想写入 S3。

try {
    StopWatch sw = StopWatch.createStarted();
    Schema avroSchema = AvroSchemaBuilder.build("pojo", message.getTransformedMessage().get(0));
    final String parquetFile = "parquet/data.parquet";
    final Path path = new Path(parquetFile);

    ParquetWriter writer = AvroParquetWriter.<GenericData.Record>builder(path)
        .withSchema(avroSchema)
        .withConf(new org.apache.hadoop.conf.Configuration())
        .withCompressionCodec(CompressionCodecName.SNAPPY)
        .withWriteMode(Mode.OVERWRITE)//probably not good for prod. (overwrites files).
        .build();

    for (Map<String, Object> row : message.getTransformedMessage()) {
      StopWatch stopWatch = StopWatch.createStarted();
      final GenericRecord record = new GenericData.Record(avroSchema);
      row.forEach((k, v) -> {
        record.put(k, v);
      });
      writer.write(record);
    }
    //todo:  Write to S3.  We should probably write via the AWS objects.  This does not show that.
    //https://stackoverflow.com/questions/47355038/how-to-generate-parquet-file-using-pure-java-including-date-decimal-types-an
    writer.close();
    System.out.println("Total Time: " + sw);

  } catch (Exception e) {
    //do somethign here.  retryable?  non-retryable?  Wrap this excetion in one of these?
    transformedParquetMessage.getOriginalMessage().getMetaData().addException(e);
  }

这可以很好地写入文件，但是如何让它流式传输到 AmazonS3 api 中？我在网上找到了一些使用 Hadoop-aws jar 的代码，但这需要一些 Windows exe 文件才能工作，当然，我们希望避免这种情况。目前我只使用：

 <dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro</artifactId>
  <version>1.9.2</version>
</dependency>
<dependency>
  <groupId>org.apache.parquet</groupId>
  <artifactId>parquet-avro</artifactId>
  <version>1.8.1</version>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-core</artifactId>
  <version>1.2.1</version>
</dependency>

所以问题是，有没有办法拦截 AvroParquetWriter 上的输出流，以便我可以将其流式传输到 S3？ 我想要这样做的主要原因是重试。 S3 最多自动重试 3 次。这对我们有很大帮助。

Answer 1

这确实取决于 hadoop-aws jar，所以如果您不愿意使用它，我不确定是否可以帮助您。但是，我在 mac 上运行并且没有任何 windows exe 文件，所以我不确定你说这些来自哪里。 AvroParquetWriter 已经依赖于 Hadoop，因此即使您无法接受这种额外的依赖，对其他人来说也可能没什么大不了的：

您可以使用 AvroParquetWriter 通过向其传递使用 URI 参数创建的 Hadoop 路径并设置正确的配置来直接流式传输到 S3。

val uri = new URI("s3a://<bucket>/<key>")
val path = new Path(uri)

val config = new Configuration()
config.set("fs.s3a.access.key", key)
config.set("fs.s3a.secret.key", secret)
config.set("fs.s3a.session.token", sessionToken)
config.set("fs.s3a.aws.credentials.provider", credentialsProvider)

val writer = AvroParquetWriter.builder[GenericRecord](path).withConf(config).withSchema(schema).build()

我使用了以下依赖项（sbt 格式）：

"org.apache.avro" % "avro" % "1.8.1"
"org.apache.hadoop" % "hadoop-common" % "2.9.0"
"org.apache.hadoop" % "hadoop-aws" % "2.9.0"
"org.apache.parquet" % "parquet-avro" % "1.8.1"

Answer 2

希望我没有误解这个问题，但看来您正在做的是将 avro 转换为 parquet 并且您想将 parquet 上传到 s3

关闭 ParquetWriter 后，您应该调用如下所示的方法（假定这不会拦截从 avro 写入 parquet 的流，它只会流式传输不再写入的 parquet 文件）：

        AmazonS3 s3Client = AmazonS3ClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("ACCESS_KEY", "SECRET_KEY"))).build();
        S3Path outputPath = new S3Path();
        outputPath.setBucket("YOUR_BUCKET");
        outputPath.setKey("YOUR_FOLDER_PATH");
        try {
            InputStream parquetStream = new FileInputStream(new File(parquetFile));
            s3Client.putObject(outputPath.getBucket(), outputPath.getKey(), parquetStream, null);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }

使用 AWS 开发工具包

<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-java-sdk</artifactId>
    <version>1.11.749</version>
</dependency>

当然，该方法将驻留在不同的 utils 类中，并且该方法的构造函数应使用凭证初始化 AmazonS3 s3Client，因此您需要做的就是调用并访问它的 s3Client 成员以放置对象

希望对你有帮助