diff --git a/labs/GSP1298/HASHNODE.md b/labs/GSP1298/HASHNODE.md new file mode 100644 index 0000000..a965b01 --- /dev/null +++ b/labs/GSP1298/HASHNODE.md @@ -0,0 +1,264 @@ +# Stream data with pipelines - Esports Use case - GSP1298 + +## Overview + +In this Google Skills lab, a Python simulator publishes esports match events to Pub/Sub, Apache Beam processes them on Dataflow, and BigQuery stores raw events and player/team score updates. Two views select each competitor's latest score and rank the leaderboard. + +The official lab contains **five Tasks and four checkpoints** (30/20/30/20 points). The [original architecture diagrams](https://www.skills.google/focuses/132337?parent=catalog) are linked, not copied, because reuse/embedding rights are unverified. + +### What you'll do + +- Identify the lab project, region, and existing Cloud Storage bucket. +- Create a Pub/Sub topic/subscription and BigQuery dataset. +- Retrieve and configure two official Python files. +- Run a synthetic event producer and Dataflow streaming job. +- Create and query live player/team leaderboard views. + +## Setup and requirements + +### Before you click the Start Lab button + +Open [GSP1298 on Google Skills](https://www.skills.google/focuses/132337?parent=catalog). Start the timed lab only when ready, use a private browser, and sign in with the **temporary student account** from the lab panel. Do not use a personal project or publish student credentials. + +The lab provisions a Cloud Storage bucket and assigns a region. Use the provided values, not guessed defaults. A streaming Dataflow job can incur usage outside the timed lab. + +### Activate Cloud Shell + +Activate Cloud Shell, authorize the lab account, and inspect the project, region, and bucket: + +```bash +gcloud auth list +export PROJECT_ID="$(gcloud config get-value project)" +export REGION="$(gcloud config get-value compute/region)" +export BUCKET_NAME="${PROJECT_ID}-bucket" +printf 'Project=%s Region=%s Bucket=%s\n' "$PROJECT_ID" "$REGION" "$BUCKET_NAME" +gcloud storage buckets describe "gs://$BUCKET_NAME" +``` + +If the configured region is unset, use the region displayed in the lab panel. The bucket must already exist; do not create a substitute. + +## Task 1. Configure environment variables in Cloud Shell + +Keep `PROJECT_ID`, `REGION`, and `BUCKET_NAME` exported for all commands. New Cloud Shell tabs may require exporting them again. + +```bash +echo "$PROJECT_ID" +echo "$REGION" +echo "$BUCKET_NAME" +gcloud config get-value project +``` + +## Task 2. Create the cloud resources + +Create the exact Pub/Sub topic and subscription, and a BigQuery dataset in the `US` multiregion as specified by Google Skills: + +```bash +gcloud services enable pubsub.googleapis.com bigquery.googleapis.com dataflow.googleapis.com +gcloud pubsub topics create esports_events_topic +gcloud pubsub subscriptions create esports_events_topic-sub --topic=esports_events_topic +bq --location=US mk --dataset esports_analytics +``` + +Check the resources before repeating any creation command: + +```bash +gcloud pubsub topics describe esports_events_topic +gcloud pubsub subscriptions describe esports_events_topic-sub +bq show "$PROJECT_ID:esports_analytics" +``` + +### Test completed task + +Click **Check my progress** for **Create the cloud resources** (30 points). Confirm the subscription belongs to the expected topic. + +## Task 3. Retrieve the Python files and configure them + +Download the two official source files, which are provided as public Google Cloud Storage objects. The simulator generates match-start, elimination and match-end messages; the pipeline reads them from Pub/Sub and writes BigQuery score tables. + +```bash +mkdir -p "$HOME/esports" +cd "$HOME/esports" +curl -fL --retry 3 -o esports-simulation.py \ + https://storage.googleapis.com/spls/gsp1298/esports-simulation.py +curl -fL --retry 3 -o esports-pipeline.py \ + https://storage.googleapis.com/spls/gsp1298/esports-pipeline.py +``` + +Update the top-level `PROJECT_ID` assignments in both files and the `REGION` and `GCS_TEMP_LOCATION` assignments in the pipeline. This Python helper replaces only the expected assignments and fails if the upstream file structure differs: + +```bash +python3 - "$HOME/esports" "$PROJECT_ID" "$REGION" "$BUCKET_NAME" <<'PY' +from pathlib import Path +import re +import sys +folder, project, region, bucket = sys.argv[1:] +for name in ("esports-simulation.py", "esports-pipeline.py"): + path = Path(folder) / name + source = path.read_text() + fields = {"PROJECT_ID": project} + if name == "esports-pipeline.py": + fields.update({"REGION": region, "GCS_TEMP_LOCATION": f"gs://{bucket}/temp"}) + for key, value in fields.items(): + source, count = re.subn( + rf"^{key}\s*=\s*.*$", f"{key} = {value!r}", source, flags=re.MULTILINE + ) + if count != 1: + raise SystemExit(f"Expected one {key} assignment in {name}, found {count}") + path.write_text(source) +PY +python3 -m py_compile esports-simulation.py esports-pipeline.py +``` + +Upload the configured files into the lab bucket under the required `esports/` prefix: + +```bash +gcloud storage cp esports-simulation.py "gs://$BUCKET_NAME/esports/" +gcloud storage cp esports-pipeline.py "gs://$BUCKET_NAME/esports/" +gcloud storage ls "gs://$BUCKET_NAME/esports/" +``` + +### Test completed task + +Click **Check my progress** for **Retrieve the Python files and configure them** (20 points). These official Python files are runtime dependencies, not copied into the ePlus repository. + +## Task 4. Generate synthetic data and run the pipeline + +### Run the simulator + +Create a Python virtual environment, install the required Pub/Sub and Beam packages, and start the simulator. It runs continuously; keep the terminal open. + +```bash +cd "$HOME/esports" +python3 -m venv df-env +source df-env/bin/activate +python3 -m pip install google-cloud-pubsub 'apache-beam[gcp]' +python3 esports-simulation.py +``` + +In the Google Cloud console, inspect Pub/Sub → Subscriptions → `esports_events_topic-sub` for incoming messages. + +### Set up Dataflow permissions + +The official lab documents a potential Dataflow service-agent IAM error. First inspect the project and pre-provisioned worker identity: + +```bash +export PROJECT_NUMBER="$(gcloud projects describe "$PROJECT_ID" --format='value(projectNumber)')" +export WORKER_SA="${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" +export DATAFLOW_AGENT="service-${PROJECT_NUMBER}@dataflow-service-producer-prod.iam.gserviceaccount.com" +gcloud beta services identity create --service=dataflow.googleapis.com --project="$PROJECT_ID" +``` + +If permissions are missing, apply the scoped Dataflow roles in the authorized lab project: + +```bash +gcloud projects add-iam-policy-binding "$PROJECT_ID" \ + --member="serviceAccount:$DATAFLOW_AGENT" \ + --role=roles/dataflow.serviceAgent --condition=None +gcloud iam service-accounts add-iam-policy-binding "$WORKER_SA" \ + --member="serviceAccount:$DATAFLOW_AGENT" \ + --role=roles/iam.serviceAccountUser +gcloud projects add-iam-policy-binding "$PROJECT_ID" \ + --member="serviceAccount:$WORKER_SA" \ + --role=roles/dataflow.worker --condition=None +``` + +A Cloud Storage **soft-delete warning** is described in the official lab. The automation does not disable bucket protection automatically; a warning alone is not a failure. + +### Launch Dataflow + +Open a second Cloud Shell tab, export the project, region, and bucket values again, activate the virtual environment, and submit the streaming job: + +```bash +cd "$HOME/esports" +source df-env/bin/activate +python3 esports-pipeline.py \ + --project="$PROJECT_ID" --region="$REGION" \ + --runner=DataflowRunner --streaming \ + --temp_location="gs://$BUCKET_NAME/temp" \ + --job_name=esports-leaderboard-pipeline \ + --worker_machine_type=e2-standard-2 +``` + +Inspect Dataflow → Jobs. Wait for a job whose name starts with `esports-leaderboard` to become **Running**. Check worker logs and the BigQuery tables `raw_events`, `player_score_updates`, and `team_score_updates`. Time-window aggregation can delay the score tables by several minutes. + +### Test completed task + +Click **Check my progress** for **Generate synthetic data and run the pipeline** (30 points). Keep the streaming job active until grading is finished. + +## Task 5. Verify the results in BigQuery + +Create the player leaderboard view, selecting each player's latest score before ranking: + +```sql +CREATE OR REPLACE VIEW esports_analytics.player_leaderboard_live AS +SELECT RANK() OVER (ORDER BY total_score DESC) AS rank, + player_id, total_score, last_updated +FROM ( + SELECT player_id, total_score, last_updated, + ROW_NUMBER() OVER (PARTITION BY player_id ORDER BY last_updated DESC) AS rn + FROM esports_analytics.player_score_updates +) +WHERE rn = 1; +``` + +Create the team leaderboard view in the same way: + +```sql +CREATE OR REPLACE VIEW esports_analytics.team_leaderboard_live AS +SELECT RANK() OVER (ORDER BY total_wins DESC) AS rank, + team_id, total_wins, last_updated +FROM ( + SELECT team_id, total_wins, last_updated, + ROW_NUMBER() OVER (PARTITION BY team_id ORDER BY last_updated DESC) AS rn + FROM esports_analytics.team_score_updates +) +WHERE rn = 1; +``` + +Query both views with BigQuery Studio or the CLI: + +```bash +bq query --use_legacy_sql=false \ + 'SELECT * FROM esports_analytics.player_leaderboard_live ORDER BY rank LIMIT 10' +bq query --use_legacy_sql=false \ + 'SELECT * FROM esports_analytics.team_leaderboard_live ORDER BY rank LIMIT 10' +``` + +The leading player and team vary with the synthetic events. Do not hardcode the result. + +### Test completed task + +Click **Check my progress** for **Verify the results in BigQuery** (20 points). The checkpoints total 100 points; grader success is not asserted. + +--- + +## Solution of Lab + +### Quick + +Run the following two commands in the **authorized temporary lab Cloud Shell**. The URL deliberately targets `main`; it will only work **after the Draft PR is merged**. The script leaves streaming jobs running for grading. + +```bash +curl -LO https://raw.githubusercontent.com/ePlus-DEV/storage/refs/heads/main/labs/GSP1298/lab.sh +source lab.sh +``` + +The script discovers the project and region, uses the existing lab bucket, downloads and syntax-checks the official Python files, provisions the resources, launches the pipeline, and creates the views. It does not press Google Skills' grader buttons. + +### Manual + +Chưa tìm thấy video YouTube hướng dẫn thủ công đã xác minh cho lab này. + +Follow Tasks 1–5 above or the [official lab](https://www.skills.google/focuses/132337?parent=catalog). + +## Sources and verification status + +- [Google Skills GSP1298 official lab](https://www.skills.google/focuses/132337?parent=catalog), last tested October 15, 2026. +- [Official simulator source](https://storage.googleapis.com/spls/gsp1298/esports-simulation.py). +- [Official Dataflow pipeline source](https://storage.googleapis.com/spls/gsp1298/esports-pipeline.py). +- [Dataflow security and permissions](https://cloud.google.com/dataflow/docs/concepts/security-and-permissions). +- [BigQuery views documentation](https://cloud.google.com/bigquery/docs/views-intro). + +**static validated only; runtime/grader unverified.** The official Python source URLs were identified but not executed in an authorized lab. Python syntax is checked at runtime, while Bash/ShellCheck require latest-commit GitHub CI. IAM propagation, Dataflow worker readiness and streaming table creation vary by session. No private credentials are embedded. This Markdown is committed for review, not published to Hashnode. + +Copyright (c) 2026 ePlus.DEV | Unlicense — original automation and explanatory prose. diff --git a/labs/GSP1298/lab.sh b/labs/GSP1298/lab.sh new file mode 100644 index 0000000..bc5a07c --- /dev/null +++ b/labs/GSP1298/lab.sh @@ -0,0 +1,179 @@ +#!/usr/bin/env bash +# GSP1298 — Stream data with pipelines - Esports Use case +# Copyright (c) 2026 ePlus.DEV | Unlicense +# static validated only; runtime/grader unverified +set -Eeuo pipefail +umask 077 +if [[ -n "${NO_COLOR+x}" ]]; then C='' G='' Y='' R=''; +elif [[ -t 1 || "${FORCE_COLOR:-0}" == 1 ]]; then + C=$'\033[36m'; G=$'\033[32m'; Y=$'\033[33m'; R=$'\033[0m' +else C='' G='' Y='' R=''; fi +info(){ printf '%s[INFO]%s %s\n' "$C" "$R" "$*"; } +ok(){ printf '%s[OK]%s %s\n' "$G" "$R" "$*"; } +warn(){ printf '%s[WARN]%s %s\n' "$Y" "$R" "$*" >&2; } +die(){ printf '[ERROR] %s\n' "$*" >&2; exit 1; } +trap 'warn "Stopped near line $LINENO; inspect errors. No grader PASS is implied."' ERR +printf '%sePlus.DEV – Starting Execution%s\n' "$C" "$R" +printf 'Copyright (c) 2026 ePlus.DEV | Unlicense\n' +for cmd in gcloud bq curl python3 jq; do command -v "$cmd" >/dev/null || die "Missing $cmd"; done +PROJECT_ID="${PROJECT_ID:-$(gcloud config get-value project 2>/dev/null)}" +[[ -n "$PROJECT_ID" && "$PROJECT_ID" != '(unset)' ]] || die 'Choose the temporary lab project.' +REGION="${REGION:-$(gcloud config get-value compute/region 2>/dev/null)}" +if [[ -z "$REGION" || "$REGION" == '(unset)' ]]; then + ZONE="$(gcloud config get-value compute/zone 2>/dev/null)" + [[ -n "$ZONE" && "$ZONE" != '(unset)' ]] || die 'Set REGION to the lab-assigned region.' + REGION="${ZONE%-*}" +fi +[[ "$REGION" =~ ^[a-z]+-[a-z0-9]+[0-9]$ ]] || die 'Invalid region.' +BUCKET_NAME="${BUCKET_NAME:-${PROJECT_ID}-bucket}" +BUCKET_NAME="${BUCKET_NAME#gs://}"; BUCKET_NAME="${BUCKET_NAME%/}" +gcloud storage buckets describe "gs://$BUCKET_NAME" --project="$PROJECT_ID" >/dev/null || + die 'Expected pre-provisioned lab bucket is missing. Set BUCKET_NAME from the lab panel.' +WORKDIR="${WORKDIR:-$HOME/esports}" +mkdir -p "$WORKDIR" +info "Project=$PROJECT_ID Region=$REGION Bucket=gs://$BUCKET_NAME" +retry(){ local count="$1" delay="$2" n; shift 2 + for ((n=1;n<=count;n++)); do + if "$@"; then return 0; fi + if ((n/dev/null +info 'Task 2: create Pub/Sub and BigQuery resources without duplicates' +gcloud services enable pubsub.googleapis.com bigquery.googleapis.com dataflow.googleapis.com --project="$PROJECT_ID" --quiet +gcloud pubsub topics describe esports_events_topic --project="$PROJECT_ID" >/dev/null 2>&1 || + gcloud pubsub topics create esports_events_topic --project="$PROJECT_ID" +gcloud pubsub subscriptions describe esports_events_topic-sub --project="$PROJECT_ID" >/dev/null 2>&1 || + gcloud pubsub subscriptions create esports_events_topic-sub --project="$PROJECT_ID" --topic=esports_events_topic +bq show --project_id="$PROJECT_ID" --format=none "$PROJECT_ID:esports_analytics" >/dev/null 2>&1 || + bq --project_id="$PROJECT_ID" --location=US mk --dataset esports_analytics +gcloud pubsub topics describe esports_events_topic --project="$PROJECT_ID" >/dev/null +gcloud pubsub subscriptions describe esports_events_topic-sub --project="$PROJECT_ID" >/dev/null +bq show --project_id="$PROJECT_ID" "$PROJECT_ID:esports_analytics" >/dev/null +ok 'Checkpoint: Create the cloud resources (30 points)' + +info 'Task 3: fetch official Python source and configure all dynamic values' +for name in esports-simulation.py esports-pipeline.py; do + if [[ ! -e "$WORKDIR/$name" ]]; then + retry 3 5 curl -fL --retry 3 "https://storage.googleapis.com/spls/gsp1298/$name" \ + -o "$WORKDIR/$name" || die "Unable to retrieve official $name" + fi + [[ -s "$WORKDIR/$name" ]] || die "Empty $name" +done +python3 - "$WORKDIR" "$PROJECT_ID" "$REGION" "$BUCKET_NAME" <<'PY' +from pathlib import Path +import re +import sys +directory, project, region, bucket = sys.argv[1:] +for name in ("esports-simulation.py", "esports-pipeline.py"): + path = Path(directory) / name + text = path.read_text(encoding="utf-8") + fields = {"PROJECT_ID": project} + if name == "esports-pipeline.py": + fields.update({"REGION": region, "GCS_TEMP_LOCATION": f"gs://{bucket}/temp"}) + for key, value in fields.items(): + pattern = rf"^{re.escape(key)}\s*=\s*.*$" + text, count = re.subn(pattern, f"{key} = {value!r}", text, flags=re.MULTILINE) + if count != 1: + raise SystemExit(f"Expected exactly one {key} assignment in {name}, got {count}") + path.write_text(text, encoding="utf-8") +PY +python3 -m py_compile "$WORKDIR/esports-simulation.py" "$WORKDIR/esports-pipeline.py" +gcloud storage cp "$WORKDIR/esports-simulation.py" "gs://$BUCKET_NAME/esports/esports-simulation.py" --quiet +gcloud storage cp "$WORKDIR/esports-pipeline.py" "gs://$BUCKET_NAME/esports/esports-pipeline.py" --quiet +gcloud storage objects describe "gs://$BUCKET_NAME/esports/esports-simulation.py" >/dev/null +gcloud storage objects describe "gs://$BUCKET_NAME/esports/esports-pipeline.py" >/dev/null +ok 'Checkpoint: Retrieve the Python files and configure them (20 points)' + +info 'Task 4: install isolated dependencies and launch bounded background processes' +if [[ ! -x "$WORKDIR/df-env/bin/python" ]]; then python3 -m venv "$WORKDIR/df-env"; fi +retry 3 8 "$WORKDIR/df-env/bin/python" -m pip install --disable-pip-version-check \ + google-cloud-pubsub 'apache-beam[gcp]' || + die 'Unable to install required Python dependencies.' +if [[ -s "$WORKDIR/simulator.pid" ]] && + kill -0 "$(cat "$WORKDIR/simulator.pid")" 2>/dev/null; then + info 'Existing simulator process is running.' +else + nohup "$WORKDIR/df-env/bin/python" "$WORKDIR/esports-simulation.py" \ + > "$WORKDIR/simulator.log" 2>&1 < /dev/null & + echo "$!" > "$WORKDIR/simulator.pid" +fi +sleep 10 +kill -0 "$(cat "$WORKDIR/simulator.pid")" 2>/dev/null || + die 'Simulator exited unexpectedly; inspect simulator.log.' + +PROJECT_NUMBER="$(gcloud projects describe "$PROJECT_ID" --format='value(projectNumber)')" +[[ "$PROJECT_NUMBER" =~ ^[0-9]+$ ]] || die 'Could not discover project number.' +WORKER_SA="${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" +DATAFLOW_AGENT="service-${PROJECT_NUMBER}@dataflow-service-producer-prod.iam.gserviceaccount.com" +gcloud beta services identity create --service=dataflow.googleapis.com --project="$PROJECT_ID" >/dev/null +# Grant the Dataflow service identity only the documented agent role. +gcloud projects add-iam-policy-binding "$PROJECT_ID" \ + --member="serviceAccount:$DATAFLOW_AGENT" \ + --role=roles/dataflow.serviceAgent --condition=None --quiet >/dev/null +# Let the Dataflow service agent impersonate the pre-provisioned worker identity. +gcloud iam service-accounts add-iam-policy-binding "$WORKER_SA" \ + --project="$PROJECT_ID" --member="serviceAccount:$DATAFLOW_AGENT" \ + --role=roles/iam.serviceAccountUser --quiet >/dev/null +gcloud projects add-iam-policy-binding "$PROJECT_ID" \ + --member="serviceAccount:$WORKER_SA" \ + --role=roles/dataflow.worker --condition=None --quiet >/dev/null + +if [[ -s "$WORKDIR/pipeline.pid" ]] && + kill -0 "$(cat "$WORKDIR/pipeline.pid")" 2>/dev/null; then + info 'Existing Dataflow launcher is running.' +elif gcloud dataflow jobs list --project="$PROJECT_ID" --region="$REGION" --format=json | + jq -e '.[] | select((.name | startswith("esports-leaderboard")) and + (.state == "Running" or .state == "JOB_STATE_RUNNING"))' >/dev/null; then + info 'Existing esports Dataflow job is running.' +else + nohup "$WORKDIR/df-env/bin/python" "$WORKDIR/esports-pipeline.py" \ + --project="$PROJECT_ID" --region="$REGION" --runner=DataflowRunner --streaming \ + --temp_location="gs://$BUCKET_NAME/temp" --job_name=esports-leaderboard-pipeline \ + --worker_machine_type=e2-standard-2 > "$WORKDIR/pipeline.log" 2>&1 < /dev/null & + echo "$!" > "$WORKDIR/pipeline.pid" +fi +running=0 +for ((i=1;i<=24;i++)); do + if gcloud dataflow jobs list --project="$PROJECT_ID" --region="$REGION" --format=json | + jq -e '.[] | select((.name | startswith("esports-leaderboard")) and + (.state == "Running" or .state == "JOB_STATE_RUNNING"))' >/dev/null; then + running=1; break + fi + sleep 15 +done +(( running == 1 )) || die 'Dataflow job did not reach RUNNING in 6 minutes. Inspect pipeline.log.' +ok 'Checkpoint: Generate synthetic data and run the pipeline (30 points)' + +info 'Task 5: wait for streaming tables, then create the two official leaderboard views' +ready=0 +for ((i=1;i<=24;i++)); do + if bq show "$PROJECT_ID:esports_analytics.player_score_updates" >/dev/null 2>&1 && + bq show "$PROJECT_ID:esports_analytics.team_score_updates" >/dev/null 2>&1; then + ready=1; break + fi + sleep 15 +done +(( ready == 1 )) || die 'Expected streaming score tables not ready; inspect Dataflow logs.' +bq query --project_id="$PROJECT_ID" --use_legacy_sql=false --format=none \ +"CREATE OR REPLACE VIEW `$PROJECT_ID.esports_analytics.player_leaderboard_live` AS + SELECT RANK() OVER (ORDER BY total_score DESC) AS rank, player_id, total_score, last_updated + FROM (SELECT player_id, total_score, last_updated, + ROW_NUMBER() OVER (PARTITION BY player_id ORDER BY last_updated DESC) AS rn + FROM `$PROJECT_ID.esports_analytics.player_score_updates`) WHERE rn=1;" +bq query --project_id="$PROJECT_ID" --use_legacy_sql=false --format=none \ +"CREATE OR REPLACE VIEW `$PROJECT_ID.esports_analytics.team_leaderboard_live` AS + SELECT RANK() OVER (ORDER BY total_wins DESC) AS rank, team_id, total_wins, last_updated + FROM (SELECT team_id, total_wins, last_updated, + ROW_NUMBER() OVER (PARTITION BY team_id ORDER BY last_updated DESC) AS rn + FROM `$PROJECT_ID.esports_analytics.team_score_updates`) WHERE rn=1;" +bq show "$PROJECT_ID:esports_analytics.player_leaderboard_live" >/dev/null +bq show "$PROJECT_ID:esports_analytics.team_leaderboard_live" >/dev/null +bq query --project_id="$PROJECT_ID" --use_legacy_sql=false \ + "SELECT * FROM `$PROJECT_ID.esports_analytics.player_leaderboard_live` ORDER BY rank LIMIT 10" +bq query --project_id="$PROJECT_ID" --use_legacy_sql=false \ + "SELECT * FROM `$PROJECT_ID.esports_analytics.team_leaderboard_live` ORDER BY rank LIMIT 10" +ok 'Checkpoint: Verify the results in BigQuery (20 points)' +info 'Streaming jobs remain running for the timed lab. Do not cancel before grader review.' +info 'static validated only; runtime/grader unverified'