I am not able to use dbt.this
on Python incremental models.
I’m trying to implement incremental Python models in dbt, but I’m running into issues when using the dbt.this
keyword due to a hyphen in my BigQuery project name (marketing-analytics
).
Main code:
if dbt.is_incremental:
# Does not work
max_from_this = f"select max(updated_at_new) from {dbt.this}" # <-- problem
df_raw = dbt.ref("interesting_data").filter(
F.col("updated_at_new") >=session.sql(max_from_this).collect()[0][0]
)
# Works
df_raw = dbt.ref("interesting_data").filter(
F.col("updated_at_new") >= F.date_add(F.current_timestamp(), F.lit(-1))
)
else:
df_core_users = dbt.ref("int_core__users")
Error I've got:
Possibly unquoted identifier marketing-analytics detected. Please consider quoting with backquotes `marketing-analytics`
max_from_this = f"select max(updated_at_new) from `{dbt.this}`"
and
max_from_this=f"select max(updated_at_new) from `{dbt.this.database}.{dbt.this.schema}.{dbt.this.identifier}`"
Error: Table or view not found `marketing-analytics.test_dataset.posts`
Even though this table exists on BigQuery...
max_from_this = f"select max(updated_at_new) from f"`{dbt.this.database}`.`{dbt.this.schema}`.`{dbt.this.identifier}`"
Error: spark_catalog requires a single-part namespace, but got [marketing-analytics, test_dataset]
Solved it by playing with bigquery-spark
connector with session. It is really unconvenient.
if dbt.is_incremental:
current_table = (
session
.read
.format("bigquery")
.option("table", f"{dbt.this.schema}.{dbt.this.identifier}")
.load()
)