-
Notifications
You must be signed in to change notification settings - Fork 320
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Materialized views for lineage events graph (#2891)
* Lineage metrics table and left joined generated query. * Removing log. * Query formatting. * Weekly metric query. * Saving update for weekly and daily metrics. * Making adjustments to storage to build materialized views from the `lineage_events` table * Adding refresh job and daily mat view. * Fixing table name. * Fixing tests. * Source code headers * Deferring materialized view updates until the hour strikes. * Fixing related to time intervals. * Code review feedback addressed. * Adding header. --------- Co-authored-by: Willy Lulciuc <willy@datakin.com>
- Loading branch information
Showing
13 changed files
with
337 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| /* | ||
| * Copyright 2018-2024 contributors to the Marquez project | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package marquez.api; | ||
|
|
||
| import static javax.ws.rs.core.MediaType.APPLICATION_JSON; | ||
|
|
||
| import com.codahale.metrics.annotation.ExceptionMetered; | ||
| import com.codahale.metrics.annotation.ResponseMetered; | ||
| import com.codahale.metrics.annotation.Timed; | ||
| import javax.ws.rs.GET; | ||
| import javax.ws.rs.Path; | ||
| import javax.ws.rs.Produces; | ||
| import javax.ws.rs.QueryParam; | ||
| import javax.ws.rs.core.Response; | ||
| import lombok.NonNull; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import marquez.api.models.Period; | ||
| import marquez.service.ServiceFactory; | ||
| import marquez.service.StatsService; | ||
|
|
||
| @Slf4j | ||
| @Path("/api/v1/stats") | ||
| public class StatsResource { | ||
|
|
||
| private final StatsService StatsService; | ||
|
|
||
| public StatsResource(@NonNull final ServiceFactory serviceFactory) { | ||
| this.StatsService = serviceFactory.getStatsService(); | ||
| } | ||
|
|
||
| @Timed | ||
| @ResponseMetered | ||
| @ExceptionMetered | ||
| @GET | ||
| @Produces(APPLICATION_JSON) | ||
| @Path("/lineage-events") | ||
| public Response getStats(@QueryParam("period") Period period) { | ||
|
|
||
| return (Period.DAY.equals(period) | ||
| ? Response.ok(StatsService.getLastDayLineageMetrics()).build() | ||
| : Period.WEEK.equals(period) | ||
| ? Response.ok(StatsService.getLastWeekLineageMetrics()).build() | ||
| : Response.status(Response.Status.BAD_REQUEST).entity("Invalid period").build()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| /* | ||
| * Copyright 2018-2024 contributors to the Marquez project | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package marquez.api.models; | ||
|
|
||
| public enum Period { | ||
| DAY, | ||
| WEEK | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * Copyright 2018-2024 contributors to the Marquez project | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package marquez.db; | ||
|
|
||
| import java.util.List; | ||
| import marquez.db.mappers.LineageMetricRowMapper; | ||
| import marquez.db.models.LineageMetric; | ||
| import org.jdbi.v3.sqlobject.config.RegisterRowMapper; | ||
| import org.jdbi.v3.sqlobject.statement.SqlQuery; | ||
|
|
||
| @RegisterRowMapper(LineageMetricRowMapper.class) | ||
| public interface StatsDao extends BaseDao { | ||
|
|
||
| @SqlQuery( | ||
| """ | ||
| WITH hour_series AS ( | ||
| SELECT | ||
| generate_series( | ||
| DATE_TRUNC('hour', NOW() - INTERVAL '1 day'), | ||
| DATE_TRUNC('hour', NOW() - INTERVAL '1 hour'), | ||
| '1 hour' | ||
| ) AS start_interval | ||
| ) | ||
| SELECT | ||
| hs.start_interval, | ||
| hs.start_interval + INTERVAL '1 hour' AS end_interval, | ||
| COALESCE(hourly_metrics.fail, 0) AS fail, | ||
| COALESCE(hourly_metrics.start, 0) AS start, | ||
| COALESCE(hourly_metrics.complete, 0) AS complete, | ||
| COALESCE(hourly_metrics.abort, 0) AS abort | ||
| FROM | ||
| hour_series hs | ||
| LEFT JOIN | ||
| lineage_events_by_type_hourly_view hourly_metrics | ||
| ON | ||
| hs.start_interval = hourly_metrics.start_interval | ||
| ORDER BY | ||
| hs.start_interval; | ||
| """) | ||
| List<LineageMetric> getLastDayMetrics(); | ||
|
|
||
| @SqlQuery( | ||
| """ | ||
| WITH day_series AS ( | ||
| SELECT | ||
| generate_series( | ||
| DATE_TRUNC('day', NOW() - INTERVAL '7 days'), -- Start 7 days ago | ||
| DATE_TRUNC('day', NOW()) - INTERVAL '1 day', -- End at the start of today | ||
| '1 day' | ||
| ) AS start_interval | ||
| ) | ||
| SELECT | ||
| ds.start_interval, | ||
| ds.start_interval + INTERVAL '1 day' AS end_interval, | ||
| COALESCE(mv.fail, 0) AS fail, | ||
| COALESCE(mv.start, 0) AS start, | ||
| COALESCE(mv.complete, 0) AS complete, | ||
| COALESCE(mv.abort, 0) AS abort | ||
| FROM | ||
| day_series ds | ||
| LEFT JOIN | ||
| lineage_events_by_type_daily_view mv | ||
| ON | ||
| ds.start_interval = mv.start_interval | ||
| ORDER BY | ||
| ds.start_interval; | ||
| """) | ||
| List<LineageMetric> getLastWeekMetrics(); | ||
| } |
31 changes: 31 additions & 0 deletions
31
api/src/main/java/marquez/db/mappers/LineageMetricRowMapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| /* | ||
| * Copyright 2018-2024 contributors to the Marquez project | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package marquez.db.mappers; | ||
|
|
||
| import static marquez.db.Columns.intOrThrow; | ||
| import static marquez.db.Columns.timestampOrThrow; | ||
|
|
||
| import java.sql.ResultSet; | ||
| import java.sql.SQLException; | ||
| import lombok.NonNull; | ||
| import marquez.db.Columns; | ||
| import marquez.db.models.LineageMetric; | ||
| import org.jdbi.v3.core.mapper.RowMapper; | ||
| import org.jdbi.v3.core.statement.StatementContext; | ||
|
|
||
| public final class LineageMetricRowMapper implements RowMapper<LineageMetric> { | ||
| @Override | ||
| public LineageMetric map(@NonNull ResultSet results, @NonNull StatementContext context) | ||
| throws SQLException { | ||
| return new LineageMetric( | ||
| timestampOrThrow(results, Columns.START_INTERVAL), | ||
| timestampOrThrow(results, Columns.END_INTERVAL), | ||
| intOrThrow(results, Columns.FAIL), | ||
| intOrThrow(results, Columns.START), | ||
| intOrThrow(results, Columns.COMPLETE), | ||
| intOrThrow(results, Columns.ABORT)); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| /* | ||
| * Copyright 2018-2024 contributors to the Marquez project | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package marquez.db.models; | ||
|
|
||
| import java.time.Instant; | ||
| import lombok.Getter; | ||
| import lombok.NonNull; | ||
| import lombok.Value; | ||
|
|
||
| @Value | ||
| public class LineageMetric { | ||
| @Getter @NonNull Instant startInterval; | ||
| @Getter @NonNull Instant endInterval; | ||
| @Getter @NonNull Integer fail; | ||
| @Getter @NonNull Integer start; | ||
| @Getter @NonNull Integer complete; | ||
| @Getter @NonNull Integer abort; | ||
| } |
77 changes: 77 additions & 0 deletions
77
api/src/main/java/marquez/jobs/MaterializeViewRefresherJob.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * Copyright 2018-2024 contributors to the Marquez project | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package marquez.jobs; | ||
|
|
||
| import com.google.common.util.concurrent.AbstractScheduledService; | ||
| import io.dropwizard.lifecycle.Managed; | ||
| import java.time.Duration; | ||
| import java.time.LocalTime; | ||
| import lombok.NonNull; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.jdbi.v3.core.Jdbi; | ||
|
|
||
| /** A job that refreshes materialized views on a fixed schedule in Marquez. */ | ||
| @Slf4j | ||
| public class MaterializeViewRefresherJob extends AbstractScheduledService implements Managed { | ||
|
|
||
| private final int FREQUENCY = 60; | ||
| private final Scheduler fixedRateScheduler; | ||
| private final Jdbi jdbi; | ||
|
|
||
| public MaterializeViewRefresherJob(@NonNull final Jdbi jdbi) { | ||
| // Connection to database retention policy will be applied. | ||
| this.jdbi = jdbi; | ||
|
|
||
| // Define fixed schedule and delay until the hour strikes. | ||
| int MINUTES_IN_HOUR = 60; | ||
| LocalTime now = LocalTime.now(); | ||
| int minutesRemaining = | ||
| MINUTES_IN_HOUR - now.getMinute(); // Get the remaining minutes in the hour | ||
| Duration duration = Duration.ofMinutes(minutesRemaining); | ||
| this.fixedRateScheduler = | ||
| Scheduler.newFixedRateSchedule(duration, Duration.ofMinutes(FREQUENCY)); | ||
| } | ||
|
|
||
| @Override | ||
| protected Scheduler scheduler() { | ||
| return fixedRateScheduler; | ||
| } | ||
|
|
||
| @Override | ||
| public void start() throws Exception { | ||
| startAsync().awaitRunning(); | ||
| log.info("Refreshing materialized views every '{}' mins.", FREQUENCY); | ||
| } | ||
|
|
||
| @Override | ||
| protected void runOneIteration() { | ||
| try { | ||
| log.info("Refreshing materialized views..."); | ||
| jdbi.useHandle( | ||
| handle -> { | ||
| handle.execute("REFRESH MATERIALIZED VIEW lineage_events_by_type_hourly_view"); | ||
| }); | ||
| log.info("Materialized view `lineage_events_by_type_hourly_view` refreshed."); | ||
|
|
||
| if (java.time.LocalTime.now().getHour() == 0) { | ||
| jdbi.useHandle( | ||
| handle -> { | ||
| handle.execute("REFRESH MATERIALIZED VIEW lineage_events_by_type_daily_view"); | ||
| }); | ||
| log.info("Materialized view `lineage_events_by_type_daily_view` refreshed."); | ||
| } | ||
|
|
||
| } catch (Exception error) { | ||
| log.error("Failed to materialize views. Retrying on next run...", error); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void stop() throws Exception { | ||
| log.info("Stopping materialized views job..."); | ||
| stopAsync().awaitTerminated(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.