Append observations from many CSVs to a single feature collection

[anayeaye]

What

This issue is to coordinate a new usage pattern for the vector ingest DAG that will aggregate observations from thousands of files into a single collection. There are caveats and if we choose to use this pattern more often in the future we will likely want to add some guardrails to the DAG.

Gemini notes from planning session with @siddharth0248 @sandrahoang686 @anayeaye

CSV Data

Air4US needs to load the observations from thousands of sensors/stations into the features API. These observations are provided in one CSV per station with rows containing a timestamp, lat lng, and the values of various sensor measurements that were recorded (PM25, PM10, O3, NO2, SO2, CO--not all sensors provide values at each station so these columns can be null). An additional CSV provides an overview of all stations and their lat/lng that can be used to spatially join observations to the relative station (but do not have a foreign key for the observation records).

Airflow

The default vector ingest DAG has been most frequently used to load all records from a file into a single table ('collection' in the features api) and name the collection with the filename. Some of the ogr2ogr args are hard coded and cannot be modified with DAG config (for example the layer name will always be set to the collection string from the payload). We can use the extra_args to customize the ogr2ogr transformation for the CSVs we are ingesting and we can configure an append instead of an overwrite and replace transaction.

veda-data-airflow/dags/veda_data_pipeline/utils/vector_ingest/handler.py

Lines 245 to 258 in fabb8f5

	print(f"running ogr2ogr import for collection: {collection}")
	options = [
	"ogr2ogr",
	"-f",
	"PostgreSQL",
	connection,
	filename,
	"-nln",
	collection,
	"-s_srs",
	source_projection,
	"-t_srs",
	target_projection,
	*extra_flags

veda-data-airflow/dags/veda_data_pipeline/utils/vector_ingest/handler.py

Lines 377 to 378 in fabb8f5

	if collection_not_provided:
	collection = payload_event.get("id_template", "{}").format(filename)

Ogr2ogr

Usage
https://gdal.org/en/stable/programs/ogr2ogr.html

• -append allows us to update an existing layer rather than overwrite/recreate. Upsert is avaialble but the CSV driver does not support it.

Note

When configuring a new ingest, using a throwaway table that will be manually deleted after testing with the delete DAG, the -debug, "on" argument can be passed in the extra args for the DAG config. This will cause an exception that will be printed in the Mapped Task logs ingest_vector and it can be very handy. That exception also fails the DAG run--the records may be loaded into a new table, however, and the manual delete is still necessary.

CSV driver options
https://gdal.org/en/stable/drivers/vector/csv.html#open-options

• From the open options we probably want to choose "AUTODETECT_TYPE=YES" to avoid loading all values as strings.
• We may also want "EMPTY_STRING_AS_NULL=YES" because not all sensor types have values in each observation record.

Example

This DAG run config will load three files into a collection named ak_aq_sensordata2. It will also reload those three files every time it is rerun, appending new records on each run.
Currently the output can be seen here (but this collection will not be persisted long term) https://eic-staging.staging.earth.gov/api/features/collections/public.ak_aq_sensordata2/items

{
  "bucket": "veda-data-store-staging",
  "prefix": "ak-test/",
  "vector": true,
  "collection": "ak_aq_sensordata2",
  "extra_flags": [
    "-append",
    "-oo",
    "X_POSSIBLE_NAMES=longitude",
    "-oo",
    "Y_POSSIBLE_NAMES=latitude",
    "-oo",
    "AUTODETECT_TYPE=YES",
    "-oo",
    "EMPTY_STRING_AS_NULL=YES"
  ],
  "filename_regex": ".*01_\\d{3}_\\d{4}\\.csv$",
  "source_projection": "EPSG:4326",
  "target_projection'": "EPSG:4326",
  "invalidate_cloudfront": true
}

Caveats

• The observations in the CSV files do not contain persistent FIDs so we cannot implement an upsert process that skips existing records with a shared FID. If the same CSV is loaded twice we will have two records of those observations in the table. Location and time are likely unique for deduplicating in downstream applications.

[anayeaye]

I forgot to mention using _ instead of -. TODO: update notes to make it clear that we should always use underscores! Also perhaps we can add some guardrails in airflow--we already have a related issue here #505