You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "sheveg (via GitHub)" <gi...@apache.org> on 2024/03/14 10:28:30 UTC
[I] Metadata not found for table created by flink [iceberg]
sheveg opened a new issue, #9958:
URL: https://github.com/apache/iceberg/issues/9958
### Apache Iceberg version
1.4.3
### Query engine
Trino
### Please describe the bug 🐞
When using Flink to put data in iceberg using Nessie as catalog, I get an Metadata not found in metadata location for table error when trying to query the data generated by flink. When I query the data generated by trino itself using its CLI, then I do not have this problem. The iceberg connector configurations is:
```
iceberg: |-
connector.name=iceberg
iceberg.catalog.type=nessie
iceberg.nessie-catalog.uri=http://nessie.nessie.svc.cluster.local:19120/api/v1
iceberg.nessie-catalog.default-warehouse-dir=s3://steadops-playground-bucket
iceberg.nessie-catalog.ref=main
iceberg.register-table-procedure.enabled=true
iceberg.file-format=PARQUET
hive.s3.path-style-access=true
hive.s3.endpoint=http://minio.minio.svc.cluster.local:80
hive.s3.aws-access-key=${ENV:MINIO_ACCESS_KEY}
hive.s3.aws-secret-key=${ENV:MINIO_SECRET_KEY}
hive.s3.ssl.enabled=false
```
When I create the data using the trino CLI, the .parquet file is named in the schema: <date>_<time>_<uuid>.parquet whereas the data files generated by flink are named as 00000_0_<uuid>.parquet. The metadata files are named according to the same name schema as far as I see: <seq_number>_.metadata.json. The only difference I see, that the metadata directory for flink does not contain any *.stats files.
Cross reference to issue in trino repository: https://github.com/trinodb/trino/issues/21007
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "sheveg (via GitHub)" <gi...@apache.org>.
sheveg commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997460536
I adjusted now the setting using the configuration you linked:
```yaml
connector.name=iceberg
iceberg.catalog.type=nessie
iceberg.nessie-catalog.uri=http://nessie.nessie.svc.cluster.local:19120/api/v1
iceberg.nessie-catalog.default-warehouse-dir=s3://steadops-playground-bucket
iceberg.nessie-catalog.ref=main
iceberg.register-table-procedure.enabled=true
iceberg.file-format=PARQUET
fs.native-s3.enabled=true
s3.path-style-access=true
s3.endpoint=http://minio.minio.svc.cluster.local:80
s3.aws-access-key=${ENV:MINIO_ACCESS_KEY}
s3.aws-secret-key=${ENV:MINIO_SECRET_KEY}
s3.region=us-east-1
```
Now I get the error that that schema created by flink does not exist. However, I can create a new schema using the cli and query it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997149838
@sheveg a detailed stack trace or a log file would be helpful. Additionally, please state which Flink version you're using. The config you posted is the Trino connector config. How do you configure Flink?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997494457
In any case, this is an issue related to Trino, so it's probably best to move this issue to Trino as the Iceberg community doesn't maintain the Iceberg Connector for Trino
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra closed issue #9958: Metadata not found for table created by flink
URL: https://github.com/apache/iceberg/issues/9958
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997225394
> Which log entries do you mean exactly?
I mean the log/stack trace when `Metadata not found in metadata location for table ...` happens
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "sheveg (via GitHub)" <gi...@apache.org>.
sheveg commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997291522
I get this error: `Metadata not found in metadata location for table`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "sheveg (via GitHub)" <gi...@apache.org>.
sheveg commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997175547
We deploy the flink cluster via Kubernetes via the operator. This is the resource file for it:
```yaml
---
# Source: flink-demo/templates/flink-session-cluster.yaml
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: flink-session-cluster
spec:
image: ghcr.io/steadforce/applications/flink-steadforce:sha-796959d
flinkVersion: v1_17
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
serviceAccount: flink
podTemplate:
spec:
containers:
- name: flink-main-container
env:
- name: AWS_ACCESS_KEY_ID
valueFrom:
secretKeyRef:
key: accesskey
name: minio-cred-secret-mirror
- name: AWS_REGION
value: us-east-1
- name: AWS_SECRET_ACCESS_KEY
valueFrom:
secretKeyRef:
key: secretkey
name: minio-cred-secret-mirror
- name: KAFKA_BOOTSTRAP_SERVERS
value: kafka-kafka-bootstrap.kafka.svc.cluster.local:9092
- name: KAFKA_CONSUMER_GROUP
value: demo-consumer-group
- name: KAFKA_SCHEMA_REGISTRY_URL
value: http://kafka-schema-registry.kafka.svc.cluster.local:8081
- name: KAFKA_TOPIC
value: steadforce.tools
- name: MINIO_HOST
value: http://minio.svc.cluster.local:80
- name: NESSIE_HOST
value: http://nessie.nessie.svc.cluster.local:19120/api/v1
- name: WAREHOUSE
value: s3://steadops-playground-bucket/steadops-playground-bucket
jobManager:
resource:
cpu: 1
memory: 2048m
taskManager:
resource:
cpu: 1
memory: 2048m
```
The `flink-conf.yaml`:
```
blob.server.port: 6124
kubernetes.jobmanager.annotations: flinkdeployment.flink.apache.org/generation:2
taskmanager.memory.process.size: 2 gb
kubernetes.internal.jobmanager.entrypoint.class: org.apache.flink.kubernetes.entrypoint.KubernetesSessionClusterEntrypoint
kubernetes.jobmanager.replicas: 1
jobmanager.rpc.address: flink-session-cluster.flink-demo
kubernetes.pod-template-file.taskmanager: /tmp/flink_op_generated_podTemplate_679649916655451497.yaml
web.cancel.enable: false
execution.target: kubernetes-session
jobmanager.memory.process.size: 2 gb
kubernetes.service-account: flink
kubernetes.cluster-id: flink-session-cluster
taskmanager.rpc.port: 6122
kubernetes.jobmanager.cpu.amount: 1.0
kubernetes.taskmanager.cpu.amount: 1.0
internal.cluster.execution-mode: NORMAL
parallelism.default: 1
kubernetes.namespace: flink-demo
taskmanager.numberOfTaskSlots: 2
kubernetes.rest-service.exposed.type: ClusterIP
kubernetes.jobmanager.owner.reference: apiVersion:flink.apache.org/v1beta1,kind:FlinkDeployment,uid:9dc2c0e8-3de5-4627-bf6b-83f3f056f78b,name:flink-session-cluster,controller:false,blockOwnerDeletion:true
$internal.flink.version: v1_17
kubernetes.pod-template-file.jobmanager: /tmp/flink_op_generated_podTemplate_14004324014081248936.yaml
kubernetes.container.image.ref: ghcr.io/steadforce/applications/flink-steadforce:sha-796959d
```
We execute the job using the Java API:
```java
public static void main(String[] args) throws Exception {
ParameterTool parameter = ParameterTool.fromArgs(args);
String kafkaBootstrapServers = System.getenv("KAFKA_BOOTSTRAP_SERVERS");
String kafkaTopic = System.getenv("KAFKA_TOPIC");
String kafkaSchemaRegistryUrl = System.getenv("KAFKA_SCHEMA_REGISTRY_URL");
String kafkaConsumerGroup = System.getenv("KAFKA_CONSUMER_GROUP");
String nessieHost = System.getenv("NESSIE_HOST");
String warehouse = System.getenv("WAREHOUSE");
String minioHost = System.getenv("MINIO_HOST");
// set up the execution environment
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000);
// set up the table environment
final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(
env,
EnvironmentSettings.newInstance().inStreamingMode().build());
// Create KafkaSource builder
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
.setBootstrapServers(kafkaBootstrapServers)
.setGroupId(kafkaConsumerGroup)
.setTopics(kafkaTopic)
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
// create the Nessie catalog
tableEnv.executeSql(
String.format(
"CREATE CATALOG iceberg WITH ("
+ "'type'='iceberg',"
+ "'catalog-impl'='org.apache.iceberg.nessie.NessieCatalog',"
+ "'io-impl'='org.apache.iceberg.aws.s3.S3FileIO',"
+ "'uri'='%s',"
+ "'authentication.type'='none',"
+ "'ref'='main',"
+ "'client.assume-role.region'='us-east-1',"
+ "'warehouse'='%s',"
+ "'s3.endpoint'='%s'"
+ ")", nessieHost, warehouse, minioHost));
// List all catalogs
TableResult result = tableEnv.executeSql("SHOW CATALOGS");
// Print the result to standard out
result.print();
// Set the current catalog to the new catalog
tableEnv.useCatalog("iceberg");
// Create a database in the current catalog
tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS db");
// Add Kafka source as a data source
DataStream<String> kafkaStream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");
// // Filter manipulated and complete rows
Random random = new Random();
DataStream<String> manipulatedRowsStream = kafkaStream.filter(row -> isManipulatedRow(row, random));
DataStream<String> completeRowsStream = kafkaStream.filter(row -> !isManipulatedRow(row, random));
// apply a map transformation to convert the Tuple2 to an JobData object
DataStream<JobData> mappedManipulatedStream = manipulatedRowsStream.map(new MapFunction<String, JobData>() {
@Override
public JobData map(String value) throws Exception {
// perform your mapping logic here and return a JobData instance
// for example:
return new JobData(random.nextLong(), value);
}
});
DataStream<JobData> mappedCompleteStream = completeRowsStream.map(new MapFunction<String, JobData>() {
@Override
public JobData map(String value) throws Exception {
// perform your mapping logic here and return a JobData instance
// for example:
return new JobData(random.nextLong(), value);
}
});
// Convert the DataStream to a Table
Table manipulated_table = tableEnv.fromDataStream(mappedManipulatedStream, $("id"), $("data"));
Table complete_table = tableEnv.fromDataStream(mappedCompleteStream,$("id"), $("data"));
// Register the Table as a temporary view
tableEnv.createTemporaryView("my_complete_table", complete_table);
tableEnv.createTemporaryView("my_manipulated_table", manipulated_table);
// Create the tables
tableEnv.executeSql(
"CREATE TABLE IF NOT EXISTS db.complete_table ("
+ "id BIGINT COMMENT 'unique id',"
+ "data STRING"
+ ")");
tableEnv.executeSql(
"CREATE TABLE IF NOT EXISTS db.manipulated_table ("
+ "id BIGINT COMMENT 'unique id',"
+ "data STRING"
+ ")");
// Write Table to Iceberg
manipulated_table.executeInsert("db.manipulated_table");
complete_table.executeInsert("db.complete_table");
// Execute the job
env.execute("Flink Job");
```
Source: https://github.com/steadforce/flink-job/blob/3d40983fd6dc74d1004837b772c073d66f0bec4f/flink_job/src/main/java/com/steadforce/flink_job/App.java
We get the above mentioned error from the trino cli, when deploying a pod manually and connecting to our trino server. We use trino version `440`, flink version `1.17` and apache iceberg version `1.4.3` using nessie catalog with image `0.77.1`. Which log entries do you mean exactly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997386469
So I think the issue is how you point to MinIO in Trino vs in your Flink app. In Flink you use `s3.endpoint`, which is correct. But I don't think you're using the correct configuration in Trino to point to MinIO. Shouldn't you be using https://trino.io/docs/current/object-storage/file-system-s3.html instead of `hive.xyz`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Metadata not found for table created by flink [iceberg]
Posted by "sheveg (via GitHub)" <gi...@apache.org>.
sheveg commented on issue #9958:
URL: https://github.com/apache/iceberg/issues/9958#issuecomment-1997411834
But when I do create a table directly using the trino cli, I can also query this data. I get this error only for the data generated by flink.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org