Lake /query reads era5_daily through Iceberg metadata #44
2 changed files with 44 additions and 25 deletions
|
|
@ -53,10 +53,12 @@ RUN useradd --system --create-home --uid 10001 thermograph \
|
|||
&& chown -R thermograph:thermograph /app /state
|
||||
|
||||
USER thermograph
|
||||
# Bake DuckDB's httpfs extension AS THE RUNTIME USER — extensions install to
|
||||
# the invoking user's ~/.duckdb, and a root-time bake leaves uid 10001 unable
|
||||
# to LOAD it (seen live: prod lake /query 500'd on exactly this).
|
||||
RUN python -c "import duckdb; duckdb.connect().execute('INSTALL httpfs')"
|
||||
# Bake DuckDB's httpfs + iceberg extensions AS THE RUNTIME USER — extensions
|
||||
# install to the invoking user's ~/.duckdb, and a root-time bake leaves uid
|
||||
# 10001 unable to LOAD them (seen live: prod lake /query 500'd on exactly
|
||||
# this). iceberg powers the era5_daily view via table metadata instead of a
|
||||
# bucket-wide glob LIST.
|
||||
RUN python -c "import duckdb; c = duckdb.connect(); c.execute('INSTALL httpfs'); c.execute('INSTALL iceberg')"
|
||||
WORKDIR /app
|
||||
|
||||
ENV PORT=8137 \
|
||||
|
|
|
|||
|
|
@ -135,34 +135,51 @@ _FORBIDDEN = re.compile(
|
|||
|
||||
|
||||
def _connect():
|
||||
"""A fresh in-memory DuckDB with the lake views registered. Per-request:
|
||||
connections are cheap, and a throwaway one means a query can't leave state
|
||||
behind for the next. Returns None when no lake is configured."""
|
||||
"""A fresh in-memory DuckDB with the lake views registered. Returns None
|
||||
when no lake is configured.
|
||||
|
||||
Bucket mode reads era5_daily through the ICEBERG table, not a hive glob:
|
||||
a glob over era5/daily/ must LIST every object under it (~1k per tile —
|
||||
minutes at a few hundred tiles, seen live timing out /query at 390),
|
||||
while Iceberg gets the file list from table metadata in a handful of
|
||||
reads. Local mode (tests, small trees) keeps the plain hive glob — no
|
||||
Iceberg table exists in the fixtures and listing a local dir is free."""
|
||||
import duckdb # noqa: PLC0415 - only the lake role pays the import
|
||||
|
||||
mode, cfg = _source()
|
||||
if mode == "none":
|
||||
return None
|
||||
con = duckdb.connect()
|
||||
if mode == "local":
|
||||
daily_glob = os.path.join(era5lake.local_dir(), era5lake.PREFIX,
|
||||
"daily/*/*/*/*.parquet")
|
||||
con.execute(f"""
|
||||
CREATE VIEW era5_daily AS
|
||||
SELECT * FROM read_parquet('{daily_glob}', hive_partitioning=1)""")
|
||||
manifest_src = os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY)
|
||||
elif mode == "bucket":
|
||||
daily_glob = f"s3://{cfg['bucket']}/{era5lake.PREFIX}/daily/*/*/*/*.parquet"
|
||||
manifest_src = f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}"
|
||||
else:
|
||||
return None
|
||||
con = duckdb.connect()
|
||||
if mode == "bucket":
|
||||
con.execute("LOAD httpfs") # baked into the image at build time
|
||||
host = cfg["endpoint"].split("://", 1)[-1]
|
||||
con.execute(f"SET s3_endpoint='{host}'")
|
||||
con.execute("SET s3_url_style='path'")
|
||||
con.execute(f"SET s3_region='{cfg['region']}'")
|
||||
con.execute("SET s3_access_key_id=?", [cfg["access_key"]])
|
||||
con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]])
|
||||
con.execute(f"""
|
||||
CREATE VIEW era5_daily AS
|
||||
SELECT * FROM read_parquet('{daily_glob}', hive_partitioning=1)""")
|
||||
con.execute(f"CREATE VIEW manifest AS SELECT * FROM read_parquet('{manifest_src}')")
|
||||
con.execute(
|
||||
f"CREATE VIEW manifest AS SELECT * FROM read_parquet('{manifest_src}')")
|
||||
return con
|
||||
|
||||
con.execute("LOAD httpfs") # both baked into the image at build time,
|
||||
con.execute("LOAD iceberg") # as the runtime user (see Dockerfile)
|
||||
host = cfg["endpoint"].split("://", 1)[-1]
|
||||
con.execute(f"SET s3_endpoint='{host}'")
|
||||
con.execute("SET s3_url_style='path'")
|
||||
con.execute(f"SET s3_region='{cfg['region']}'")
|
||||
con.execute("SET s3_access_key_id=?", [cfg["access_key"]])
|
||||
con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]])
|
||||
# Newest metadata JSON wins (one single-page LIST of metadata/ — cheap);
|
||||
# pyiceberg names them NNNNN-uuid.metadata.json, so lexicographic max is
|
||||
# the numeric max.
|
||||
meta = con.execute(
|
||||
f"SELECT max(file) FROM glob('s3://{cfg['bucket']}/iceberg/era5_daily/"
|
||||
f"metadata/*.metadata.json')").fetchone()[0]
|
||||
if meta is None:
|
||||
return None # lake bucket exists but the Iceberg table doesn't yet
|
||||
con.execute(f"CREATE VIEW era5_daily AS SELECT * FROM iceberg_scan('{meta}')")
|
||||
con.execute(f"""CREATE VIEW manifest AS SELECT * FROM
|
||||
read_parquet('s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}')""")
|
||||
return con
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue