fix: add more specific DELTA partition for lamp ingestion,

runners/run_ingest_s3_files.py

@@ -4,4 +4,4 @@
 
 # metadata_queue, rds_process = start_rds_writer_process()
 
-ingest_s3_files(None, bucket_filter="lamp")
+ingest_s3_files(None, bucket_filter="lamp/delta")

src/lamp_py/ingestion/convert_gtfs_rt_fullset.py

@@ -110,6 +110,8 @@ def convert(self) -> None:
 
                 os.makedirs(Path(local_path).parent, exist_ok=True)
 
+                table = table.drop_columns(["year", "month", "day"])
+
                 self.write_local_pq_partition(table, local_path)
 
                 # in backfill mode, we don't want to move files around in s3 -

src/lamp_py/ingestion/ingest_gtfs.py

@@ -24,7 +24,7 @@
     NoImplException,
     IgnoreIngestion,
 )
-from lamp_py.runtime_utils.remote_files import LAMP, S3_ERROR, S3_INCOMING, S3_SPRINGBOARD, S3Location
+from lamp_py.runtime_utils.remote_files import DELTA, LAMP, S3_ERROR, S3_INCOMING, S3_SPRINGBOARD, S3Location
 from lamp_py.ingestion.utils import group_sort_file_list
 from lamp_py.ingestion.compress_gtfs.gtfs_to_parquet import gtfs_to_parquet
 
@@ -63,7 +63,7 @@ def ingest_gtfs_archive(metadata_queue: Queue[Optional[str]]) -> None:
     logger.log_complete()
 
 
-def ingest_s3_files(metadata_queue: Queue[Optional[str]], bucket_filter: str = LAMP) -> None:
+def ingest_s3_files(metadata_queue: Queue[Optional[str]], bucket_filter: str = DELTA) -> None:
     """
     Get all of the filepaths currently in the incoming bucket, sort them into
     batches of similar gtfs-rt files, convert each batch into tables, write the
@@ -146,7 +146,7 @@ def ingest_s3_files(metadata_queue: Queue[Optional[str]], bucket_filter: str = L
     logger.log_complete()
 
 
-def ingest_gtfs(metadata_queue: Queue[Optional[str]], bucket_filter: str = LAMP) -> None:
+def ingest_gtfs(metadata_queue: Queue[Optional[str]], bucket_filter: str = DELTA) -> None:
     """
     Ingest all gtfs file types
 

src/lamp_py/ingestion/pipeline.py

@@ -14,7 +14,7 @@
 from lamp_py.ingestion.ingest_gtfs import ingest_gtfs
 from lamp_py.ingestion.glides import ingest_glides_events
 
-from lamp_py.runtime_utils.remote_files import LAMP
+from lamp_py.runtime_utils.remote_files import DELTA
 from lamp_py.utils.clear_folder import clear_folder
 
 logging.getLogger().setLevel("INFO")
@@ -45,7 +45,7 @@ def main() -> None:
     while True:
         process_logger = ProcessLogger(process_name="main")
         process_logger.log_start()
-        bucket_filter = LAMP
+        bucket_filter = DELTA
         check_for_sigterm(metadata_queue, rds_process)
 
         ingest_gtfs(metadata_queue, bucket_filter=bucket_filter)

src/lamp_py/runtime_utils/remote_files.py

@@ -11,6 +11,7 @@
 
 # prefix constants
 LAMP = "lamp"
+DELTA = os.path.join(LAMP, "delta")
 TM = os.path.join(LAMP, "TM")
 TABLEAU = os.path.join(LAMP, "tableau")