Enable dbt python models via serverless Dataproc and create example model

[atvaccaro]

Description

I had to enable Google Private Access on our default VPC, and also grant access in the prod project (BigQuery and GCS) to the staging default compute service account since local dbt runs use the staging project.

Still waiting on dbt-labs/dbt-bigquery#636 but we can fork if necessary
Blocked on dbt-labs/dbt-bigquery#609

We had a meeting today (2023-05-31) to review the tech spec, so I'd like to merge this ahead of a to-be-scheduled pairing session to test an initial model.

Type of change

• Bug fix (non-breaking change which fixes an issue)
• New feature
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• Documentation
• agencies.yml

How has this been tested?

I had created and tested an example model, but we don't actually want/need it. Pasting the source code here for posterity; it ran on our custom image provided to Dataproc.

Screenshots (optional)

import geopandas as gpd  # noqa: F401
import pandas as pd  # noqa: F401
import pyspark.sql.functions as F
import pyspark.sql.types as T
from shapely import wkt
from shapely.geometry import LineString


@F.udf(returnType=T.FloatType())
def point_array_to_linestring(arr):
    if len(arr) == 1:
        return 0
    return LineString([wkt.loads(p) for p in arr]).length


def model(dbt, session):
    dbt.config(materialized="incremental", packages=["pandas", "geopandas"])

    df = dbt.ref("dim_shapes_arrays_wkt")

    if dbt.is_incremental and session.catalog.tableExists(repr(dbt.this)):
        # only new rows compared to max in current table
        max_from_this = f"select max(_feed_valid_from) from `{dbt.this}`"
        df = df.filter(df._feed_valid_from > session.sql(max_from_this).collect()[0][0])

    df = df.withColumn(
        "length_meters", point_array_to_linestring(F.col("pt_array_wkt"))
    )

    return df

[github-actions]

Warehouse report 📦

New models 🌱

mart.gtfs.dim_shapes_arrays_enriched
mart.gtfs.dim_shapes_arrays_wkt

Changed models 🔀

mart.gtfs_quality.fct_daily_organization_combined_guideline_checks
mart.gtfs_quality.fct_daily_service_combined_guideline_checks
intermediate.gtfs_quality.guidelines_checks.int_gtfs_quality__modification_date_present_rt

DAG

[lauriemerrell]

Still waiting on dbt-labs/dbt-bigquery#636 but we can fork if necessary

Is this actually a blocker to merge?

Would it be useful to add some documentation about this functionality to the warehouse README? Are there any links we could add about where to check if you need to troubleshoot any of this setup?

[atvaccaro]

Is this actually a blocker to merge?

No, but developer experience will be poor; I gave this caveat to tiffany and eric in our tech spec meeting yesterday.

I've added some other documentation!