[Python] spark_catalog requires a single-part namespace in dbt python incremental model

Discussão em 'Python' iniciado por Stack, Novembro 19, 2024.

  1. Stack

    Stack Membro Participativo

    Description:


    • Using the dbt functionality that allows one to create a python model, I created a model that reads from some BigQuery table, performs some calculations and writes back to BigQuery.


    • It uses dataproc (serverless submission mode) to submit the model as a PySpark job.
    Issue


    When one runs the model with a table materialization, everything works as intended. However, when trying to use the incremental materialization and using the property dbt.this to acess the location of the current model, the code breaks.

    Here's the faulty code:

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


    Here's the error output:

    df = df.filter(df.created_at >= session.sql(max_from_this).collect()[0][0])
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 1034, in sql
    File "/usr/lib/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1321, in __call__
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 196, in deco
    pyspark.sql.utils.AnalysisException: spark_catalog requires a single-part namespace, but got [x, y]

    References:


    I've tried a different approach where instead of querying the table using session.sql I first retrieve the table using session.table and then perform a simple computation with the dataframe returned, but the error was still present

    Continue reading...

Compartilhe esta Página