Skip to content

include flood risk zone in the new collections for testing #70

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
eveleighoj wants to merge 9 commits into
base: main
Choose a base branch
from
Open

include flood risk zone in the new collections for testing #70

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
6a29e31
include flood risk zone in the new collections for testing
eveleighoj Apr 14, 2026
dbe0a03
update missing params that have been added
eveleighoj Apr 14, 2026
2a1e20f
edit EMR serverless job submission to use archives fo python dependen…
eveleighoj Apr 18, 2026
fd33ab7
include python environment in the the driver option as well
eveleighoj Apr 19, 2026
e172b18
remove driver configg for batch jobs
eveleighoj Apr 19, 2026
a62f874
corect how archives is being passed inn
eveleighoj Apr 19, 2026
0195dd0
remove conf and rely on custom image instead
eveleighoj Apr 19, 2026
2ded468
correct variabe name for package ecs task
eveleighoj May 14, 2026
30c11f1
revert name change as it was a scrript error
eveleighoj May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions dags/new_collection_generator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@
collections = get_collections_dict(datasets_dict.values())


filtered_collections = {k: v for k, v in collections.items() if k in ["central-activities-zone", "transport-access-node", "title-boundary"]}
filtered_collections = {k: v for k, v in collections.items() if k in ["central-activities-zone", "transport-access-node", "title-boundary", "flood-risk-zone"]}

# read config from file and environment
config = get_config()
Expand Down Expand Up @@ -69,6 +69,7 @@
"transform-batch-size": Param(default=200, type="integer"),
"incremental-loading-override": Param(default=False, type="boolean"),
"regenerate-log-override": Param(default=False, type="boolean"),
"force-reprocessing": Param(default=False, type="boolean"),
},
render_template_as_native_obj=True,
is_paused_upon_creation=False,
Expand All @@ -93,6 +94,7 @@ def configure_dag(**kwargs):
transform_batch_size = int(kwargs["params"].get("transform-batch-size"))
incremental_loading_override = bool(kwargs["params"].get("incremental-loading-override"))
regenerate_log_override = bool(kwargs["params"].get("regenerate-log-override"))
force_reprocessing = bool(kwargs["params"].get("force-reprocessing"))

# Push values to XCom
ti.xcom_push(key="memory", value=memory)
Expand All @@ -102,6 +104,7 @@ def configure_dag(**kwargs):
ti.xcom_push(key="transform-batch-size", value=transform_batch_size)
ti.xcom_push(key="incremental-loading-override", value=incremental_loading_override)
ti.xcom_push(key="regenerate-log-override", value=regenerate_log_override)
ti.xcom_push(key="force-reprocessing", value="True" if force_reprocessing else "")

# add collection_data bucket # add collection bucket name
collection_dataset_bucket_name = kwargs["conf"].get(section="custom", key="collection_dataset_bucket_name")
Expand Down Expand Up @@ -160,6 +163,7 @@ def configure_dag(**kwargs):
"value": '\'{{ task_instance.xcom_pull(task_ids="configure-dag", key="incremental-loading-override") | string }}\'',
},
{"name": "REGENERATE_LOG_OVERRIDE", "value": '\'{{ task_instance.xcom_pull(task_ids="configure-dag", key="regenerate-log-override") | string }}\''},
{"name": "REPROCESS", "value": '\'{{ task_instance.xcom_pull(task_ids="configure-dag", key="force-reprocessing") }}\''},
],
},
]
Expand Down Expand Up @@ -228,6 +232,7 @@ def get_emr_application_id(**context):
S3_LOG_URI = f"s3://{S3_LOG_BUCKET}/"
S3_DEPENDENCIES_PATH = f"s3://{S3_BUCKET}/pkg/dependencies/dependencies.zip"
S3_DATA_PATH = f"s3://{S3_SOURCE_DATA_PATH}/"
S3_ENVIRONMENT_PATH = f"s3://{S3_BUCKET}/pkg/dependencies/environment.tar.gz"

assemble_emr_task = EmrServerlessStartJobOperator(
task_id="assemble-emr-job",
Expand All @@ -248,7 +253,8 @@ def get_emr_application_id(**context):
"--parquet-datasets-path",
f"s3://{ENV}-parquet-datasets",
],
"sparkSubmitParameters": f"--jars /usr/lib/spark/jars/postgresql-42.7.4.jar --py-files {S3_WHEEL_FILE},{S3_DEPENDENCIES_PATH} "
"sparkSubmitParameters": f"--jars /usr/lib/spark/jars/postgresql-42.7.4.jar "
f"--py-files {S3_WHEEL_FILE} "
"--conf spark.serializer=org.apache.spark.serializer.KryoSerializer "
"--conf spark.kryo.registrator=org.apache.sedona.core.serde.SedonaKryoRegistrator "
"--conf spark.sql.extensions=org.apache.sedona.sql.SedonaSqlExtensions",
Expand Down
Loading