本文介紹了我正在嘗試通過帶有ParquetIO的FileIO使用假設(shè)性角色向Amazon S3寫信的處理方法，對(duì)大家解決問題具有一定的參考價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)吧！
問題描述
第1步：Assum角色
public static AWSCredentialsProvider getCredentials() {
if (roleARN.length() > 0) {
STSAssumeRoleSessionCredentialsProvider credentialsProvider = new STSAssumeRoleSessionCredentialsProvider
.Builder(roleARN, Constants.SESSION_NAME)
.withStsClient(AWSSecurityTokenServiceClientBuilder.defaultClient())
.build();
return credentialsProvider;
}
return new ProfileCredentialsProvider();
}
第二步：將憑據(jù)設(shè)置為管道
credentials = getCredentials();
pipeline.getOptions().as(AwsOptions.class).setAwsRegion(Regions.US_WEST_2.getName());
pipeline.getOptions().as(AwsOptions.class).setAwsCredentialsProvider(new AWSStaticCredentialsProvider(new BasicAWSCredentials(credentials.getCredentials().getAWSAccessKeyId(), credentials.getCredentials().getAWSAccessKeyId())));
第3步：運(yùn)行管道以寫入S3
PCollection<GenericRecord> parquetRecord = formattedEvent
.apply("ParquetRecord", ParDo.of(new ParquetWriter()))
.setCoder(AvroCoder.of(getOutput_schema()));
parquetRecord.apply(FileIO.<GenericRecord, GenericRecord>writeDynamic()
.by(elm -> elm)
.via(ParquetIO.sink(getOutput_schema()))
.to(outputPath).withNumShards(1)
.withNaming(type -> FileNaming.getNaming("part", ".snappy.parquet", "" + DateTime.now().getMillisOfSecond()))
.withDestinationCoder(AvroCoder.of(getOutput_schema())));
我正在使用'org.apache.beam:beam-sdks-java-io-parquet:jar:2.22.0'和
'org.apache.beam:beam-sdks-java-io-amazon-web-services:jar:2.22.0'
問題：當(dāng)前假定角色似乎不起作用。
錯(cuò)誤：
org.apache.beam.sdk.util.UserCodeException: java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.io.IOException: com.amazonaws.services.s3.model.AmazonS3Exception: The AWS Access Key Id you provided does not exist in our records.
或
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Unexpected IOException (of type java.io.IOException): Failed to serialize and deserialize property 'awsCredentialsProvider' with value 'com.amazonaws.auth.InstanceProfileCredentialsProvider@71262020'
推薦答案
最近發(fā)布的BEAM(2.24.0)具有承擔(dān)角色的功能。
這篇關(guān)于我正在嘗試通過帶有ParquetIO的FileIO使用假設(shè)性角色向Amazon S3寫信的文章就介紹到這了，希望我們推薦的答案對(duì)大家有所幫助，
