(fix) tableau output fixes - datatypes, archive bucket, and sort order (#507)

* sort objects from list_s3 query by date to get the "last X files" logic to work properly

* add correct str representation for pyarrow types to convert to the proper hyper type

* output devgreen data to the private archive bucket instead of public

Author: huangh

runners/run_gtfs_rt_parquet_converter.py

@@ -22,10 +22,18 @@ def start_devgreen_gtfs_rt_parquet_updates_local() -> None:
     parquet_update_jobs: List[HyperJob] = [HyperDevGreenGtfsRtVehiclePositions, HyperDevGreenGtfsRtTripUpdates]
 
     for job in parquet_update_jobs:
-        # breakpoint()
         job.run_parquet(None)
-        # outs = job.create_local_hyper()
-        # print(outs)
+
+
+def start_devgreen_gtfs_rt_parquet_updates_local_hyper() -> None:
+    """Run all gtfs_rt Parquet Update jobs"""
+
+    parquet_update_jobs: List[HyperJob] = [HyperDevGreenGtfsRtVehiclePositions, HyperDevGreenGtfsRtTripUpdates]
+
+    for job in parquet_update_jobs:
+        job.run_parquet(None)
+        outs = job.create_local_hyper()
+        print(outs)
 
 
 def start_gtfs_rt_parquet_updates_local() -> None:
@@ -41,5 +49,6 @@ def start_gtfs_rt_parquet_updates_local() -> None:
 
 
 if __name__ == "__main__":
-    start_gtfs_rt_parquet_updates_local()
-    start_devgreen_gtfs_rt_parquet_updates_local()
+    # start_gtfs_rt_parquet_updates_local()
+    # start_devgreen_gtfs_rt_parquet_updates_local()
+    start_devgreen_gtfs_rt_parquet_updates_local_hyper()

src/lamp_py/runtime_utils/remote_files.py

@@ -165,12 +165,12 @@ def s3_uri(self) -> str:
 
 # DEVGREEN
 tableau_devgreen_rt_vehicle_positions_lightrail_60_day = S3Location(
-    bucket=S3_PUBLIC,
+    bucket=S3_ARCHIVE,
     prefix=os.path.join(TABLEAU, "devgreen-gtfs-rt", "LAMP_DEVGREEN_RT_VehiclePositions_LR_60_day.parquet"),
 )
 # light rail output file - to be converted to .hyper
 tableau_devgreen_rt_trip_updates_lightrail_60_day = S3Location(
-    bucket=S3_PUBLIC,
+    bucket=S3_ARCHIVE,
     prefix=os.path.join(TABLEAU, "devgreen-gtfs-rt", "LAMP_DEVGREEN_RT_TripUpdates_LR_60_day.parquet"),
 )
 

src/lamp_py/tableau/hyper.py

@@ -95,14 +95,16 @@ def convert_parquet_dtype(self, dtype: pyarrow.DataType) -> SqlType:
         dtype = str(dtype)
         dtype_map = {
             "int8": SqlType.small_int(),
+            "uint8": SqlType.small_int(),
             "int16": SqlType.small_int(),
+            "uint16": SqlType.int(),
             "int32": SqlType.int(),
             "uint32": SqlType.big_int(),
             "int64": SqlType.big_int(),
             "bool": SqlType.bool(),
-            "float16": SqlType.double(),
-            "float32": SqlType.double(),
-            "float64": SqlType.double(),
+            "halffloat": SqlType.double(),
+            "float": SqlType.double(),
+            "double": SqlType.double(),
         }
 
         map_check = dtype_map.get(dtype)

src/lamp_py/tableau/jobs/filtered_hyper.py

@@ -79,6 +79,7 @@ def update_parquet(self, _: None) -> bool:
         if self.first_run:
             self.create_tableau_parquet(num_files=self.rollup_num_days)
             self.first_run = False
+            return True
 
         # only run once per day after 11AM UTC
         if object_exists(self.remote_input_location.s3_uri):
@@ -106,7 +107,15 @@ def create_tableau_parquet(self, num_files: Optional[int]) -> None:
         s3_uris = file_list_from_s3(
             bucket_name=self.remote_input_location.bucket, file_prefix=file_prefix, in_filter=self.object_filter
         )
+
         ds_paths = [s.replace("s3://", "") for s in s3_uris]
+
+        # s3 list returns in lexicographical order,
+        # so month=4/day=4 comes after month=4/day=30. This sort grabs the last part,
+        # e.g. 2025-04-22T00:00:00.parquet as the sort key, and re-orders by that instead
+        # to get it in date order
+        ds_paths = sorted(ds_paths, key=lambda x: os.path.split(x)[1])
+
         if num_files is not None:
             ds_paths = ds_paths[-num_files:]
 
@@ -115,8 +124,9 @@ def create_tableau_parquet(self, num_files: Optional[int]) -> None:
             format="parquet",
             filesystem=S3FileSystem(),
         )
-        process_logger = ProcessLogger("filtered_hyper_create", file_prefix=file_prefix)
+        process_logger = ProcessLogger("filtered_hyper_create", file_prefix=file_prefix, num_days=num_files)
         process_logger.log_start()
+        process_logger.add_metadata(first_file=ds_paths[0], last_file=ds_paths[-1])
         with pq.ParquetWriter(self.local_parquet_path, schema=self.processed_schema) as writer:
             for batch in ds.to_batches(
                 batch_size=500_000, columns=self.processed_schema.names, filter=self.parquet_filter