S3Main.java

package org.apache.beam.sdk.io.aws;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.aws.options.S3Options;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;

/** S3Main. */
public class S3Main {
  public static void main(String[] args) throws InterruptedException {
    S3Options options = PipelineOptionsFactory.fromArgs(args).create().as(S3Options.class);
    options.setAwsRegion("eu-west-1");

    // SSE-C
    //    com.amazonaws.services.s3.model.SSECustomerKey sseCustomerKey =
    //        new com.amazonaws.services.s3.model.SSECustomerKey(
    //            "86glyTlCNZgccSxW8JxMa6ZdjdK3N141glAysPUZ3AA=");
    //    options.setSSECustomerKey(sseCustomerKey);

    // SSE-S3
    //    options.setSSEAlgorithm("AES256");

    // SSE-KMS
    String awsKmsKeyId =
        "arn:aws:kms:eu-west-1:...";
    com.amazonaws.services.s3.model.SSEAwsKeyManagementParams sseAwsKeyManagementParams =
        new com.amazonaws.services.s3.model.SSEAwsKeyManagementParams(awsKmsKeyId);
    options.setSSEAwsKeyManagementParams(sseAwsKeyManagementParams);

    String output = "s3://bucket/output";
    String input = output + "*";

    Pipeline writePipeline = Pipeline.create(options);
    writePipeline
        .apply("CreateSomeData", Create.of("1", "2", "3"))
        .apply("WriteCounts", TextIO.write().to(output));
    writePipeline.run().waitUntilFinish();

    Pipeline readPipeline = Pipeline.create(options);
    readPipeline
        .apply("ReadLines", TextIO.read().from(input))
        .apply(
            "PrintLines",
            MapElements.via(
                new SimpleFunction(
                    x -> {
                      System.out.println(x);
                      return x;
                    }) {}));
    readPipeline.run().waitUntilFinish();
  }
}