jtcohen6
commented
2 years ago @danfran Thanks for opening the issue, and apologies for getting back to you here (and over in https://github.com/dbt-labs/dbt-spark/issues/215).
Background
The reason we use show table extended in <schema> like '*' is to build the adapter cache at the start of every run. This is how dbt can know which models already exist, and (crucially) whether they exist as a view or table, so that subsequent calls to get_relation within materializations (e.g. here and here) can be cache lookups rather than redundant, slow introspective queries.
We found we need to use show table extended in <schema> like '*', instead of the much less verbose (and more performant) show tables in <schema>, because show tables does not include the relation type (view or table), only whether the object is temporary. On other databases, this kind of metadata is easy and fast to access, but Apache Spark lacks a real information schema.
(A neat side effect of having to use show table extended in <schema> like '*' is that we can resolve get_columns_in_relation from the adapter cache, when available, which we cannot do on other adapters. This is purely a bonus, though, and it would be worth it to give up this capability in exchange for faster, more reliable cache construction.)
I know that show table extended in <schema> like '*' can be very slow, especially when accessing metadata for schemas with thousands of objects. This feels especially frustrating when most of those objects are not relevant to dbt! As a best practice, we highly recommend that you create dbt objects in a dedicated schema, separate from other processes. I understand it isn't always possible to isolate source tables, but this only impacts catalog querying (docs generate) rather than cache construction (run etc), since schemas with sources are catalogued but not cached.
Proposed change
Okay, now that we've gotten that background out of the way—I think you're onto something here:
So I can think ideally to replace the method:
show table extended in <schema> like '*'with something like:
show table extended in <schema> like ('<table1>|<table2>|…')
I think this might just be doable!
There's a base adapter method, _get_cache_schemas, that the adapter uses to identify which schemas it ought to cache. That list is derived from the enabled, non-ephemeral relations in the project. Normally, the list includes identifier-less relations, because we just need to run a single metadata query for the entire schema. The list of schema-only identifiers is then passed into the list_relations_without_caching method, i.e. the list_relations_without_caching macro, i.e. show table extended in <schema> like '*'.
But what if we instead passed in a schema relation that did have an identifier? That identifier could look like '<table1>|<table2>|…', and the list_relations_without_caching macro could use it to run a more targeted metadata query.
Any methods defined in BaseAdapter can be overridden within SparkAdapter, so even though _get_cache_schemas isn't defined there now, we can reimplement it.
Here's my rough attempt:
def _get_cache_schemas(self, manifest: Manifest) -> Set[BaseRelation]:
"""Get the set of schema relations that the cache logic needs to
populate. This means only executable nodes are included.
"""
# the cache only cares about executable nodes
relations = [
self.Relation.create_from(self.config, node) # keep the identifier
for node in manifest.nodes.values()
if (
node.is_relational and not node.is_ephemeral_model
)
]
# group up relations by common schema
import collections
relmap = collections.defaultdict(list)
for r in relations:
relmap[r.schema].append(r)
# create a single relation for each schema
# set the identifier to a '|' delimited string of relation names, or '*'
schemas = [
self.Relation.create(
schema=schema,
identifier=(
'|'.join(r.identifier for r in rels)
# there's probably some limit to how many we can include by name
if len(rels) < 100 else '*'
)
) for schema, rels in relmap.items()
]
return schemasThen, the macro becomes:
{% macro spark__list_relations_without_caching(relation) %}
{% call statement('list_relations_without_caching', fetch_result=True) -%}
show table extended in {{ relation.schema }} like '{{ relation.identifier or "*" }}'
{% endcall %}
{% do return(load_result('list_relations_without_caching').table) %}
{% endmacro %}That works for me locally:
show table extended in dev_jerco like 'my_first_dbt_model|my_second_dbt_model'Next steps
- What do you think of an approach like that? Would you be interested in taking that and running with it?
- Should we also try to do something similar for
_get_catalog_schemas, to speed updocs generate? - I'm going to transfer this issue to the
dbt-sparkrepo, since (I believe) that's where we can make all the code changes we need. - Semi-related: In the process of teasing apart this functionality, I noticed that the
add_schema_to_cachemethod doesn't seem to be called/used anywhere. It might be a holdover from a long time ago. It's always a good idea to remove totally unused code.
danfran
ssimeonov
jeremyyeo
crystalro0
ryanClift-sd
TalkWIthKeyboard
spenaustin
boxysean
theromsh
tinolyu
VShkaberda
pan3793
despewerba
vinjai
MatatatDespe
Fokko
daniel-gomez-mng
AlanCortesK
Describe the feature
I am currently facing an issue using DBT with Spark on AWS/Glue/EMR environment as discussed already in https://github.com/dbt-labs/dbt-spark/issues/215 (but already raised here https://github.com/dbt-labs/dbt-spark/issues/93).
The current issue is about the adapter's method
list_relations_without_caching:https://github.com/dbt-labs/dbt/blob/HEAD/core/dbt/include/global_project/macros/adapters/common.sql#L240
which in the Spark Adapter implementation is:
https://github.com/dbt-labs/dbt-spark/blob/a8a85c54d10920af1c5efcbb4d2a51eb7cfcad11/dbt/include/spark/macros/adapters.sql#L133-L139
In this case you can see that the command
show table extended in {{ relation }} like '*'is executed. It will force Spark to go through all the tables info in the schema (as Spark has not Information Schema Layer) in a sort of "discovery mode" and this approach produces two main issues:1) Bad performance: some environments can have hundreds or even thousands of tables generated not only by DBT but also by other processes in the same schema. In that case this operation can be very costly, especially when you have different DBT processes that run some updates at different times on a few tables.
2) Instability, as I verified in AWS/Glue/EMR environment, where you can have views without "S3 Location" defined, like an Athena/Presto view, that will make crash a DBT process running SparkSql on EMR with errors like:
Describe alternatives you've considered
I do not see the reason why DBT process should care of the "rest of the world" like the Athena views from before or tables created from other processes that are in the same schema.
So I can think ideally to replace the method:
show table extended in <schema> like '*'with something like:
show table extended in <schema> like ('<table1>|<table2>|…')where my
<table1>,<table2>, etc. are determined automatically when I run a command likedbt run --models my_folderwhere
my_foldercontains the files:table1.sql,table2.sql, etcbut from the current method interface, only the schema params can be passed.
Two questions here: How can I infer automatically the name of the tables involved when a command like
dbt run --models my_folderrun and how can I pass them eventually to thelist_relations_without_caching?Additional context
I found it relevant for Spark on AWS environment but can be potentially a similar issue for other implementations.
Who will this benefit?
On DBT's slack channel I talked to another used "affected" by similar issue, but probably whoever is going to use Spark in distributed environment can be affected by this (AWS and non).
Are you interested in contributing this feature?
Sure, both coding and testing.