Revert "Revert "MVP of just producing JSON from RT protos""

Author: atvaccaro

.gitignore

@@ -1,6 +1,10 @@
 # Vim
 .*.sw[po]
 
+# PyCharm, etc.
+.idea/
+.DS_Store
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

.pre-commit-config.yaml

@@ -3,6 +3,7 @@ repos:
     rev: v2.4.0
     hooks:
       - id: flake8
+        args: ["--ignore=E501,W503"] # line too long and line before binary operator (black is ok with these)
         types:
           - python
       - id: trailing-whitespace

airflow/README.md

@@ -17,6 +17,7 @@ This project is developed using docker and docker-compose. Before getting starte
 First, if you're on linux, you'll need to make sure that the UID and GID of the container match, to do so, run
 
 ```console
+cd airflow (if you are not already in the airflow directory)
 mkdir ./dags ./logs ./plugins
 echo -e "AIRFLOW_UID=$(id -u)\nAIRFLOW_GID=0" > .env
 ```
@@ -37,7 +38,7 @@ To run the gtfs_downloader dags, it is required to generate a yml file of the ag
 secrets filled in. To run this, do the following:
 
 1. Setup a virtual environment within the `script` folder by running `python -m venv .venv` in the `script` folder.
-2. Install the needed requirements via `pip install -r requirements.txt`
+2. Install the needed requirements via `pip install -r requirements.txt` (the requirements in `script`)
 3. Copy `airflow/data/example-secrets.csv` to `airflow/data/secrets.csv` and fill in the secret keys as needed
 4. run `python yml_convert.py ../airflow/data/agencies.yml ../airflow/data/agencies.filled.yml ../airflow/data/secrets.csv`
 5. Copy `/airflow/data/agencies.yml` to `/airflow/data/agencies_raw.yml`

airflow/dags/rt_loader/external_service_alerts.sql

@@ -0,0 +1,12 @@
+---
+operator: operators.SqlQueryOperator
+
+dependencies:
+  - parse_rt_service_alerts
+---
+
+CREATE OR REPLACE EXTERNAL TABLE `gtfs_rt.service_alerts`
+OPTIONS (
+    format = "JSON",
+    uris = ["{{get_bucket()}}/rt-processed/service_alerts/*.jsonl.gz"]
+)

airflow/dags/rt_loader/external_trip_updates.sql

@@ -0,0 +1,12 @@
+---
+operator: operators.SqlQueryOperator
+
+dependencies:
+  - parse_rt_trip_updates
+---
+
+CREATE OR REPLACE EXTERNAL TABLE `gtfs_rt.trip_updates`
+OPTIONS (
+    format = "JSON",
+    uris = ["{{get_bucket()}}/rt-processed/trip_updates/*.jsonl.gz"]
+)

airflow/dags/rt_loader/external_vehicle_positions.sql

@@ -1,9 +1,12 @@
 ---
 operator: operators.SqlQueryOperator
+
+dependencies:
+  - parse_rt_vehicle_positions
 ---
 
 CREATE OR REPLACE EXTERNAL TABLE `gtfs_rt.vehicle_positions`
 OPTIONS (
-    format = "PARQUET",
-    uris = ["gs://gtfs-data/rt-processed/vehicle_positions/*.parquet"]
+    format = "JSON",
+    uris = ["{{get_bucket()}}/rt-processed/vehicle_positions/*.jsonl.gz"]
 )

airflow/dags/rt_loader/parse_rt_service_alerts.yml

@@ -0,0 +1,5 @@
+operator: operators.RealtimeToFlattenedJSONOperator
+provide_context: true
+
+filename_prefix: al
+rt_file_substring: service_alerts

airflow/dags/rt_loader/parse_rt_trip_updates.yml

@@ -0,0 +1,5 @@
+operator: operators.RealtimeToFlattenedJSONOperator
+provide_context: true
+
+filename_prefix: tu
+rt_file_substring: trip_updates

airflow/dags/rt_loader/parse_rt_vehicle_positions.py

@@ -0,0 +1,5 @@
+operator: operators.RealtimeToFlattenedJSONOperator
+provide_context: true
+
+filename_prefix: rt
+rt_file_substring: vehicle_positions

airflow/dags/rt_loader/parse_rt_vehicle_positions.yml

@@ -0,0 +1,7 @@
+operator: operators.RealtimeToFlattenedJSONOperator
+provide_context: true
+
+filename_prefix: tu
+rt_file_substring: trip_updates
+subfolder: rt-processed-sandbox
+limit: 5

airflow/dags/sandbox/realtime_to_flattened_json.yml

@@ -8,3 +8,6 @@
 from operators.python_to_warehouse_operator import PythonToWarehouseOperator
 from operators.sql_query_operator import SqlQueryOperator
 from operators.airtable_to_warehouse import AirtableToWarehouseOperator
+from operators.realtime_to_flattened_json_operator import (
+    RealtimeToFlattenedJSONOperator,
+)