diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_by_capital_use.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_by_capital_use.yml
deleted file mode 100644
index d1b9d543d8..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_by_capital_use.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Capital Expenses (by Capital Use)'
-product_short: 'capital_expenses_by_capital_use'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'fphd-jyyj'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_by_mode.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_by_mode.yml
deleted file mode 100644
index f8497ac57d..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_by_mode.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Capital Expenses (by Mode)'
-product_short: 'capital_expenses_by_mode'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: '2667-vitc'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_for_existing_service.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_for_existing_service.yml
deleted file mode 100644
index 01b3babd4c..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_for_existing_service.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Capital Expenses (for Existing Service)'
-product_short: 'capital_expenses_for_existing_service'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: '7kqv-yqbn'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_for_expansion_of_service.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_for_expansion_of_service.yml
deleted file mode 100644
index 47ef8bc86e..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_capital_expenses_for_expansion_of_service.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Capital Expenses (for Expansion of Service)'
-product_short: 'capital_expenses_for_expansion_of_service'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'nvgd-g6pj'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_mode_and_employee_type.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_mode_and_employee_type.yml
deleted file mode 100644
index 0df94f2dce..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_mode_and_employee_type.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Employees (by Mode and Employee Type)'
-product_short: 'employees_by_mode_and_employee_type'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'uyv8-9jek'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_fuel_and_energy_by_agency.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_fuel_and_energy_by_agency.yml
deleted file mode 100644
index a56a87413f..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_fuel_and_energy_by_agency.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Fuel and Energy (by Agency)'
-product_short: 'fuel_and_energy_by_agency'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'wwem-ata9'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_by_expense_type.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_by_expense_type.yml
deleted file mode 100644
index f2bd0e58d4..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_by_expense_type.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Funding Sources (by Expense Type)'
-product_short: 'funding_sources_by_expense_type'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: '4tmr-gwuu'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_directly_generated.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_directly_generated.yml
deleted file mode 100644
index 55a5b77411..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_directly_generated.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Funding Sources (Directly Generated)'
-product_short: 'funding_sources_directly_generated'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'yuaq-zdvc'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_federal.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_federal.yml
deleted file mode 100644
index 68ae7882be..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_federal.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Funding Sources (Federal)'
-product_short: 'funding_sources_federal'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'qpjk-b3zw'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_taxes_levied_by_agency.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_taxes_levied_by_agency.yml
deleted file mode 100644
index f34848139d..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_taxes_levied_by_agency.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Funding Sources (Taxes Levied by Agency)'
-product_short: 'funding_sources_taxes_levied_by_agency'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'c8k8-y2cj'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_maintenance_facilities_by_agency.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_maintenance_facilities_by_agency.yml
deleted file mode 100644
index f41d1a9bc6..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_maintenance_facilities_by_agency.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Maintenance Facilities (by Agency)'
-product_short: 'maintenance_facilities_by_agency'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 's68b-wvgx'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_function.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_function.yml
deleted file mode 100644
index 3cc547b46a..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_function.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Operating Expenses (by Function)'
-product_short: 'operating_expenses_by_function'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'dkxx-zjd6'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_function_and_agency.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_function_and_agency.yml
deleted file mode 100644
index 8868d01b93..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_function_and_agency.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Operating Expenses (by Function and Agency)'
-product_short: 'operating_expenses_by_function_and_agency'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'i5ki-dc58'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_type.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_type.yml
deleted file mode 100644
index 9c910af2ee..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_type.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Operating Expenses (by Type)'
-product_short: 'operating_expenses_by_type'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'j5uj-anzx'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_type_and_agency.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_type_and_agency.yml
deleted file mode 100644
index 792ea8ff69..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_operating_expenses_by_type_and_agency.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Operating Expenses (by Type and Agency)'
-product_short: 'operating_expenses_by_type_and_agency'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'i4ua-cjx4'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_mode_and_time_period.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_mode_and_time_period.yml
deleted file mode 100644
index 16379d2f65..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_mode_and_time_period.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Service (by Mode and Time Period)'
-product_short: 'service_by_mode_and_time_period'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'wwdp-t4re'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_stations_and_facilities_by_agency_and_facility_type.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_stations_and_facilities_by_agency_and_facility_type.yml
deleted file mode 100644
index ffe93e3fdf..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_stations_and_facilities_by_agency_and_facility_type.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Stations and Facilities (by Agency and Facility Type)'
-product_short: 'stations_and_facilities_by_agency_and_facility_type'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'aqct-knjk'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_stations_by_mode_and_age.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_stations_by_mode_and_age.yml
deleted file mode 100644
index 2ea26c5f79..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_stations_by_mode_and_age.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Stations (by Mode and Age)'
-product_short: 'stations_by_mode_and_age'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'wfz2-eft6'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_by_agency.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_by_agency.yml
deleted file mode 100644
index 663a3e53bd..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_by_agency.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Track & Roadway (by Agency)'
-product_short: 'track_and_roadway_by_agency'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'pvgq-a73e'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_by_mode.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_by_mode.yml
deleted file mode 100644
index f80589bab8..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_by_mode.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Track & Roadway (by Mode)'
-product_short: 'track_and_roadway_by_mode'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'fzbb-f6kc'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_guideway_age_distribution.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_guideway_age_distribution.yml
deleted file mode 100644
index 5e3e6007bb..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_track_and_roadway_guideway_age_distribution.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Track & Roadway (Guideway Age Distribution)'
-product_short: 'track_and_roadway_guideway_age_distribution'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'j9q7-53ae'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_vehicles_age_distribution.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_vehicles_age_distribution.yml
deleted file mode 100644
index 4bddfae2b2..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_vehicles_age_distribution.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Vehicles (Age Distribution)'
-product_short: 'vehicles_age_distribution'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: '6abt-uhgq'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_vehicles_type_count_by_agency.yml b/airflow/dags/sync_ntd_data_api/2022_reporting/2022_vehicles_type_count_by_agency.yml
deleted file mode 100644
index d07edcda18..0000000000
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_vehicles_type_count_by_agency.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: '2022'
-product_long: 'Vehicles (Type Count by Agency)'
-product_short: 'vehicles_type_count_by_agency'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: 'nimp-626k'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_breakdowns.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/breakdowns.yml
similarity index 54%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_breakdowns.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/breakdowns.yml
index f64671529e..3a1791b829 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_breakdowns.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/breakdowns.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Breakdowns'
-product_short: 'breakdowns'
+product: 'breakdowns'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: 'amkt-4ehs'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_breakdowns_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/breakdowns_by_agency.yml
similarity index 50%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_breakdowns_by_agency.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/breakdowns_by_agency.yml
index 91952fea23..3c6461324e 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_breakdowns_by_agency.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/breakdowns_by_agency.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Breakdowns (by Agency)'
-product_short: 'breakdowns_by_agency'
+product: 'breakdowns_by_agency'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: 'fk8n-qvag'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_by_capital_use.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_by_capital_use.yml
new file mode 100644
index 0000000000..a5b52d6d11
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_by_capital_use.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'capital_expenses_by_capital_use'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'fphd-jyyj'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_by_mode.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_by_mode.yml
new file mode 100644
index 0000000000..eef16e0b4b
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_by_mode.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'capital_expenses_by_mode'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: '2667-vitc'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_for_existing_service.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_for_existing_service.yml
new file mode 100644
index 0000000000..03e7e85290
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_for_existing_service.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'capital_expenses_for_existing_service'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: '7kqv-yqbn'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_for_expansion_of_service.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_for_expansion_of_service.yml
new file mode 100644
index 0000000000..ca5e4167ba
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/capital_expenses_for_expansion_of_service.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'capital_expenses_for_expansion_of_service'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'nvgd-g6pj'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_agency.yml
similarity index 50%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_agency.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_agency.yml
index fad41a490c..255ea95f91 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_agency.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_agency.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Employees (By Agency)'
-product_short: 'employees_by_agency'
+product: 'employees_by_agency'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: 'brbd-9azc'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_mode.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_mode.yml
similarity index 51%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_mode.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_mode.yml
index 8f63cfd0a5..21a91c3d15 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_employees_by_mode.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_mode.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Employees (By Mode)'
-product_short: 'employees_by_mode'
+product: 'employees_by_mode'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: 'wsxw-2rpq'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_mode_and_employee_type.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_mode_and_employee_type.yml
new file mode 100644
index 0000000000..f247147046
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/employees_by_mode_and_employee_type.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'employees_by_mode_and_employee_type'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'uyv8-9jek'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_fuel_and_energy.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/fuel_and_energy.yml
similarity index 52%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_fuel_and_energy.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/fuel_and_energy.yml
index 0db01e6e95..851b9e0367 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_fuel_and_energy.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/fuel_and_energy.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Fuel and Energy'
-product_short: 'fuel_and_energy'
+product: 'fuel_and_energy'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: '8ehq-7his'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/fuel_and_energy_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/fuel_and_energy_by_agency.yml
new file mode 100644
index 0000000000..8219da6e4e
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/fuel_and_energy_by_agency.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'fuel_and_energy_by_agency'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'wwem-ata9'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_by_expense_type.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_by_expense_type.yml
new file mode 100644
index 0000000000..36c079dfe6
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_by_expense_type.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'funding_sources_by_expense_type'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: '4tmr-gwuu'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_directly_generated.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_directly_generated.yml
new file mode 100644
index 0000000000..a2e9cc3886
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_directly_generated.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'funding_sources_directly_generated'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'yuaq-zdvc'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_federal.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_federal.yml
new file mode 100644
index 0000000000..331aa4aac0
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_federal.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'funding_sources_federal'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'qpjk-b3zw'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_local.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_local.yml
similarity index 50%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_local.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_local.yml
index 33ba4271a0..61de1f318a 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_local.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_local.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Funding Sources (Local)'
-product_short: 'funding_sources_local'
+product: 'funding_sources_local'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: '8tvb-ywj3'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_state.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_state.yml
similarity index 50%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_state.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_state.yml
index b0eeed59b3..a67524baa7 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_funding_sources_state.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_state.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Funding Sources (State)'
-product_short: 'funding_sources_state'
+product: 'funding_sources_state'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: 'dd43-h6wv'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_taxes_levied_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_taxes_levied_by_agency.yml
new file mode 100644
index 0000000000..2147afd0e9
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/funding_sources_taxes_levied_by_agency.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'funding_sources_taxes_levied_by_agency'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'c8k8-y2cj'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_maintenance_facilities.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/maintenance_facilities.yml
similarity index 50%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_maintenance_facilities.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/maintenance_facilities.yml
index 7cd9a502c6..00b191f417 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_maintenance_facilities.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/maintenance_facilities.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Maintenance Facilities'
-product_short: 'maintenance_facilities'
+product: 'maintenance_facilities'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: '9yj4-fiiz'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/maintenance_facilities_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/maintenance_facilities_by_agency.yml
new file mode 100644
index 0000000000..5ec0d6087c
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/maintenance_facilities_by_agency.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'maintenance_facilities_by_agency'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 's68b-wvgx'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_metrics.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/metrics.yml
similarity index 56%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_metrics.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/metrics.yml
index 030cbc853b..a4b2e15c68 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_metrics.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/metrics.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Metrics'
-product_short: 'metrics'
+product: 'metrics'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: 'ekg5-frzt'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_function.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_function.yml
new file mode 100644
index 0000000000..05df595b6f
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_function.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'operating_expenses_by_function'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'dkxx-zjd6'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_function_and_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_function_and_agency.yml
new file mode 100644
index 0000000000..a709d83a03
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_function_and_agency.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'operating_expenses_by_function_and_agency'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'i5ki-dc58'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_type.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_type.yml
new file mode 100644
index 0000000000..d62e1f8a44
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_type.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'operating_expenses_by_type'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'j5uj-anzx'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_type_and_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_type_and_agency.yml
new file mode 100644
index 0000000000..a369d196de
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/operating_expenses_by_type_and_agency.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'operating_expenses_by_type_and_agency'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'i4ua-cjx4'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_agency.yml
similarity index 51%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_agency.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_agency.yml
index 8163edb719..309b558bd5 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_agency.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_agency.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Service (by Agency)'
-product_short: 'service_by_agency'
+product: 'service_by_agency'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: '6y83-7vuw'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_mode.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_mode.yml
similarity index 52%
rename from airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_mode.yml
rename to airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_mode.yml
index 8544bcfb32..4596edb9f7 100644
--- a/airflow/dags/sync_ntd_data_api/2022_reporting/2022_service_by_mode.yml
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_mode.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: '2022'
-product_long: 'Service (by Mode)'
-product_short: 'service_by_mode'
+product: 'service_by_mode'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: '4fir-qbim'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_mode_and_time_period.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_mode_and_time_period.yml
new file mode 100644
index 0000000000..9e7b5132d5
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/service_by_mode_and_time_period.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'service_by_mode_and_time_period'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'wwdp-t4re'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/stations_and_facilities_by_agency_and_facility_type.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/stations_and_facilities_by_agency_and_facility_type.yml
new file mode 100644
index 0000000000..867e1fe875
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/stations_and_facilities_by_agency_and_facility_type.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'stations_and_facilities_by_agency_and_facility_type'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'aqct-knjk'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/stations_by_mode_and_age.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/stations_by_mode_and_age.yml
new file mode 100644
index 0000000000..a9bf9ac8a1
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/stations_by_mode_and_age.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'stations_by_mode_and_age'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'wfz2-eft6'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_by_agency.yml
new file mode 100644
index 0000000000..f1b5eef729
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_by_agency.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'track_and_roadway_by_agency'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'pvgq-a73e'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_by_mode.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_by_mode.yml
new file mode 100644
index 0000000000..cb8d35ee43
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_by_mode.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'track_and_roadway_by_mode'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'fzbb-f6kc'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_guideway_age_distribution.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_guideway_age_distribution.yml
new file mode 100644
index 0000000000..788a67f796
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/track_and_roadway_guideway_age_distribution.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'track_and_roadway_guideway_age_distribution'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'j9q7-53ae'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/vehicles_age_distribution.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/vehicles_age_distribution.yml
new file mode 100644
index 0000000000..644a8eb74b
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/vehicles_age_distribution.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'vehicles_age_distribution'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: '6abt-uhgq'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/annual_reporting/2022/vehicles_type_count_by_agency.yml b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/vehicles_type_count_by_agency.yml
new file mode 100644
index 0000000000..dab3584afb
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/annual_reporting/2022/vehicles_type_count_by_agency.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: '2022'
+product: 'vehicles_type_count_by_agency'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: 'nimp-626k'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_FRA_regulated_mode_major_security_events.yml b/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_FRA_regulated_mode_major_security_events.yml
deleted file mode 100644
index fd0ba9d86c..0000000000
--- a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_FRA_regulated_mode_major_security_events.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: 'historical'
-product_long: 'FRA Regulated Mode Major Security Events'
-product_short: 'FRA_regulated_mode_major_security_events'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: '65fa-qbkf'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_monthly_modal_time_series_safety_and_service.yml b/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_monthly_modal_time_series_safety_and_service.yml
deleted file mode 100644
index 8e3e8ac542..0000000000
--- a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_monthly_modal_time_series_safety_and_service.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: 'historical'
-product_long: 'Monthly Modal Time Series (Safety and Service)'
-product_short: 'monthly_modal_time_series_safety_and_service'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: '65fa-qbkf'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_nonmajor_safety_and_security_events.yml b/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_nonmajor_safety_and_security_events.yml
deleted file mode 100644
index 9ecb580ef6..0000000000
--- a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_nonmajor_safety_and_security_events.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-operator: operators.NtdDataProductAPIOperator
-
-bucket: 'gs://calitp-ntd-api-products'
-year: 'historical'
-product_long: 'Non-Major Safety and Security Events'
-product_short: 'nonmajor_safety_and_security_events'
-root_url: 'https://data.transportation.gov/resource/'
-endpoint_id: '9ivb-8ae9'
-file_format: '.csv'
diff --git a/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/FRA_regulated_mode_major_security_events.yml b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/FRA_regulated_mode_major_security_events.yml
new file mode 100644
index 0000000000..01f31a0a3d
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/FRA_regulated_mode_major_security_events.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: 'historical'
+product: 'FRA_regulated_mode_major_security_events'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: '65fa-qbkf'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_major_safety_events.yml b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/major_safety_events.yml
similarity index 52%
rename from airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_major_safety_events.yml
rename to airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/major_safety_events.yml
index b024c5898a..8996761deb 100644
--- a/airflow/dags/sync_ntd_data_api/safety_service_and_security/historical_major_safety_events.yml
+++ b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/major_safety_events.yml
@@ -1,9 +1,7 @@
 operator: operators.NtdDataProductAPIOperator
 
-bucket: 'gs://calitp-ntd-api-products'
 year: 'historical'
-product_long: 'Major Safety Events'
-product_short: 'major_safety_events'
+product: 'major_safety_events'
 root_url: 'https://data.transportation.gov/resource/'
 endpoint_id: '9ivb-8ae9'
-file_format: '.csv'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/monthly_modal_time_series_safety_and_service.yml b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/monthly_modal_time_series_safety_and_service.yml
new file mode 100644
index 0000000000..8aeacf7d21
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/monthly_modal_time_series_safety_and_service.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: 'historical'
+product: 'monthly_modal_time_series_safety_and_service'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: '65fa-qbkf'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/nonmajor_safety_and_security_events.yml b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/nonmajor_safety_and_security_events.yml
new file mode 100644
index 0000000000..38dd1124ad
--- /dev/null
+++ b/airflow/dags/sync_ntd_data_api/safety_service_and_security_historical/nonmajor_safety_and_security_events.yml
@@ -0,0 +1,7 @@
+operator: operators.NtdDataProductAPIOperator
+
+year: 'historical'
+product: 'nonmajor_safety_and_security_events'
+root_url: 'https://data.transportation.gov/resource/'
+endpoint_id: '9ivb-8ae9'
+file_format: '.json'
diff --git a/airflow/dags/sync_ntd_data_xlsx/ridership/historical_complete_monthly_ridership_with_adjustments_and_estimates.yml b/airflow/dags/sync_ntd_data_xlsx/ridership_historical/complete_monthly_ridership_with_adjustments_and_estimates.yml
similarity index 100%
rename from airflow/dags/sync_ntd_data_xlsx/ridership/historical_complete_monthly_ridership_with_adjustments_and_estimates.yml
rename to airflow/dags/sync_ntd_data_xlsx/ridership_historical/complete_monthly_ridership_with_adjustments_and_estimates.yml
diff --git a/airflow/dags/sync_ntd_data_xlsx/ridership/historical_raw_monthly_ridership_no_adjustments_or_estimates.yml b/airflow/dags/sync_ntd_data_xlsx/ridership_historical/raw_monthly_ridership_no_adjustments_or_estimates.yml
similarity index 100%
rename from airflow/dags/sync_ntd_data_xlsx/ridership/historical_raw_monthly_ridership_no_adjustments_or_estimates.yml
rename to airflow/dags/sync_ntd_data_xlsx/ridership_historical/raw_monthly_ridership_no_adjustments_or_estimates.yml
diff --git a/airflow/dags/sync_ntd_data_xlsx/safety_service_and_security/historical_safety_and_security_major_event_time_series.yml b/airflow/dags/sync_ntd_data_xlsx/safety_service_and_security_historical/safety_and_security_major_event_time_series.yml
similarity index 100%
rename from airflow/dags/sync_ntd_data_xlsx/safety_service_and_security/historical_safety_and_security_major_event_time_series.yml
rename to airflow/dags/sync_ntd_data_xlsx/safety_service_and_security_historical/safety_and_security_major_event_time_series.yml
diff --git a/airflow/dags/sync_ntd_data_xlsx/safety_service_and_security/historical_safety_and_security_time_series_threshold_adjusted.yml b/airflow/dags/sync_ntd_data_xlsx/safety_service_and_security_historical/safety_and_security_time_series_threshold_adjusted.yml
similarity index 100%
rename from airflow/dags/sync_ntd_data_xlsx/safety_service_and_security/historical_safety_and_security_time_series_threshold_adjusted.yml
rename to airflow/dags/sync_ntd_data_xlsx/safety_service_and_security_historical/safety_and_security_time_series_threshold_adjusted.yml
diff --git a/airflow/plugins/operators/scrape_ntd_api.py b/airflow/plugins/operators/scrape_ntd_api.py
index 556b573037..25c22c0b7e 100644
--- a/airflow/plugins/operators/scrape_ntd_api.py
+++ b/airflow/plugins/operators/scrape_ntd_api.py
@@ -1,158 +1,125 @@
-# import csv
+# import os
 import gzip
-
-# import json
 import logging
-import os
-
-# import re
-from typing import Optional
+from typing import ClassVar, List  # , Optional
 
+import pandas as pd  # type: ignore
 import pendulum
 import requests
-from calitp_data_infra.storage import get_fs  # type: ignore
-from pydantic import BaseModel  # , Json
-
-from airflow.models import BaseOperator  # type: ignore
 
+# import typer
+from calitp_data_infra.storage import PartitionedGCSArtifact, get_fs  # type: ignore
+from pydantic import HttpUrl, parse_obj_as
 
-def write_to_log(logfilename):
-    """
-    Creates a logger object that outputs to a log file, to the filename specified,
-    and also streams to console.
-    """
-    logger = logging.getLogger(__name__)
-    logger.setLevel(logging.INFO)
-    formatter = logging.Formatter(
-        "%(asctime)s:%(levelname)s: %(message)s", datefmt="%y-%m-%d %H:%M:%S"
-    )
-    file_handler = logging.FileHandler(logfilename)
-    file_handler.setFormatter(formatter)
-    stream_handler = logging.StreamHandler()
-    stream_handler.setFormatter(formatter)
+from airflow.models import BaseOperator  # type: ignore
 
-    if not logger.hasHandlers():
-        logger.addHandler(file_handler)
-        logger.addHandler(stream_handler)
+# Restore for prod
+# API_BUCKET = os.environ["CALITP_BUCKET__NTD_API_DATA_PRODUCTS"]
 
-    return logger
+API_BUCKET = "gs://calitp-ntd-api-products"
 
 
-class NtdDataProductAPIExtract(BaseModel):
+class NtdDataProductAPIExtract(PartitionedGCSArtifact):
+    bucket: ClassVar[str]
     year: str
-    product_short: str
+    product: str
+    execution_ts: pendulum.DateTime = pendulum.now()
+    dt: pendulum.Date = execution_ts.date()
     root_url: str
     endpoint_id: str
     file_format: str
-    data: Optional[bytes]
+    partition_names: ClassVar[List[str]] = ["dt", "execution_ts"]
 
-    logger: Optional[logging.Logger]
-    extract_time: Optional[pendulum.DateTime]
+    @property
+    def table(self) -> str:
+        return self.product
 
-    logger = write_to_log("load_ntd_api_output.log")
-    extract_time = pendulum.now()
+    @property
+    def filename(self) -> str:
+        return self.table
 
-    # pydantic doesn't know dataframe type
-    # see https://stackoverflow.com/a/69200069
     class Config:
         arbitrary_types_allowed = True
 
     def fetch_from_ntd_api(self):
         """ """
 
-        self.logger.info(
-            f"Downloading NTD data for {self.year} / {self.product_short}."
-        )
-
-        url = self.root_url + self.endpoint_id + self.file_format + "?$limit=5000000"
-        response = requests.get(url).content
-        # decoded_response = response.decode("utf-8")
-        # load_response_json = json.loads(decode_respose)
-        # json_filepath = f"{self.endpoint_id}_data.jsonl"
+        logging.info(f"Downloading NTD data for {self.year} / {self.product}.")
 
-        if response is None or len(response) == 0:
-            self.logger.info(
-                f"There is no data to download for {self.year} / {self.product_short}. Ending pipeline."
+        try:
+            url = (
+                self.root_url + self.endpoint_id + self.file_format + "?$limit=5000000"
             )
-            pass
-        else:
-            # self.data = decoded_response
-            self.data = response
 
-            self.logger.info(
-                f"Downloaded {self.product_short} data for {self.year} with {len(self.data)} rows!"
-            )
+            validated_url = parse_obj_as(HttpUrl, url)
 
-    def make_hive_path(self, product_short: str, bucket: str):
-        if not self.extract_time:
-            raise ValueError(
-                "An extract time must be set before a hive path can be generated."
-            )
-        return os.path.join(
-            bucket,
-            f"{self.product_short}_{self.year}",
-            f"year={self.extract_time.format('YYYY')}",
-            f"dt={self.extract_time.to_date_string()}",
-            f"ts={self.extract_time.to_iso8601_string()}",
-            f"{self.product_short}_{self.year}.csv.gz",
-            # f"{self.product_short}_{self.year}_raw.jsonl.gz",
-        )
+            response = requests.get(validated_url).content
 
-    def save_to_gcs(self, fs, bucket):
-        """ """
-        hive_path = self.make_hive_path(self.product_short, bucket)
-        self.logger.info(f"Uploading to GCS at {hive_path}")
+            if response is None or len(response) == 0:
+                logging.info(
+                    f"There is no data to download for {self.year} / {self.product}. Ending pipeline."
+                )
 
-        if self.data is None:  # or (len(self.data) == 0):
-            self.logger.info(
-                f"There is no data for {self.year} / {self.product_short}, not saving anything. Pipeline exiting."
-            )
-            pass
-        else:
-            fs.pipe(hive_path, gzip.compress(self.data))  # .encode()
-        return hive_path
+                pass
+            else:
+                logging.info(
+                    f"Downloaded {self.product} data for {self.year} with {len(response)} rows!"
+                )
+
+                return response
+
+        except requests.exceptions.RequestException as e:
+            logging.info(f"An error occurred: {e}")
+
+            raise
+
+
+class CSVExtract(NtdDataProductAPIExtract):
+    bucket = API_BUCKET
 
 
 class NtdDataProductAPIOperator(BaseOperator):
-    template_fields = ("bucket",)
+    template_fields = ("year", "product", "root_url", "endpoint_id", "file_format")
 
     def __init__(
         self,
-        bucket,
         year,
-        product_short,
+        product,
         root_url,
         endpoint_id,
         file_format,
         **kwargs,
     ):
-        self.bucket = bucket
+        self.year = year
+        self.product = product
+        self.root_url = root_url
+        self.endpoint_id = endpoint_id
+        self.file_format = file_format
         """An operator that downloads all data from a NTD API
-            and saves it as one JSON file hive-partitioned by date in Google Cloud
-            Storage (GCS). DELETE:(Each org's data will be in 1 row, and for each separate table in the API,
-            a nested column will hold all of it's data.)
-
-        Args:
-            bucket (str): GCS bucket where the scraped NTD data will be saved.
+            and saves it as one CSV file hive-partitioned by date in Google Cloud
         """
 
-        # if self.bucket and os.environ["AIRFLOW_ENV"] == "development":
-        #     self.bucket = re.sub(r"gs://([\w-]+)", r"gs://test-\1", self.bucket)
-
-        # Instantiating an instance of the NtdDataProductAPIExtract()
-        self.extract = NtdDataProductAPIExtract(
-            year=year,
-            product_short=product_short,
-            root_url=root_url,
-            endpoint_id=endpoint_id,
-            file_format=file_format,
+        # Save CSV files to the bucket
+        self.extract = CSVExtract(
+            year=self.year,
+            product=self.product + "/" + self.year,
+            root_url=self.root_url,
+            endpoint_id=self.endpoint_id,
+            file_format=self.file_format,
+            filename=f"{self.year}__{self.product}.jsonl.gz",
         )
 
         super().__init__(**kwargs)
 
     def execute(self, **kwargs):
-        fs = get_fs()
+        api_content = self.extract.fetch_from_ntd_api()
+
+        decode_api_content = api_content.decode("utf-8")
+
+        df = pd.read_json(decode_api_content)
+
+        self.gzipped_content = gzip.compress(
+            df.to_json(orient="records", lines=True).encode()
+        )
 
-        self.extract.fetch_from_ntd_api()
-        # inserts into xcoms
-        return self.extract.save_to_gcs(fs, self.bucket)
+        self.extract.save_content(fs=get_fs(), content=self.gzipped_content)
diff --git a/airflow/plugins/operators/scrape_ntd_xlsx.py b/airflow/plugins/operators/scrape_ntd_xlsx.py
index a2d65471a4..fa0817e0b8 100644
--- a/airflow/plugins/operators/scrape_ntd_xlsx.py
+++ b/airflow/plugins/operators/scrape_ntd_xlsx.py
@@ -20,8 +20,8 @@
 from airflow.models import BaseOperator  # type: ignore
 
 # Restore for prod
-# RAW_XLSX_BUCKET = os.environ["RAW_XLSX_BUCKET"]
-# CLEAN_XLSX_BUCKET = os.environ["CLEAN_XLSX_BUCKET"]
+# RAW_XLSX_BUCKET = os.environ["CALITP_BUCKET__NTD_XLSX_DATA_PRODUCTS__RAW"]
+# CLEAN_XLSX_BUCKET = os.environ["CALITP_BUCKET__NTD_XLSX_DATA_PRODUCTS__CLEAN"]
 
 RAW_XLSX_BUCKET = "gs://calitp-ntd-xlsx-products-raw"
 CLEAN_XLSX_BUCKET = "gs://calitp-ntd-xlsx-products-clean"