Add import aggegation helper #1829
Conversation
Summary of ChangesHello @vish-cs, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a new BigQuery aggregation helper Cloud Function and seamlessly integrates it into the existing data ingestion workflow. The primary goal is to enhance the data pipeline by adding a dedicated, automated step for performing post-ingestion data aggregation or validation within BigQuery for various imports. This allows for more flexible and specific data processing after the initial dataflow jobs are completed. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new Cloud Function, aggregation-helper, designed to run post-import aggregation queries in BigQuery. The function is integrated into the existing spanner-ingestion-workflow. The changes also include updating the Cloud Build configuration to deploy this new function and modifying the workflow YAML to invoke it.
My review has identified a critical issue in the new Cloud Function related to SQL syntax that will cause it to fail. I've also included several suggestions to improve code quality, maintainability, and configuration hygiene.
| except Exception as e: | ||
| logging.warning(f"Failed to initialize BigQuery client: {e}") |
There was a problem hiding this comment.
Catching a broad Exception can hide bugs and make debugging difficult. It's better to catch more specific exceptions. For BigQuery client initialization, consider catching google.auth.exceptions.DefaultCredentialsError or other specific exceptions from the google-cloud-bigquery library if you know what might go wrong.
| logging.warning(f"Failed to initialize BigQuery client: {e}") | ||
| bq_client = None | ||
|
|
||
| BQ_DATASET_ID = os.environ.get('BQ_DATASET_ID') |
There was a problem hiding this comment.
The environment variable BQ_DATASET_ID is loaded but it's never used in the code. This can be confusing for future developers. Please either use it in your queries (e.g., to specify the dataset for tables) or remove it if it's not needed.
| if "india_census" in import_name: | ||
| # Placeholder for India Census specific logic | ||
| query = """ | ||
| SELECT @import_name as import_name CURRENT_TIMESTAMP() as execution_time | ||
| """ | ||
| elif "us_census" in import_name: | ||
| # Placeholder for US Census specific logic | ||
| query = """ | ||
| SELECT @import_name as import_name CURRENT_TIMESTAMP() as execution_time | ||
| """ | ||
| else: | ||
| logging.info( | ||
| f"No specific aggregation logic for import: {import_name}") | ||
| continue |
There was a problem hiding this comment.
The if/elif chain to select a query based on import_name can become hard to maintain as more import types are added. Consider refactoring this to use a dictionary to map import name patterns to query logic. This would make the code more scalable and easier to read.
For example:
QUERY_MAP = {
"india_census": """
SELECT @import_name as import_name, CURRENT_TIMESTAMP() as execution_time
""",
"us_census": """
SELECT @import_name as import_name, CURRENT_TIMESTAMP() as execution_time
""",
}
query = None
for key, q in QUERY_MAP.items():
if key in import_name:
query = q
break
if not query:
logging.info(f"No specific aggregation logic for import: {import_name}")
continue| @@ -0,0 +1,2 @@ | |||
| functions-framework==3.* | |||
| google-cloud-bigquery | |||
There was a problem hiding this comment.
It's a good practice to pin dependency versions in requirements.txt to ensure reproducible builds and prevent unexpected breakages from library updates. Please consider specifying a version for google-cloud-bigquery.
google-cloud-bigquery==3.17.2
|
|
||
| - id: 'import-aggregation-helper' | ||
| name: 'gcr.io/cloud-builders/gcloud' | ||
| args: ['functions', 'deploy', 'import-aggregation-helper', '--runtime', 'python312', '--source', 'aggregation-helper', '--no-allow-unauthenticated', '--trigger-http', '--entry-point', 'aggregation_helper', '--project', '${_PROJECT_ID}', '--set-env-vars', 'PROJECT_ID=${_PROJECT_ID},SPANNER_PROJECT_ID=${_SPANNER_PROJECT_ID},SPANNER_INSTANCE_ID=${_SPANNER_INSTANCE_ID},SPANNER_DATABASE_ID=${_SPANNER_DATABASE_ID},GCS_BUCKET_ID=${_GCS_BUCKET_ID},LOCATION=${_LOCATION},BQ_DATASET_ID=${_BQ_DATASET_ID}'] |
There was a problem hiding this comment.
The import-aggregation-helper function is being configured with several environment variables that it does not appear to use, such as SPANNER_PROJECT_ID, SPANNER_INSTANCE_ID, SPANNER_DATABASE_ID, GCS_BUCKET_ID, and LOCATION. To improve security and reduce configuration complexity, it's best practice to only provide the environment variables that are actually required by the function. The aggregation-helper function only seems to require BQ_DATASET_ID.
args: ['functions', 'deploy', 'import-aggregation-helper', '--runtime', 'python312', '--source', 'aggregation-helper', '--no-allow-unauthenticated', '--trigger-http', '--entry-point', 'aggregation_helper', '--project', '${_PROJECT_ID}', '--set-env-vars', 'BQ_DATASET_ID=${_BQ_DATASET_ID}']
1 participant
No description provided.