[Question] Is there a way to have a per metalake configuration for the iceberg-rest catalog
- Dominant language
- Java
- Stars
- 3.2k
- Forks
- 935
- Avg merge
- 1d 15h
- Merged PRs (30d)
- 315
Description
Say I have two metalakes, metalake_0 and metalake_1
Say I have two buckets to which I want data written for iceberg: s3://metalake_0 and s3://metalake_1
How do I get spark, the iceberg-rest-catalog, and gravitino to all play nicely together in this way?
I'm trying to simulate all this in minio based off the gravitino-playground
okay here's how I create "metalakes"
```scala
case class Metalake(
name: String
)
def createMetalake(client: Client[IO], metalakeName: String): IO[Unit] = for {
_ <- IO.println("Creating metalake")
_ <- client
.expect[Unit](
Request[IO](
Method.POST,
uri"http://gravitino:8090/api/metalakes"
).withEntity(Metalake(metalakeName).asJson)
)
.handleErrorWith {
case e: UnexpectedStatus if e.status == Status.Conflict =>
IO.println("Metalake already exists. Continuing")
case e => IO.raiseError(e)
}
} yield ()
```
and catalogs...
```scala
case class Catalog(
name: String,
`type`: String,
provider: String,
properties: Map[String, String]
)
def createCatalog(client: Client[IO], metalakeName: String): IO[Unit] = {
val catalogProperties = Map(
"uri" -> "http://gravitino:9002/iceberg",
"catalog-backend" -> "rest",
"warehouse" -> f"s3://$metalakeName",
"io-impl" -> "org.apache.iceberg.aws.s3.S3FileIO",
"s3-access-key-id" -> "minioadmin",
"s3-secret-access-key" -> "minioadmin",
"s3-endpoint" -> "http://minio:9000",
"s3-region" -> "us-east-1"
)
val requestPayload = Catalog(
name = "mylake",
`type` = "relational",
provider = "lakehouse-iceberg",
properties = catalogProperties
)
val acceptHeader = Accept(MediaType.unsafeParse("application/vnd.gravitino.v1+json"))
val request = Request[IO](
Method.POST,
uri"http://gravitino:8090".withPath(path"api/metalakes" / metalakeName / "catalogs")
).withEntity(requestPayload.asJson).putHeaders(acceptHeader)
for {
_ <- IO.println("Creating catalog")
_ <- client
.expect[String](request)
.flatMap { successBody =>
IO.println(s"Successfully created catalog. Response: $successBody")
}
.handleErrorWith {
case e: UnexpectedStatus if e.status == Status.Conflict =>
IO.println("Catalog already exists")
case e => IO.println(e).flatMap(_ => IO.raiseError(e))
}
} yield ()
}
```
and here's how I make the "schema"
```scala
case class Schema(
name: String,
properties: Map[String, String]
)
def createSchema(client: Client[IO], metalakeName: String): IO[Unit] = {
val schemaName = "db"
val requestPayload = Schema(
name = schemaName,
properties = Map(
"warehouse" -> f"s3://$metalakeName",
"location" -> f"s3://$metalakeName/$schemaName"
)
)
val acceptHeader = Accept(MediaType.unsafeParse("application/vnd.gravitino.v1+json"))
val request = Request[IO](
Method.POST,
uri"http://gravitino:8090".withPath(
path"api/metalakes" / metalakeName / "catalogs" / "mylake" / "schemas")
).withEntity(requestPayload.asJson).putHeaders(acceptHeader)
for {
_ <- IO.println(s"Creating schema '$schemaName'...")
_ <- client
.expect[String](request)
.flatMap { successBody =>
IO.println(s"Successfully created schema. Response: $successBody")
}
.handleErrorWith {
case e: UnexpectedStatus if e.status == Status.Conflict =>
IO.println("Schema already exists")
case e => IO.println(e).flatMap(_ => IO.raiseError(e))
}
} yield ()
}
```
however when I go to create any table it all tries to use whatever default is in gravitino.conf, and if I remove that gravitino itself complains about the warehouse parameter not being set...
```scala
val spark = SparkSession
.builder()
.appName("create-tables")
.master("local[*]")
.config(
Map(
// Gravitino
"spark.plugins" -> "org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin",
"spark.sql.gravitino.uri" -> "http://gravitino:8090",
"spark.sql.gravitino.metalake" -> metalakeName,
"spark.sql.gravitino.enableIcebergSupport" -> "true"
// Iceberg catalog
// "spark.sql.catalog.mylake" -> "org.apache.gravitino.spark.connector.GravitinoSparkCatalog"
))
.getOrCreate()
spark.sql("""use mylake;""")
// spark.sql("""
// |CREATE DATABASE IF NOT EXISTS db
// |""".stripMargin)
spark.sql("""use db;""")
// Create nats gameserver events test table
spark
.sql(
s"""
|CREATE TABLE IF NOT EXISTS ${config.someEventName} (
| session_id STRING,
| position_x FLOAT,
| position_y FLOAT,
| position_z FLOAT,
| user_id STRING,
| time TIMESTAMP,
|)
|USING iceberg
|PARTITIONED BY (
| hours(time)
|);
|""".stripMargin
)
.show()
```
any ideas why this is happening or how to fix it?
Contributor guide
Research direction
Start with the SparkSession configuration and the catalog, schema, and table creation requests shown in the issue, then compare their per-metalake properties with the warehouse setting in gravitino.conf. Reproduce the two-metalake MinIO setup and verify that creating a table after selecting each metalake writes to its corresponding bucket rather than the default warehouse.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, scala
- Domain
- data-engineering, databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 30/100