cshuo commented on code in PR #20243: URL: https://github.com/apache/hudi/pull/20243#discussion_r4348227896
########## hudi-lakehouse/charts/hudi-spark-connect/README.md: ########## @@ -0,0 +1,171 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +# Hudi Spark Connect endpoint + +Deploys Spark Connect 4.1.3 with Hudi 1.2.0, Scala 2.13, and Java 17. +Tables use an external Hive Metastore and S3-compatible storage. +See the [Dockerfile](../../images/spark-connect/Dockerfile) for dependency versions. + +## Local-dev quickstart + +Requires Docker, minikube, Helm, and kubectl. Run from the repository root: + +```bash +# Start MinIO, Hive Metastore, and Spark Connect +./hudi-lakehouse/scripts/quickstart.sh --spark-connect + +# Also write example data and query it through Spark Connect (requires Maven and a JDK) +./hudi-lakehouse/scripts/quickstart.sh --run-example --spark-connect +``` + +The example uses the existing Spark writer to populate and register `default.trips`, +then prints city counts: `chennai = 33`, `san_francisco = 34`, `sao_paulo = 33`. +The query runs through PySpark inside the endpoint pod; no host client is needed. +Omit `--spark-connect` to query through Trino instead. + +To rebuild the image and reload the running endpoint: + +```bash +./hudi-lakehouse/scripts/quickstart.sh --spark-connect --rebuild-images +kubectl -n hudi-lakehouse rollout restart deployment/hudi-spark-connect +kubectl -n hudi-lakehouse rollout status deployment/hudi-spark-connect +``` + +## Install against existing infrastructure + +Build an image and supply your metastore, warehouse, and storage credentials: + +```bash +./hudi-lakehouse/scripts/build-connect-image.sh --registry my.registry/team --push +helm upgrade --install hudi-spark-connect hudi-lakehouse/charts/hudi-spark-connect \ + --namespace hudi-lakehouse --create-namespace \ + --set image.repository=my.registry/team/hudi-lakehouse-spark-connect \ + --set catalog.metastoreUri=thrift://my-hms:9083 \ + --set catalog.warehouse=s3a://my-bucket/connect \ + --set storage.s3.existingSecret=my-s3-credentials \ + --wait --timeout 10m +``` + +The Secret must be in the same namespace and contain `accessKey` and `secretKey`. +For temporary credentials, also set `storage.s3.sessionTokenKey`. +Without a Secret, Hadoop uses its default credential provider chain. +Local-dev credentials are for development only. + +Other settings in [values.yaml](values.yaml): + +| Setting | Purpose | +|---|---| +| `storage.s3.endpoint`, `region`, `pathStyleAccess` | MinIO or custom S3 endpoints | +| `spark.cores` | Local worker threads | +| `spark.driverMemory` | JVM heap; keep the pod memory limit higher | +| `resources` | Pod CPU and memory requests/limits | +| `spark.extraConf` | Additional Spark properties; no credentials or overrides of chart-owned settings | + +The endpoint runs one shared driver in local mode, without separate executors. +Keep one replica. Restarting it ends active sessions; tables remain in the Review Comment: Documented service ownership and writer/lock configuration. Inline compaction and clustering are off by default in Hudi 1.2.0; automatic cleaning is enabled. The endpoint adds no scheduler or cross-session writer coordination. External service scheduling remains operator-owned; timeline-server reuse is not enabled. ########## hudi-lakehouse/local-dev/example/connect_smoke.py: ########## @@ -0,0 +1,97 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""ADBC-only CRUD smoke test; each invocation creates a new Connect session. + +Use prepare -> restart the endpoint -> verify -> cleanup to check that both +catalog metadata and data survive. Only a uniquely named test database is used. +""" +import argparse +import re +from pathlib import Path + +from adbc_driver_manager import dbapi + + +def identifier(value): + # This CLI interpolates only its own test identifiers, never arbitrary SQL. + if not re.fullmatch(r"connect_smoke_[a-z0-9_]+", value): + raise argparse.ArgumentTypeError("database must match connect_smoke_[a-z0-9_]+") + return value + + +def run(cursor, sql, expected=None): + print(sql, flush=True) + cursor.execute(sql) + # DDL/DML do not necessarily have result sets; never fetch unconditionally. + if expected is not None: + actual = [tuple(row) for row in cursor.fetchall()] + if actual != expected: + raise AssertionError(f"Expected {expected!r}, got {actual!r}") + print(f"PASS: {actual!r}", flush=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("phase", choices=["prepare", "verify", "cleanup"]) + parser.add_argument("--database", required=True, type=identifier) + parser.add_argument("--uri", default="spark://localhost:15002?api=connect&auth_type=none&tls=false") + parser.add_argument("--driver", default="spark", help="Foundry driver name or native library path") + parser.add_argument("--created-marker", type=Path, + help="prepare only: record database ownership for the shell runner's cleanup") + args = parser.parse_args() + if args.created_marker: + if args.phase != "prepare": + parser.error("--created-marker is only valid for prepare") + if args.created_marker.exists(): + parser.error("--created-marker must be a new file") + database = args.database + table = f"{database}.orders" + + with dbapi.connect(driver=args.driver, db_kwargs={"uri": args.uri}, autocommit=True) as conn: + with conn.cursor() as cursor: + if args.phase == "prepare": + # No IF NOT EXISTS: refuse to reuse another run's database. + run(cursor, f"CREATE DATABASE {database}") + # A failed/ambiguous CREATE must not authorize automatic deletion. + if args.created_marker: + with args.created_marker.open("x") as marker: + marker.write(database) + run(cursor, f"""CREATE TABLE {table} ( + id INT, region STRING, amount INT, ts BIGINT + ) USING hudi + TBLPROPERTIES ( + type = 'cow', primaryKey = 'id', preCombineField = 'ts', Review Comment: Added checks for MERGE, MoR reads/compaction, partitioned tables with Trino readback, CALL, time travel, and incremental reads. Following the community meeting, required v1 delivery focuses on SQL reads/writes, not DataFrame-write integration or general job submission. ########## hudi-lakehouse/local-dev/example/connect_smoke.py: ########## @@ -0,0 +1,97 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""ADBC-only CRUD smoke test; each invocation creates a new Connect session. + +Use prepare -> restart the endpoint -> verify -> cleanup to check that both +catalog metadata and data survive. Only a uniquely named test database is used. +""" +import argparse +import re +from pathlib import Path + +from adbc_driver_manager import dbapi + + +def identifier(value): + # This CLI interpolates only its own test identifiers, never arbitrary SQL. + if not re.fullmatch(r"connect_smoke_[a-z0-9_]+", value): + raise argparse.ArgumentTypeError("database must match connect_smoke_[a-z0-9_]+") + return value + + +def run(cursor, sql, expected=None): + print(sql, flush=True) + cursor.execute(sql) + # DDL/DML do not necessarily have result sets; never fetch unconditionally. + if expected is not None: + actual = [tuple(row) for row in cursor.fetchall()] + if actual != expected: + raise AssertionError(f"Expected {expected!r}, got {actual!r}") + print(f"PASS: {actual!r}", flush=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("phase", choices=["prepare", "verify", "cleanup"]) + parser.add_argument("--database", required=True, type=identifier) + parser.add_argument("--uri", default="spark://localhost:15002?api=connect&auth_type=none&tls=false") + parser.add_argument("--driver", default="spark", help="Foundry driver name or native library path") + parser.add_argument("--created-marker", type=Path, + help="prepare only: record database ownership for the shell runner's cleanup") + args = parser.parse_args() + if args.created_marker: + if args.phase != "prepare": + parser.error("--created-marker is only valid for prepare") + if args.created_marker.exists(): + parser.error("--created-marker must be a new file") + database = args.database + table = f"{database}.orders" + + with dbapi.connect(driver=args.driver, db_kwargs={"uri": args.uri}, autocommit=True) as conn: + with conn.cursor() as cursor: + if args.phase == "prepare": + # No IF NOT EXISTS: refuse to reuse another run's database. + run(cursor, f"CREATE DATABASE {database}") + # A failed/ambiguous CREATE must not authorize automatic deletion. + if args.created_marker: + with args.created_marker.open("x") as marker: + marker.write(database) + run(cursor, f"""CREATE TABLE {table} ( + id INT, region STRING, amount INT, ts BIGINT + ) USING hudi + TBLPROPERTIES ( + type = 'cow', primaryKey = 'id', preCombineField = 'ts', + 'hoodie.write.table.version' = '8', Review Comment: Added `hudi.defaults` and `HUDI_CONF_DIR`. Local-dev now defaults to table version 8 with auto-upgrade disabled; the generic chart remains configurable. The live run created v8 tables without statement-level overrides. Trino readback remains tracked in #20233. ########## hudi-lakehouse/local-dev/scripts/test-spark-connect.sh: ########## @@ -0,0 +1,94 @@ +#!/usr/bin/env bash +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Opt-in acceptance test for an ALREADY deployed endpoint. Restarts only the +# named deployment; do not run against a service used by other clients. +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +PYTHON="${PYTHON:-python3}" +NAMESPACE="${NAMESPACE:-hudi-lakehouse}" +DEPLOYMENT="${DEPLOYMENT:-hudi-spark-connect}" +SERVICE="${SERVICE:-hudi-spark-connect}" +PORT="${PORT:-15002}" +SERVICE_PORT="${SERVICE_PORT:-15002}" +DRIVER="${ADBC_DRIVER:-spark}" +DATABASE="connect_smoke_$(date +%s)_${RANDOM}" +SCRIPT="$HERE/example/connect_smoke.py" +LOG_DIR="$(mktemp -d)" +PF_PID="" +CREATED_MARKER="$LOG_DIR/database-created" + +if [[ "${1:-}" != "--restart-endpoint" || $# != 1 ]]; then + echo "Usage: $0 --restart-endpoint" >&2 + echo "Runs CRUD and restarts $NAMESPACE/$DEPLOYMENT; active sessions will be lost." >&2 + exit 1 +fi +[[ "$PORT" =~ ^[0-9]+$ && "$SERVICE_PORT" =~ ^[0-9]+$ ]] || { echo "ports must be integers" >&2; exit 1; } + +stop_forward() { + if [[ -n "$PF_PID" ]]; then + kill "$PF_PID" 2>/dev/null || true + wait "$PF_PID" 2>/dev/null || true + PF_PID="" + fi +} +start_forward() { + kubectl -n "$NAMESPACE" port-forward --address 127.0.0.1 "svc/$SERVICE" "$PORT:$SERVICE_PORT" >"$LOG_DIR/port-forward.log" 2>&1 & + PF_PID=$! + for ((i=0; i<60; i++)); do + if ! kill -0 "$PF_PID" 2>/dev/null; then + cat "$LOG_DIR/port-forward.log" >&2 + return 1 + fi + if grep -q "Forwarding from 127.0.0.1:$PORT" "$LOG_DIR/port-forward.log"; then return 0; fi + sleep 1 + done + echo "Timed out waiting for port-forward; see $LOG_DIR" >&2 + return 1 +} +smoke() { + "$PYTHON" "$SCRIPT" "$@" --database "$DATABASE" --driver "$DRIVER" \ + --uri "spark://127.0.0.1:$PORT?api=connect&auth_type=none&tls=false" +} +finish() { + status=$? + trap - EXIT + if [[ -f "$CREATED_MARKER" && -n "$PF_PID" ]] && kill -0 "$PF_PID" 2>/dev/null; then + smoke cleanup || echo "Cleanup failed; retry connect_smoke.py cleanup --database $DATABASE after reconnecting." >&2 + elif [[ -f "$CREATED_MARKER" ]]; then + echo "Reconnect and run connect_smoke.py cleanup --database $DATABASE" >&2 + fi + stop_forward + echo "Port-forward log: $LOG_DIR/port-forward.log" + exit "$status" +} +trap finish EXIT +trap 'exit 130' INT +trap 'exit 143' TERM + +kubectl -n "$NAMESPACE" rollout status "deployment/$DEPLOYMENT" --timeout=300s +start_forward +echo "Test database: $DATABASE (use connect_smoke.py cleanup if prepare is interrupted)" +smoke prepare --created-marker "$CREATED_MARKER" +stop_forward +kubectl -n "$NAMESPACE" rollout restart "deployment/$DEPLOYMENT" +kubectl -n "$NAMESPACE" rollout status "deployment/$DEPLOYMENT" --timeout=300s +start_forward +smoke verify +smoke cleanup +rm "$CREATED_MARKER" Review Comment: Added a same-session freshness test covering external commits, schema changes, and table recreation. Explicit REFRESH TABLE is tested separately. ########## hudi-lakehouse/charts/hudi-spark-connect/values.yaml: ########## @@ -0,0 +1,66 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +nameOverride: "" +fullnameOverride: "" + +image: + repository: hudi-lakehouse-spark-connect + tag: "4.1.3-hudi1.2.0" + pullPolicy: IfNotPresent +imagePullSecrets: [] + +# v1 runs one shared Spark driver with local worker threads in the same pod. +# This is NOT a Kubernetes executor deployment or a multi-tenant service. +spark: + cores: 2 + driverMemory: 2g + shufflePartitions: 2 + extraConf: {} + +catalog: + # Required for durable metadata; no embedded Derby catalog in the endpoint. + metastoreUri: "" + # Required, e.g. s3a://my-bucket/warehouse. Local pod storage is not durable. + warehouse: "" + +storage: + s3: + endpoint: "" + region: us-east-1 + pathStyleAccess: false + # Leave empty to use the SDK's default credential provider chain. + # Never put credentials in Spark properties or the ConfigMap. + existingSecret: "" + accessKeyKey: accessKey + secretKeyKey: secretKey + # Optional key in existingSecret for temporary credentials. + sessionTokenKey: "" + +# No public ingress or unauthenticated LoadBalancer is provisioned. +service: + port: 15002 + +resources: Review Comment: Tracked Spark UI access, event logs/history server, metrics, and request/job/commit correlation in #20233. These are not implemented yet. Spark job status and timeline inspection were essential during the fault tests. ########## hudi-lakehouse/charts/hudi-spark-connect/templates/deployment.yaml: ########## @@ -0,0 +1,121 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "hudi-spark-connect.fullname" . }} + labels: + {{- include "hudi-spark-connect.labels" . | nindent 4 }} +spec: + replicas: 1 + # Sessions are in-memory; do not load-balance between independent drivers. + strategy: + type: Recreate + selector: + matchLabels: + {{- include "hudi-spark-connect.selectorLabels" . | nindent 6 }} + template: + metadata: + labels: + {{- include "hudi-spark-connect.selectorLabels" . | nindent 8 }} + annotations: + checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }} + spec: + automountServiceAccountToken: false + terminationGracePeriodSeconds: 60 + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + securityContext: + runAsNonRoot: true + runAsUser: 185 + runAsGroup: 185 + fsGroup: 185 + containers: + - name: spark-connect + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + ports: + - name: connect + containerPort: 15002 + env: + - name: SPARK_LOCAL_DIRS + value: /tmp/spark-local + - name: AWS_REGION + value: {{ .Values.storage.s3.region | quote }} + {{- with .Values.storage.s3.existingSecret }} + - name: AWS_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.storage.s3.accessKeyKey | quote }} + - name: AWS_SECRET_ACCESS_KEY + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.storage.s3.secretKeyKey | quote }} + {{- if $.Values.storage.s3.sessionTokenKey }} + - name: AWS_SESSION_TOKEN + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.storage.s3.sessionTokenKey | quote }} + {{- end }} + {{- end }} + volumeMounts: + - name: config + mountPath: /opt/spark/conf/spark-defaults.conf + subPath: spark-defaults.conf + readOnly: true + - name: scratch + mountPath: /tmp + # TCP probes check the listener, not catalog or write correctness. + # The ADBC smoke test verifies the complete data path. + startupProbe: + tcpSocket: + port: connect + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: Review Comment: Tracked in #20233. The current probe checks only the listener. SELECT 1 would check SQL execution, but HMS and S3 access need separate checks. ########## hudi-lakehouse/charts/hudi-spark-connect/README.md: ########## @@ -0,0 +1,171 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +# Hudi Spark Connect endpoint + +Deploys Spark Connect 4.1.3 with Hudi 1.2.0, Scala 2.13, and Java 17. +Tables use an external Hive Metastore and S3-compatible storage. +See the [Dockerfile](../../images/spark-connect/Dockerfile) for dependency versions. + +## Local-dev quickstart + +Requires Docker, minikube, Helm, and kubectl. Run from the repository root: + +```bash +# Start MinIO, Hive Metastore, and Spark Connect +./hudi-lakehouse/scripts/quickstart.sh --spark-connect + +# Also write example data and query it through Spark Connect (requires Maven and a JDK) +./hudi-lakehouse/scripts/quickstart.sh --run-example --spark-connect +``` + +The example uses the existing Spark writer to populate and register `default.trips`, +then prints city counts: `chennai = 33`, `san_francisco = 34`, `sao_paulo = 33`. +The query runs through PySpark inside the endpoint pod; no host client is needed. +Omit `--spark-connect` to query through Trino instead. + +To rebuild the image and reload the running endpoint: + +```bash +./hudi-lakehouse/scripts/quickstart.sh --spark-connect --rebuild-images +kubectl -n hudi-lakehouse rollout restart deployment/hudi-spark-connect +kubectl -n hudi-lakehouse rollout status deployment/hudi-spark-connect +``` + +## Install against existing infrastructure + +Build an image and supply your metastore, warehouse, and storage credentials: + +```bash +./hudi-lakehouse/scripts/build-connect-image.sh --registry my.registry/team --push +helm upgrade --install hudi-spark-connect hudi-lakehouse/charts/hudi-spark-connect \ + --namespace hudi-lakehouse --create-namespace \ + --set image.repository=my.registry/team/hudi-lakehouse-spark-connect \ + --set catalog.metastoreUri=thrift://my-hms:9083 \ + --set catalog.warehouse=s3a://my-bucket/connect \ + --set storage.s3.existingSecret=my-s3-credentials \ + --wait --timeout 10m +``` + +The Secret must be in the same namespace and contain `accessKey` and `secretKey`. +For temporary credentials, also set `storage.s3.sessionTokenKey`. +Without a Secret, Hadoop uses its default credential provider chain. +Local-dev credentials are for development only. + +Other settings in [values.yaml](values.yaml): + +| Setting | Purpose | +|---|---| +| `storage.s3.endpoint`, `region`, `pathStyleAccess` | MinIO or custom S3 endpoints | +| `spark.cores` | Local worker threads | +| `spark.driverMemory` | JVM heap; keep the pod memory limit higher | +| `resources` | Pod CPU and memory requests/limits | +| `spark.extraConf` | Additional Spark properties; no credentials or overrides of chart-owned settings | + +The endpoint runs one shared driver in local mode, without separate executors. +Keep one replica. Restarting it ends active sessions; tables remain in the +external metastore and object store. + +The service has no authentication or TLS. Keep it on a trusted network: +any client that can connect can read and write using the server's storage credentials. + +## Connect and validate with ADBC + +Forward the endpoint for local clients: + +```bash +kubectl -n hudi-lakehouse port-forward svc/hudi-spark-connect 15002:15002 +``` + +| Client | URI | +|---|---| +| PySpark | `sc://localhost:15002` | +| Foundry ADBC | `spark://localhost:15002?api=connect&auth_type=none&tls=false` | + +ADBC requires `api=connect`; its default is Thrift. In-cluster clients can use +`hudi-spark-connect.hudi-lakehouse.svc:15002` without port-forwarding. +The ADBC driver is installed on the client, not the server. + +For the CRUD and restart test, prepare a Python 3.11 environment: + +```bash +python3.11 -m venv .venv-connect +. .venv-connect/bin/activate +pip install -r hudi-lakehouse/local-dev/example/connect-requirements.txt dbc==0.3.1 +dbc install 'spark=0.2.1' +``` + +Stop any manual port-forward before running the test; it manages its own: + +```bash +PYTHON="$VIRTUAL_ENV/bin/python" \ + ./hudi-lakehouse/local-dev/scripts/test-spark-connect.sh --restart-endpoint +``` + +The test creates a temporary table, checks INSERT/UPDATE/DELETE, restarts the +endpoint, verifies persistence, and cleans up. It interrupts active sessions. +On failure, follow the printed cleanup instructions. + +<a id="spark-connect-client-tutorials"></a> + +## BI clients + +- **Superset 5.0:** requires a custom SQLAlchemy dialect and engine spec for ADBC; + neither is included here. The built-in PyHive connector uses Thrift, not Connect. + See [Superset engine specs](https://github.com/apache/superset/blob/5.0.0/superset/db_engine_specs/README.md). +- **Zeppelin 0.12:** use ADBC from the `%python` interpreter, with the client + packages above installed in its Python environment. See the + [Python interpreter guide](https://zeppelin.apache.org/docs/0.12.0/interpreter/python.html). + +[Foundry Spark ADBC 0.2.1](https://adbc-drivers.org/drivers/spark/) uses autocommit, +buffers results before fetching, and does not support statement cancellation. +Use `LIMIT` for interactive queries and check table state before retrying writes. Review Comment: The fault tests showed that disconnecting during INSERT did not prevent commit, and retrying after a lost response produced duplicates. After a driver restart at the requested-instant stage, the next write rolled back the unfinished instant and committed. These are observed outcomes, not universal retry guarantees. Foundry `adbc_cancel()` returned a client cancellation error; server-side write cancellation was not established. Gateway-level cancellation stopped Spark jobs, but Chat UI Stop did not. That fix is tracked in #20232; broader fault coverage is tracked in #20233. ########## hudi-lakehouse/charts/hudi-spark-connect/README.md: ########## @@ -0,0 +1,171 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +# Hudi Spark Connect endpoint + +Deploys Spark Connect 4.1.3 with Hudi 1.2.0, Scala 2.13, and Java 17. +Tables use an external Hive Metastore and S3-compatible storage. +See the [Dockerfile](../../images/spark-connect/Dockerfile) for dependency versions. + +## Local-dev quickstart + +Requires Docker, minikube, Helm, and kubectl. Run from the repository root: + +```bash +# Start MinIO, Hive Metastore, and Spark Connect +./hudi-lakehouse/scripts/quickstart.sh --spark-connect + +# Also write example data and query it through Spark Connect (requires Maven and a JDK) +./hudi-lakehouse/scripts/quickstart.sh --run-example --spark-connect +``` + +The example uses the existing Spark writer to populate and register `default.trips`, +then prints city counts: `chennai = 33`, `san_francisco = 34`, `sao_paulo = 33`. +The query runs through PySpark inside the endpoint pod; no host client is needed. +Omit `--spark-connect` to query through Trino instead. + +To rebuild the image and reload the running endpoint: + +```bash +./hudi-lakehouse/scripts/quickstart.sh --spark-connect --rebuild-images +kubectl -n hudi-lakehouse rollout restart deployment/hudi-spark-connect +kubectl -n hudi-lakehouse rollout status deployment/hudi-spark-connect +``` + +## Install against existing infrastructure + +Build an image and supply your metastore, warehouse, and storage credentials: + +```bash +./hudi-lakehouse/scripts/build-connect-image.sh --registry my.registry/team --push +helm upgrade --install hudi-spark-connect hudi-lakehouse/charts/hudi-spark-connect \ + --namespace hudi-lakehouse --create-namespace \ + --set image.repository=my.registry/team/hudi-lakehouse-spark-connect \ + --set catalog.metastoreUri=thrift://my-hms:9083 \ + --set catalog.warehouse=s3a://my-bucket/connect \ + --set storage.s3.existingSecret=my-s3-credentials \ + --wait --timeout 10m +``` + +The Secret must be in the same namespace and contain `accessKey` and `secretKey`. +For temporary credentials, also set `storage.s3.sessionTokenKey`. +Without a Secret, Hadoop uses its default credential provider chain. +Local-dev credentials are for development only. + +Other settings in [values.yaml](values.yaml): + +| Setting | Purpose | +|---|---| +| `storage.s3.endpoint`, `region`, `pathStyleAccess` | MinIO or custom S3 endpoints | +| `spark.cores` | Local worker threads | +| `spark.driverMemory` | JVM heap; keep the pod memory limit higher | +| `resources` | Pod CPU and memory requests/limits | +| `spark.extraConf` | Additional Spark properties; no credentials or overrides of chart-owned settings | + +The endpoint runs one shared driver in local mode, without separate executors. +Keep one replica. Restarting it ends active sessions; tables remain in the +external metastore and object store. + +The service has no authentication or TLS. Keep it on a trusted network: +any client that can connect can read and write using the server's storage credentials. + +## Connect and validate with ADBC + +Forward the endpoint for local clients: + +```bash +kubectl -n hudi-lakehouse port-forward svc/hudi-spark-connect 15002:15002 +``` + +| Client | URI | +|---|---| +| PySpark | `sc://localhost:15002` | +| Foundry ADBC | `spark://localhost:15002?api=connect&auth_type=none&tls=false` | + +ADBC requires `api=connect`; its default is Thrift. In-cluster clients can use +`hudi-spark-connect.hudi-lakehouse.svc:15002` without port-forwarding. +The ADBC driver is installed on the client, not the server. + +For the CRUD and restart test, prepare a Python 3.11 environment: + +```bash +python3.11 -m venv .venv-connect +. .venv-connect/bin/activate +pip install -r hudi-lakehouse/local-dev/example/connect-requirements.txt dbc==0.3.1 +dbc install 'spark=0.2.1' +``` + +Stop any manual port-forward before running the test; it manages its own: + +```bash +PYTHON="$VIRTUAL_ENV/bin/python" \ + ./hudi-lakehouse/local-dev/scripts/test-spark-connect.sh --restart-endpoint +``` + +The test creates a temporary table, checks INSERT/UPDATE/DELETE, restarts the +endpoint, verifies persistence, and cleans up. It interrupts active sessions. +On failure, follow the printed cleanup instructions. + +<a id="spark-connect-client-tutorials"></a> + +## BI clients + +- **Superset 5.0:** requires a custom SQLAlchemy dialect and engine spec for ADBC; + neither is included here. The built-in PyHive connector uses Thrift, not Connect. + See [Superset engine specs](https://github.com/apache/superset/blob/5.0.0/superset/db_engine_specs/README.md). +- **Zeppelin 0.12:** use ADBC from the `%python` interpreter, with the client + packages above installed in its Python environment. See the + [Python interpreter guide](https://zeppelin.apache.org/docs/0.12.0/interpreter/python.html). + +[Foundry Spark ADBC 0.2.1](https://adbc-drivers.org/drivers/spark/) uses autocommit, Review Comment: Tested 100k rows (~98 MiB payload) with the default 2g heap. Foundry ADBC hit its 4 MiB gRPC receive limit; setting `spark.connect.grpc.arrow.maxBatchSize=1m` allowed the tested reads to complete. This limits batches, not total results. Million-row testing, TIMESTAMP_NTZ, `_hoodie_*` checks, and the nullable-array write failure are tracked in #20233. Gateway timestamp/struct serialization fixes are tracked in #20232. ########## hudi-lakehouse/charts/hudi-spark-connect/README.md: ########## @@ -0,0 +1,171 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> + +# Hudi Spark Connect endpoint + +Deploys Spark Connect 4.1.3 with Hudi 1.2.0, Scala 2.13, and Java 17. +Tables use an external Hive Metastore and S3-compatible storage. +See the [Dockerfile](../../images/spark-connect/Dockerfile) for dependency versions. + +## Local-dev quickstart + +Requires Docker, minikube, Helm, and kubectl. Run from the repository root: + +```bash +# Start MinIO, Hive Metastore, and Spark Connect +./hudi-lakehouse/scripts/quickstart.sh --spark-connect + +# Also write example data and query it through Spark Connect (requires Maven and a JDK) +./hudi-lakehouse/scripts/quickstart.sh --run-example --spark-connect +``` + +The example uses the existing Spark writer to populate and register `default.trips`, +then prints city counts: `chennai = 33`, `san_francisco = 34`, `sao_paulo = 33`. +The query runs through PySpark inside the endpoint pod; no host client is needed. +Omit `--spark-connect` to query through Trino instead. + +To rebuild the image and reload the running endpoint: + +```bash +./hudi-lakehouse/scripts/quickstart.sh --spark-connect --rebuild-images +kubectl -n hudi-lakehouse rollout restart deployment/hudi-spark-connect +kubectl -n hudi-lakehouse rollout status deployment/hudi-spark-connect +``` + +## Install against existing infrastructure + +Build an image and supply your metastore, warehouse, and storage credentials: + +```bash +./hudi-lakehouse/scripts/build-connect-image.sh --registry my.registry/team --push +helm upgrade --install hudi-spark-connect hudi-lakehouse/charts/hudi-spark-connect \ + --namespace hudi-lakehouse --create-namespace \ + --set image.repository=my.registry/team/hudi-lakehouse-spark-connect \ + --set catalog.metastoreUri=thrift://my-hms:9083 \ + --set catalog.warehouse=s3a://my-bucket/connect \ + --set storage.s3.existingSecret=my-s3-credentials \ + --wait --timeout 10m +``` + +The Secret must be in the same namespace and contain `accessKey` and `secretKey`. +For temporary credentials, also set `storage.s3.sessionTokenKey`. +Without a Secret, Hadoop uses its default credential provider chain. +Local-dev credentials are for development only. + +Other settings in [values.yaml](values.yaml): + +| Setting | Purpose | +|---|---| +| `storage.s3.endpoint`, `region`, `pathStyleAccess` | MinIO or custom S3 endpoints | +| `spark.cores` | Local worker threads | +| `spark.driverMemory` | JVM heap; keep the pod memory limit higher | +| `resources` | Pod CPU and memory requests/limits | +| `spark.extraConf` | Additional Spark properties; no credentials or overrides of chart-owned settings | + +The endpoint runs one shared driver in local mode, without separate executors. +Keep one replica. Restarting it ends active sessions; tables remain in the +external metastore and object store. + +The service has no authentication or TLS. Keep it on a trusted network: +any client that can connect can read and write using the server's storage credentials. + +## Connect and validate with ADBC + +Forward the endpoint for local clients: + +```bash +kubectl -n hudi-lakehouse port-forward svc/hudi-spark-connect 15002:15002 +``` + +| Client | URI | +|---|---| +| PySpark | `sc://localhost:15002` | +| Foundry ADBC | `spark://localhost:15002?api=connect&auth_type=none&tls=false` | + +ADBC requires `api=connect`; its default is Thrift. In-cluster clients can use +`hudi-spark-connect.hudi-lakehouse.svc:15002` without port-forwarding. +The ADBC driver is installed on the client, not the server. + +For the CRUD and restart test, prepare a Python 3.11 environment: + +```bash +python3.11 -m venv .venv-connect +. .venv-connect/bin/activate +pip install -r hudi-lakehouse/local-dev/example/connect-requirements.txt dbc==0.3.1 +dbc install 'spark=0.2.1' +``` + +Stop any manual port-forward before running the test; it manages its own: + +```bash +PYTHON="$VIRTUAL_ENV/bin/python" \ + ./hudi-lakehouse/local-dev/scripts/test-spark-connect.sh --restart-endpoint +``` + +The test creates a temporary table, checks INSERT/UPDATE/DELETE, restarts the +endpoint, verifies persistence, and cleans up. It interrupts active sessions. +On failure, follow the printed cleanup instructions. + +<a id="spark-connect-client-tutorials"></a> + +## BI clients Review Comment: Linked #20232 from both READMEs and documented the plan to keep the initial SQLAlchemy adapter under `hudi-lakehouse`. Gateway and BI integrations remain separate PRs. The combined endpoint/Gateway branch passed a real three-turn Chat UI query flow; its remaining cancellation and serialization gaps are tracked in #20232. ########## hudi-lakehouse/charts/hudi-spark-connect/templates/configmap.yaml: ########## @@ -0,0 +1,71 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +{{- $uri := required "catalog.metastoreUri is required (use an external Hive Metastore)" .Values.catalog.metastoreUri -}} +{{- $warehouse := required "catalog.warehouse is required (use durable object storage)" .Values.catalog.warehouse -}} +{{- $conf := dict + "spark.master" (printf "local[%v]" .Values.spark.cores) + "spark.driver.memory" .Values.spark.driverMemory + "spark.connect.grpc.binding.address" "0.0.0.0" + "spark.connect.grpc.binding.port" "15002" + "spark.sql.catalogImplementation" "hive" + "spark.hadoop.hive.metastore.uris" $uri + "spark.sql.warehouse.dir" $warehouse + "spark.sql.extensions" "org.apache.spark.sql.hudi.HoodieSparkSessionExtension" + "spark.sql.catalog.spark_catalog" "org.apache.spark.sql.hudi.catalog.HoodieCatalog" + "spark.serializer" "org.apache.spark.serializer.KryoSerializer" + "spark.kryo.registrator" "org.apache.spark.HoodieSparkKryoRegistrar" + "spark.sql.shuffle.partitions" .Values.spark.shufflePartitions + "spark.hadoop.fs.s3a.impl" "org.apache.hadoop.fs.s3a.S3AFileSystem" + "spark.hadoop.fs.s3a.endpoint.region" .Values.storage.s3.region + "spark.hadoop.fs.s3a.path.style.access" .Values.storage.s3.pathStyleAccess +-}} +{{- with .Values.storage.s3.endpoint -}} +{{- $_ := set $conf "spark.hadoop.fs.s3a.endpoint" . -}} +{{- end -}} +{{- range $key, $value := .Values.spark.extraConf -}} +{{- if hasKey $conf $key -}} +{{- fail (printf "%s is managed by the chart; use the corresponding value instead" $key) -}} +{{- end -}} +{{- if not (regexMatch "^spark\\.[A-Za-z0-9_.-]+$" $key) -}} Review Comment: Added configurable `spark.master`, ServiceAccount/token settings, and Kubernetes executor configuration, with a fixed-executor example. ########## hudi-lakehouse/charts/hudi-spark-connect/templates/deployment.yaml: ########## @@ -0,0 +1,121 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "hudi-spark-connect.fullname" . }} + labels: + {{- include "hudi-spark-connect.labels" . | nindent 4 }} +spec: + replicas: 1 + # Sessions are in-memory; do not load-balance between independent drivers. + strategy: + type: Recreate + selector: + matchLabels: + {{- include "hudi-spark-connect.selectorLabels" . | nindent 6 }} + template: + metadata: + labels: + {{- include "hudi-spark-connect.selectorLabels" . | nindent 8 }} + annotations: + checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }} + spec: + automountServiceAccountToken: false + terminationGracePeriodSeconds: 60 + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + securityContext: + runAsNonRoot: true + runAsUser: 185 + runAsGroup: 185 + fsGroup: 185 + containers: + - name: spark-connect + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + ports: + - name: connect + containerPort: 15002 + env: + - name: SPARK_LOCAL_DIRS + value: /tmp/spark-local + - name: AWS_REGION + value: {{ .Values.storage.s3.region | quote }} + {{- with .Values.storage.s3.existingSecret }} + - name: AWS_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.storage.s3.accessKeyKey | quote }} + - name: AWS_SECRET_ACCESS_KEY + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.storage.s3.secretKeyKey | quote }} + {{- if $.Values.storage.s3.sessionTokenKey }} + - name: AWS_SESSION_TOKEN + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.storage.s3.sessionTokenKey | quote }} + {{- end }} + {{- end }} + volumeMounts: + - name: config + mountPath: /opt/spark/conf/spark-defaults.conf + subPath: spark-defaults.conf + readOnly: true + - name: scratch + mountPath: /tmp + # TCP probes check the listener, not catalog or write correctness. + # The ADBC smoke test verifies the complete data path. + startupProbe: + tcpSocket: + port: connect + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: + tcpSocket: + port: connect + periodSeconds: 10 + resources: + {{- toYaml .Values.resources | nindent 12 }} + volumes: + - name: config + configMap: + name: {{ include "hudi-spark-connect.fullname" . }}-config + - name: scratch + emptyDir: {} Review Comment: Added optional `scratch.sizeLimit` and disk-sizing guidance, including ephemeral-storage requests/limits and eviction risks. This covers driver scratch only; executor scratch needs separate sizing. ########## .github/workflows/agent_gateway_ci.yml: ########## @@ -110,6 +110,13 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + - name: Spark Connect chart contract tests + run: | + pip install PyYAML==6.0.3 + python -m unittest discover -s hudi-lakehouse/tests -v Review Comment: Added an opt-in kind live-CI task to #20233, using allowlisted actions. The 51 offline tests pass, but the local Helm and Chat UI runs do not replace repeatable live CI. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
