Clen Moras
07/28/2023, 5:38 AM... 42 more
Caused by: java.nio.file.AccessDeniedException: <s3a://clen/results-store/.hoodie>: getFileStatus on <s3a://clen/results-store/.hoodie>: com.amazonaws.services.s3.model.AmazonS3Exception: Forbidden (Service: Amazon S3; Status Code: 403; Error Code: 403 Forbidden; Request ID: QKBF6K7H0D1; S3 Extended Request ID: +pWzTD5f6hHSWmN39T9izjZ4888VsMQQSoMwm9B/fjBicsfKc=; Proxy: null), S3 Extended Request ID: +pWzTD5f6hHSWmN39T9izjZ4888VsMQQKc=:403 Forbidden
at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:246)
at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:171)
at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:3645)
at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:3551)
at org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$exists$33(S3AFileSystem.java:4530)
at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.invokeTrackingDuration(IOStatisticsBinding.java:547)
at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.lambda$trackDurationOfOperation$5(IOStatisticsBinding.java:528)
at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDuration(IOStatisticsBinding.java:449)
at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2401)
at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2420)
at org.apache.hadoop.fs.s3a.S3AFileSystem.exists(S3AFileSystem.java:4528)
at org.apache.hudi.util.StreamerUtil.getTableConfig(StreamerUtil.java:295)
Based on documentation https://hudi.apache.org/docs/s3_hoodie/, only explicitly defining s3 creds in core-site.xml is correct way.
I would like to avoid using creds if possible, is there an option to handle it this way?
Has anyone used org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider or org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider
From the looks of it , it only only ssems to be happily working org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider
org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider: this one i m not clear about
Any help would be greatly appreciated.Martijn Visser
07/28/2023, 6:22 AMNeha
07/28/2023, 6:24 AMClen Moras
07/28/2023, 7:48 AMAre you making sure that you use S3 Presto for checkpointing, and S3 Hadoop for reading/writing to filesystem?How do i confirm this martin? i m currently s3://. for checkpointing/savepointing and s3a://. for any activity which involves parquet files ->s3
I have used org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProviderNeha, if you dont mind, could you tell me what fields have you added in your core-site.xml
Clen Moras
07/28/2023, 7:57 AMflinkConfiguration:
classloader.parent-first-patterns.additional: org.rocksdb.
classloader.resolve-order: parent-first
cluster.evenly-spread-out-slots: "true"
containerized.master.env.SINK_FILE_PATH: <s3a://flink-clen-datasets/datasets/results1>
containerized.master.env.SINK_FILE_PATH_RCA: <s3a://flink-clen-datasets/datasets/results2>
containerized.master.env.SINK_PARALLELISM: "4"
execution.attached: "true"
execution.checkpointing.interval: 5m
execution.checkpointing.max-concurrent-checkpoints: "1"
execution.checkpointing.min-pause: 5m
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.snapshot-compression: "true"
execution.checkpointing.timeout: 5m
execution.checkpointing.unaligned: "true"
execution.savepoint-restore-mode: CLAIM
execution.savepoint.ignore-unclaimed-state: "true"
fs.s3a.aws.credentials.provider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
high-availability.storageDir: <s3://flink-clen/deployments/test-piepline1/ha>
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: "9250"
metrics.reporters: prom
rest.flamegraph.enabled: "true"
state.backend: rocksdb
state.backend.incremental: "true"
state.backend.local-recovery: "true"
state.backend.rocksdb.checkpoint.transfer.thread.num: "4"
state.backend.rocksdb.localdir: /opt/flink/rocksdb
state.backend.rocksdb.log.dir: /opt/flink/log/
state.backend.rocksdb.log.level: DEBUG_LEVEL
state.backend.rocksdb.metrics.background-errors: "true"
state.backend.rocksdb.metrics.bytes-read: "true"
state.backend.rocksdb.metrics.bytes-written: "true"
state.backend.rocksdb.metrics.estimate-num-keys: "true"
state.backend.rocksdb.metrics.is-write-stopped: "true"
state.checkpoints.dir: <s3://flink-clen/deployments/test-piepline1/checkpoints>
state.checkpoints.num-retained: "2"
state.savepoints.dir: <s3://flink-clen/deployments/test-piepline1/savepoints>
taskmanager.memory.managed.fraction: "0.2"
taskmanager.numberOfTaskSlots: "2"
web.cancel.enable: "false"
this is my flink config.
flinkVersion: v1_17
serviceAccount: flink
apiVersion: v1
kind: ServiceAccount
metadata:
annotations:
<http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam:::role/role
labels:
<http://app.kubernetes.io/name|app.kubernetes.io/name>: flink-kubernetes-operator
<http://app.kubernetes.io/version|app.kubernetes.io/version>: 1.5.0
<http://helm.sh/chart|helm.sh/chart>: flink-kubernetes-operator-1.5.0
name: flink
trust relation
"Action": "sts:AssumeRoleWithWebIdentity",
"Condition": {
"StringEquals": {
"<http://oidc.eks.us-east-1.amazonaws.com/id/:sub|oidc.eks.us-east-1.amazonaws.com/id/:sub>": "system:serviceaccount:{ns}:flink"
}
i have covered this, this seem to be good.Clen Moras
07/28/2023, 8:01 AM<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!--
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
<http://www.apache.org/licenses/LICENSE-2.0>
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. See accompanying LICENSE file.
-->
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
<name>fs.s3a.assumed.role.arn</name>
<value>arn:aws:iam:::role/role</value>
<description>
AWS ARN for the role to be assumed.
Required if the fs.s3a.aws.credentials.provider contains
org.apache.hadoop.fs.s3a.AssumedRoleCredentialProvider
</description>
</property>
<property>
<name>fs.s3a.assumed.role.session.name</name>
<value>s3-connect</value>
<description>
Session name for the assumed role, must be valid characters according to
the AWS APIs.
Only used if AssumedRoleCredentialProvider is the AWS credential provider.
If not set, one is generated from the current Hadoop/Kerberos username.
</description>
</property>
<property>
<name>fs.s3a.assumed.role.session.duration</name>
<value>1h</value>
<description>
Duration of assumed roles before a refresh is attempted.
Only used if AssumedRoleCredentialProvider is the AWS credential provider.
Range: 15m to 1h
</description>
</property>
<property>
<name>fs.s3a.assumed.role.credentials.provider</name>
<value>org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider</value>
<description>
List of credential providers to authenticate with the STS endpoint and
retrieve short-lived role credentials.
Used by AssumedRoleCredentialProvider and the S3A Session Delegation Token
and S3A Role Delegation Token bindings.
</description>
</property>
</configuration>
jars that i have added
flink-hadoop-compatibility_2.12
hadoop-common
hadoop-aws
parquet-hadoop
aws-java-sdk-s3
hudi-flink1.16-bundle
created plugins for
plugins/s3-fs-hadoop
plugins/azure-fs-hadoopClen Moras
08/03/2023, 2:21 PM<configuration>
<property>
<name>fs.s3a.impl</name>
<value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
</property>
<property>
<name>fs.s3a.access.key</name>
<value>YOUR_AWS_ACCESS_KEY</value>
</property>
<property>
<name>fs.s3a.secret.key</name>
<value>YOUR_AWS_SECRET_KEY</value>
</property>
<!-- Optionally, set other S3 configurations, e.g., region, endpoint, etc. -->
<property>
<name>fs.s3a.region</name>
<value>us-east-1</value>
</property>
<!-- Set the buffer directory for S3A -->
<property>
<name>fs.s3a.buffer.dir</name>
<value>/tmp</value>
</property>
</configuration>