Add tag to limit concurrency for high memory use assets

src/pudl/analysis/state_demand.py

@@ -449,7 +449,7 @@ def melt_ferc714_hourly_demand_matrix(
             ),
         ),
     },
-    op_tags={"datasource": "ferc714"},
+    op_tags={"memory-use": "high"},
 )
 def _out_ferc714__hourly_demand_matrix(
     context, _out_ferc714__hourly_pivoted_demand_matrix: pd.DataFrame
@@ -473,7 +473,10 @@ def _out_ferc714__hourly_demand_matrix(
     return df
 
 
-@asset(compute_kind="Python")
+@asset(
+    compute_kind="Python",
+    op_tags={"memory-use": "high"},
+)
 def _out_ferc714__hourly_imputed_demand(
     _out_ferc714__hourly_demand_matrix: pd.DataFrame,
     _out_ferc714__utc_offset: pd.DataFrame,

src/pudl/etl/__init__.py

@@ -209,18 +209,13 @@ def _get_keys_from_assets(
     "epacems_io_manager": epacems_io_manager,
 }
 
-# By default, limit CEMS and FERC 714 processing concurrency to prevent memory overload.
+# Limit the number of concurrent workers when launch assets that use a lot of memory.
 default_tag_concurrency_limits = [
     {
-        "key": "datasource",
-        "value": "epacems",
+        "key": "memory-use",
+        "value": "high",
         "limit": 2,
     },
-    {
-        "key": "datasource",
-        "value": "ferc714",
-        "limit": 1,
-    },
 ]
 default_config = pudl.helpers.get_dagster_execution_config(
     tag_concurrency_limits=default_tag_concurrency_limits

src/pudl/etl/cli.py

@@ -148,18 +148,13 @@ def pudl_etl(
         },
     }
 
-    # By default, limit CEMS and FERC 714 processing concurrency to prevent memory overload.
+    # Limit the number of concurrent workers when launch assets that use a lot of memory.
     tag_concurrency_limits = [
         {
-            "key": "datasource",
-            "value": "epacems",
+            "key": "memory-use",
+            "value": "high",
             "limit": 2,
         },
-        {
-            "key": "datasource",
-            "value": "ferc714",
-            "limit": 1,
-        },
     ]
 
     run_config.update(

src/pudl/etl/epacems_assets.py

@@ -58,7 +58,7 @@ def get_years_from_settings(context):
 
 @op(
     required_resource_keys={"datastore", "dataset_settings"},
-    tags={"datasource": "epacems"},
+    tags={"memory-use": "high"},
 )
 def process_single_year(
     context,