本文介绍了我正在尝试通过FileIO与ParquetIO一起使用假定角色来写入S3的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
第1步:AssumeRole
Step1 : AssumeRole
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(); }第2步:将凭据设置为管道
Step 2 : Set Credentials to pipeline
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步:运行管道以写入s3
Step 3 : Run pipeline to write to 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'
问题:当前假设角色似乎无法正常工作.
Issue : Currently assumeRole seems to be not working.
错误:
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' 推荐答案最近释放梁(2.24.0)具有承担作用的功能.
Recently release of beam (2.24.0) has the feature to assume role.
更多推荐
我正在尝试通过FileIO与ParquetIO一起使用假定角色来写入S3
发布评论