Skip to content

Commit 022f86d

Browse files
CHEUKFUNGWUclaude
CHEUKFUNGWU
and
claude
committed
Add Apache Gravitino metadata catalog integration
Register a Gravitino service that federates the Paimon lakehouse and Doris serving catalogs under the ai_observability metalake. Flink and Spark sessions use the Gravitino-managed catalog when GRAVITINO_URI is set, with standalone direct-Paimon fallbacks preserved for local runs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 90cb634 commit 022f86d

9 files changed

Lines changed: 205 additions & 18 deletions

File tree

‎Makefile‎

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
.PHONY: test lint pipeline infra-light infra-serving infra-dashboard infra-stop init-superset dashboard-stop seed-data flink-up flink-submit flink-jobs flink-savepoint flink-cancel flink-restore batch-backfill sync-doris health demo demo-streaming demo-serving clean
1+
.PHONY: test lint pipeline infra-light infra-serving infra-dashboard infra-stop init-superset dashboard-stop seed-data flink-up flink-submit flink-submit-standalone flink-jobs flink-savepoint flink-cancel flink-restore batch-backfill sync-doris health demo demo-streaming demo-serving gravitino-up gravitino-status gravitino-catalogs clean
22

33
test:
44
uv run pytest -v
@@ -10,9 +10,10 @@ pipeline:
1010
uv run python -m scripts.spark_paimon_backfill --input data/raw/mock_llm_requests/events.jsonl
1111

1212
infra-light:
13-
docker compose up -d postgres kafka flink-jobmanager flink-taskmanager
13+
docker compose up -d postgres kafka gravitino flink-jobmanager flink-taskmanager
1414
scripts/prepare_flink_warehouse.sh
1515
scripts/create_kafka_topics.sh
16+
bash scripts/init_gravitino.sh
1617

1718
infra-serving:
1819
docker compose up -d doris-fe doris-be doris-init
@@ -51,6 +52,17 @@ flink-submit:
5152
flink/sql/20_build_dwd_from_kafka_ods.sql \
5253
flink/sql/30_build_dws_from_dwd.sql
5354

55+
flink-submit-standalone:
56+
scripts/run_flink_sql_sequence.sh \
57+
flink/sql/00_catalogs_standalone.sql \
58+
flink/sql/01_source_postgres_cdc.sql \
59+
flink/sql/02_ods_kafka_tables.sql \
60+
flink/sql/03_dwd_paimon_tables.sql \
61+
flink/sql/04_dws_paimon_tables.sql \
62+
flink/sql/10_ingest_ods_to_kafka.sql \
63+
flink/sql/20_build_dwd_from_kafka_ods.sql \
64+
flink/sql/30_build_dws_from_dwd.sql
65+
5466
flink-jobs:
5567
curl -s http://localhost:8081/jobs/overview
5668

@@ -90,5 +102,17 @@ demo-streaming:
90102
demo-serving:
91103
scripts/run_serving_demo.sh
92104

105+
gravitino-up:
106+
docker compose up -d gravitino
107+
@echo "Waiting for Gravitino to start..."
108+
@sleep 5
109+
bash scripts/init_gravitino.sh
110+
111+
gravitino-status:
112+
curl -sf http://localhost:8090/api/version | python3 -m json.tool
113+
114+
gravitino-catalogs:
115+
curl -sf http://localhost:8090/api/metalakes/ai_observability/catalogs | python3 -m json.tool
116+
93117
clean:
94118
rm -rf data/

‎config/gravitino/gravitino.conf‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
# Gravitino server configuration for AI Observability Lakehouse
2+
gravitino.server.webserver.host = 0.0.0.0
3+
gravitino.server.webserver.httpPort = 8090
4+
gravitino.server.shutdown.timeout = 3000

‎docker-compose.yml‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,19 @@ services:
2020
- postgres_data:/var/lib/postgresql/data
2121
- ./sql/source_postgres_schema.sql:/docker-entrypoint-initdb.d/01_source_postgres_schema.sql:ro
2222

23+
gravitino:
24+
image: apache/gravitino:1.2.0
25+
container_name: ai-observability-gravitino
26+
ports:
27+
- "8090:8090"
28+
environment:
29+
GRAVITINO_MEM: "-Xms512m -Xmx512m -XX:MaxMetaspaceSize=256m"
30+
volumes:
31+
- paimon_warehouse:/workspace/data/paimon
32+
- ./config/gravitino/gravitino.conf:/opt/gravitino/conf/gravitino.conf:ro
33+
depends_on:
34+
- postgres
35+
2336
flink-jobmanager:
2437
build:
2538
context: .
@@ -50,6 +63,7 @@ services:
5063
depends_on:
5164
- postgres
5265
- kafka
66+
- gravitino
5367

5468
flink-taskmanager:
5569
image: ai-observability-flink:1.20-paimon
@@ -75,6 +89,7 @@ services:
7589
- paimon_warehouse:/workspace/data/paimon
7690
depends_on:
7791
- flink-jobmanager
92+
- gravitino
7893

7994
flink-sql-client:
8095
image: ai-observability-flink:1.20-paimon

‎docker/flink/Dockerfile‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,10 @@ ARG PAIMON_VERSION=1.2.0
66
ARG FLINK_CDC_VERSION=3.2.1
77
ARG FLINK_SHADED_HADOOP_VERSION=2.8.3-10.0
88
ARG FLINK_KAFKA_VERSION=3.3.0-1.20
9+
ARG GRAVITINO_VERSION=1.2.0
910

1011
USER root
1112

12-
# Download connector JARs.
13-
# The Hadoop uber JAR is required by Paimon even for local filesystem.
14-
# The CDC connector + Hadoop uber JAR have a commons-collections classloader
15-
# conflict — resolved by adding classloader.parent-first-patterns.additional:
16-
# org.apache.commons.collections in flink-conf / FLINK_PROPERTIES.
1713
RUN set -eux; \
1814
curl -fSL \
1915
"https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.20/${PAIMON_VERSION}/paimon-flink-1.20-${PAIMON_VERSION}.jar" \
@@ -27,6 +23,9 @@ RUN set -eux; \
2723
curl -fSL \
2824
"https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/${FLINK_KAFKA_VERSION}/flink-sql-connector-kafka-${FLINK_KAFKA_VERSION}.jar" \
2925
-o "/opt/flink/lib/flink-sql-connector-kafka-${FLINK_KAFKA_VERSION}.jar"; \
26+
curl -fSL \
27+
"https://repo.maven.apache.org/maven2/org/apache/gravitino/gravitino-flink-connector-runtime-1.18_2.12/${GRAVITINO_VERSION}/gravitino-flink-connector-runtime-1.18_2.12-${GRAVITINO_VERSION}.jar" \
28+
-o "/opt/flink/lib/gravitino-flink-connector-runtime-1.18_2.12-${GRAVITINO_VERSION}.jar"; \
3029
mkdir -p /workspace/data/paimon && chown -R flink:flink /workspace/data/paimon
3130

3231
USER flink

‎flink/sql/00_catalogs.sql‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
1-
-- Flink SQL catalog bootstrap for the stream-batch lakehouse.
1+
-- Flink SQL catalog bootstrap — Gravitino-managed Paimon catalog.
22

33
CREATE CATALOG paimon_lake WITH (
4-
'type' = 'paimon',
5-
'warehouse' = 'file:///workspace/data/paimon'
4+
'type' = 'gravitino',
5+
'gravitino.metalake' = 'ai_observability',
6+
'gravitino.uri' = 'http://gravitino:8090'
67
);
78

89
CREATE DATABASE IF NOT EXISTS paimon_lake.ods;

‎flink/sql/00_catalogs_standalone.sql‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
-- Flink SQL catalog bootstrap — direct Paimon (no Gravitino).
2+
3+
CREATE CATALOG paimon_lake WITH (
4+
'type' = 'paimon',
5+
'warehouse' = 'file:///workspace/data/paimon'
6+
);
7+
8+
CREATE DATABASE IF NOT EXISTS paimon_lake.ods;
9+
CREATE DATABASE IF NOT EXISTS paimon_lake.dwd;
10+
CREATE DATABASE IF NOT EXISTS paimon_lake.dws;
11+
CREATE DATABASE IF NOT EXISTS paimon_lake.dim;
12+
CREATE DATABASE IF NOT EXISTS paimon_lake.ads;
13+
14+
USE CATALOG default_catalog;
15+
USE default_database;

‎scripts/init_gravitino.sh‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
#!/usr/bin/env bash
2+
set -euo pipefail
3+
4+
GRAVITINO_URL="${GRAVITINO_URL:-http://localhost:8090}"
5+
METALAKE="ai_observability"
6+
7+
# Wait for Gravitino server to be ready
8+
gravitino_ready=false
9+
for i in $(seq 1 30); do
10+
if curl -sf "${GRAVITINO_URL}/api/version" > /dev/null 2>&1; then
11+
gravitino_ready=true
12+
break
13+
fi
14+
echo "Waiting for Gravitino server... (${i}/30)"
15+
sleep 2
16+
done
17+
18+
if [ "$gravitino_ready" = false ]; then
19+
echo "ERROR: Gravitino server did not start within 60 seconds" >&2
20+
exit 1
21+
fi
22+
23+
# Create metalake
24+
curl -sf -X POST "${GRAVITINO_URL}/api/metalakes" \
25+
-H "Content-Type: application/json" \
26+
-d "{
27+
\"name\": \"${METALAKE}\",
28+
\"comment\": \"AI Observability Lakehouse metadata\"
29+
}" || echo "Metalake already exists"
30+
31+
# Register Paimon catalog
32+
curl -sf -X POST "${GRAVITINO_URL}/api/metalakes/${METALAKE}/catalogs" \
33+
-H "Content-Type: application/json" \
34+
-d '{
35+
"name": "paimon_lake",
36+
"type": "RELATIONAL",
37+
"provider": "lakehouse-paimon",
38+
"comment": "Paimon stream-batch lakehouse tables (ODS/DWD/DWS/DIM/ADS)",
39+
"properties": {
40+
"catalog-backend": "filesystem",
41+
"warehouse": "file:///workspace/data/paimon"
42+
}
43+
}' || echo "Paimon catalog already exists"
44+
45+
# Register Doris catalog
46+
curl -sf -X POST "${GRAVITINO_URL}/api/metalakes/${METALAKE}/catalogs" \
47+
-H "Content-Type: application/json" \
48+
-d '{
49+
"name": "doris_serving",
50+
"type": "RELATIONAL",
51+
"provider": "jdbc-doris",
52+
"comment": "Doris OLAP serving layer",
53+
"properties": {
54+
"jdbc-url": "jdbc:mysql://doris-fe:9030",
55+
"jdbc-driver": "com.mysql.jdbc.Driver",
56+
"jdbc-user": "root",
57+
"jdbc-password": ""
58+
}
59+
}' || echo "Doris catalog already exists"
60+
61+
echo "Gravitino initialization complete."

‎scripts/spark_utils.py‎

Lines changed: 58 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,7 @@ def build_spark_session(app_name: str) -> SparkSession:
2020
)
2121

2222

23-
def build_paimon_spark_session(app_name: str, warehouse: str | None = None) -> SparkSession:
24-
os.environ.pop("SPARK_HOME", None)
25-
os.environ["PYSPARK_PYTHON"] = sys.executable
26-
os.environ["PYSPARK_DRIVER_PYTHON"] = sys.executable
27-
23+
def _reset_spark_state() -> None:
2824
active_session = SparkSession.getActiveSession()
2925
if active_session is not None:
3026
active_session.stop()
@@ -38,8 +34,60 @@ def build_paimon_spark_session(app_name: str, warehouse: str | None = None) -> S
3834
SparkContext._gateway = None
3935
SparkContext._jvm = None
4036

37+
38+
def build_paimon_spark_session(app_name: str, warehouse: str | None = None) -> SparkSession:
39+
os.environ.pop("SPARK_HOME", None)
40+
os.environ["PYSPARK_PYTHON"] = sys.executable
41+
os.environ["PYSPARK_DRIVER_PYTHON"] = sys.executable
42+
43+
_reset_spark_state()
44+
45+
gravitino_uri = os.environ.get("GRAVITINO_URI")
46+
47+
if gravitino_uri:
48+
return _build_gravitino_paimon_session(app_name, gravitino_uri)
49+
50+
return _build_direct_paimon_session(app_name, warehouse)
51+
52+
53+
def _build_gravitino_paimon_session(app_name: str, gravitino_uri: str) -> SparkSession:
54+
gravitino_metalake = os.environ.get("GRAVITINO_METALAKE", "ai_observability")
55+
paimon_package = os.environ.get(
56+
"PAIMON_SPARK_PACKAGE", "org.apache.paimon:paimon-spark-3.5:1.2.0"
57+
)
58+
gravitino_package = os.environ.get(
59+
"GRAVITINO_SPARK_PACKAGE",
60+
"org.apache.gravitino:gravitino-spark-connector-runtime-3.5_2.12:1.2.0",
61+
)
62+
63+
return (
64+
SparkSession.builder.appName(app_name)
65+
.master("local[*]")
66+
.config("spark.sql.session.timeZone", "UTC")
67+
.config("spark.jars.packages", f"{paimon_package},{gravitino_package}")
68+
.config("spark.jars.excludes", "org.slf4j:slf4j-api")
69+
.config("spark.sql.gravitino.uri", gravitino_uri)
70+
.config("spark.sql.gravitino.metalake", gravitino_metalake)
71+
.config("spark.sql.gravitino.enablePaimonSupport", "true")
72+
.config(
73+
"spark.sql.catalog.paimon_lake",
74+
"org.apache.gravitino.spark.connector.GravitinoCatalog",
75+
)
76+
.config(
77+
"spark.sql.extensions",
78+
"org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions",
79+
)
80+
.config("spark.driver.host", "127.0.0.1")
81+
.config("spark.driver.bindAddress", "127.0.0.1")
82+
.getOrCreate()
83+
)
84+
85+
86+
def _build_direct_paimon_session(app_name: str, warehouse: str | None) -> SparkSession:
4187
wh = warehouse or os.environ.get("PAIMON_WAREHOUSE", "data/paimon")
42-
paimon_package = os.environ.get("PAIMON_SPARK_PACKAGE", "org.apache.paimon:paimon-spark-3.5:1.2.0")
88+
paimon_package = os.environ.get(
89+
"PAIMON_SPARK_PACKAGE", "org.apache.paimon:paimon-spark-3.5:1.2.0"
90+
)
4391

4492
return (
4593
SparkSession.builder.appName(app_name)
@@ -49,7 +97,10 @@ def build_paimon_spark_session(app_name: str, warehouse: str | None = None) -> S
4997
.config("spark.jars.excludes", "org.slf4j:slf4j-api")
5098
.config("spark.sql.catalog.paimon_lake", "org.apache.paimon.spark.SparkCatalog")
5199
.config("spark.sql.catalog.paimon_lake.warehouse", wh)
52-
.config("spark.sql.extensions", "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
100+
.config(
101+
"spark.sql.extensions",
102+
"org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions",
103+
)
53104
.config("spark.driver.host", "127.0.0.1")
54105
.config("spark.driver.bindAddress", "127.0.0.1")
55106
.getOrCreate()

‎tests/test_flink_sql_assets.py‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@
4747

4848
EXPECTED_FLINK_SQL_FILES = [
4949
"00_catalogs.sql",
50+
"00_catalogs_standalone.sql",
5051
"01_source_postgres_cdc.sql",
5152
"02_ods_kafka_tables.sql",
5253
"03_dwd_paimon_tables.sql",
@@ -73,14 +74,25 @@ def test_catalog_sql_defines_paimon_lake():
7374
sql = read_asset("flink/sql/00_catalogs.sql")
7475

7576
assert "CREATE CATALOG paimon_lake" in sql
76-
assert "'type' = 'paimon'" in sql
77+
assert "'type' = 'gravitino'" in sql
78+
assert "'gravitino.metalake' = 'ai_observability'" in sql
79+
assert "'gravitino.uri' = 'http://gravitino:8090'" in sql
7780
assert "CREATE DATABASE IF NOT EXISTS paimon_lake.ods" in sql
7881
assert "CREATE DATABASE IF NOT EXISTS paimon_lake.dwd" in sql
7982
assert "CREATE DATABASE IF NOT EXISTS paimon_lake.dws" in sql
8083
assert "CREATE DATABASE IF NOT EXISTS paimon_lake.dim" in sql
8184
assert "CREATE DATABASE IF NOT EXISTS paimon_lake.ads" in sql
8285

8386

87+
def test_standalone_catalog_sql_uses_direct_paimon():
88+
sql = read_asset("flink/sql/00_catalogs_standalone.sql")
89+
90+
assert "CREATE CATALOG paimon_lake" in sql
91+
assert "'type' = 'paimon'" in sql
92+
assert "'warehouse' = 'file:///workspace/data/paimon'" in sql
93+
assert "CREATE DATABASE IF NOT EXISTS paimon_lake.ods" in sql
94+
95+
8496
def test_source_sql_uses_postgres_cdc_connector():
8597
sql = read_asset("flink/sql/01_source_postgres_cdc.sql")
8698

@@ -296,17 +308,22 @@ def test_flink_dockerfile_installs_paimon_postgres_and_kafka_connectors():
296308
assert "ARG FLINK_CDC_VERSION=3.2.1" in dockerfile
297309
assert "ARG FLINK_SHADED_HADOOP_VERSION=2.8.3-10.0" in dockerfile
298310
assert "ARG FLINK_KAFKA_VERSION=3.3.0-1.20" in dockerfile
311+
assert "ARG GRAVITINO_VERSION=1.2.0" in dockerfile
299312
assert "paimon-flink-1.20" in dockerfile
300313
assert "flink-sql-connector-postgres-cdc" in dockerfile
301314
assert "flink-sql-connector-kafka" in dockerfile
302315
assert "flink-shaded-hadoop-2-uber" in dockerfile
316+
assert "gravitino-flink-connector-runtime-1.18_2.12" in dockerfile
303317
assert "/opt/flink/lib" in dockerfile
304318

305319

306320
def test_compose_defines_stream_batch_runtime_services():
307321
compose = read_asset("docker-compose.yml")
308322

309323
assert "kafka:" in compose
324+
assert "gravitino:" in compose
325+
assert "apache/gravitino:1.2.0" in compose
326+
assert "8090:8090" in compose
310327
assert "flink-jobmanager:" in compose
311328
assert "flink-taskmanager:" in compose
312329
assert "flink-sql-client:" in compose

0 commit comments

Comments
 (0)