Skip to content

Commit 93f5368

Browse files
oavdeevoavdeev
committed
integration test for k8s spark operator support
Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
1 parent 5b76f9a commit 93f5368

9 files changed

Lines changed: 77 additions & 8 deletions

File tree

‎infra/scripts/codebuild_runner.py‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import sys
1212
import argparse
1313
import boto3
14+
from botocore.config import Config
1415

1516

1617
class LogTailer:
@@ -125,7 +126,14 @@ def source_version_from_prow_job_spec(job_spec: Dict[str, Any]) -> str:
125126

126127
async def run_build(project_name: str, source_version: str, source_location: str):
127128
print(f"Building {project_name} at {source_version}", file=sys.stderr)
128-
logs_client = boto3.client("logs", region_name="us-west-2")
129+
130+
config = Config(
131+
retries = {
132+
'max_attempts': 10,
133+
}
134+
)
135+
136+
logs_client = boto3.client("logs", region_name="us-west-2", config=config)
129137
codebuild_client = boto3.client("codebuild", region_name="us-west-2")
130138

131139
print("Submitting the build..", file=sys.stderr)

‎infra/scripts/setup-e2e-env-sparkop.sh‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
#!/bin/bash
2+
3+
make compile-protos-python
4+
5+
python -m pip install --upgrade pip==20.2 setuptools wheel
6+
7+
python -m pip install -qr sdk/python/requirements-dev.txt
8+
python -m pip install -qr tests/requirements.txt
9+
10+
# Using mvn -q to make it less verbose. This step happens after docker containers were
11+
# succesfully built so it should be unlikely to fail, therefore we likely won't need detailed logs.
12+
echo "########## Building ingestion jar"
13+
TIMEFORMAT='########## took %R seconds'
14+
15+
time make build-java-no-tests REVISION=develop MAVEN_EXTRA_OPTS="-q --no-transfer-progress"

‎infra/scripts/test-end-to-end-sparkop.sh‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
#!/usr/bin/env bash
2+
3+
set -euo pipefail
4+
5+
pip install "s3fs" "boto3" "urllib3>=1.25.4"
6+
7+
export DISABLE_FEAST_SERVICE_FIXTURES=1
8+
export DISABLE_SERVICE_FIXTURES=1
9+
10+
export FEAST_SPARK_K8S_NAMESPACE=sparkop
11+
12+
PYTHONPATH=sdk/python pytest tests/e2e/ \
13+
--feast-version develop \
14+
--core-url sparkop-feast-core:6565 \
15+
--serving-url sparkop-feast-online-serving:6566 \
16+
--env k8s \
17+
--staging-path $STAGING_PATH \
18+
--redis-url sparkop-redis-master.sparkop.svc.cluster.local:6379 \
19+
--kafka-brokers sparkop-kafka.sparkop.svc.cluster.local:9092 \
20+
-m "not bq"

‎sdk/python/feast/constants.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -184,7 +184,7 @@ class ConfigOptions(metaclass=ConfigMeta):
184184
SPARK_K8S_NAMESPACE = "default"
185185

186186
# expect k8s spark operator to be running in the same cluster as Feast
187-
SPARK_K8S_USE_INCLUSTER_CONFIG = True
187+
SPARK_K8S_USE_INCLUSTER_CONFIG = "True"
188188

189189
# SparkApplication resource template
190190
SPARK_K8S_JOB_TEMPLATE_PATH = None

‎sdk/python/feast/pyspark/launchers/k8s/k8s.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -223,6 +223,7 @@ def historical_feature_retrieval(
223223
jars=[],
224224
extra_metadata={METADATA_OUTPUT_URI: job_params.get_destination_path()},
225225
arguments=job_params.get_arguments(),
226+
namespace=self._namespace,
226227
)
227228

228229
job_info = _submit_job(
@@ -276,6 +277,7 @@ def offline_to_online_ingestion(
276277
jars=[],
277278
extra_metadata={},
278279
arguments=ingestion_job_params.get_arguments(),
280+
namespace=self._namespace,
279281
)
280282

281283
job_info = _submit_job(
@@ -317,6 +319,7 @@ def start_stream_to_online_ingestion(
317319
jars=extra_jar_paths,
318320
extra_metadata={METADATA_JOBHASH: job_hash},
319321
arguments=ingestion_job_params.get_arguments(),
322+
namespace=self._namespace,
320323
)
321324

322325
job_info = _submit_job(

‎sdk/python/feast/pyspark/launchers/k8s/k8s_utils.py‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,14 +112,19 @@ def _prepare_job_resource(
112112
jars: List[str],
113113
extra_metadata: Dict[str, str],
114114
arguments: List[str],
115+
namespace: str,
115116
) -> Dict[str, Any]:
116117
""" Prepare SparkApplication custom resource configs """
117118
job = deepcopy(job_template)
118119

119120
labels = {LABEL_JOBID: job_id, LABEL_JOBTYPE: job_type}
120121

121122
_add_keys(job, ("metadata", "labels"), labels)
122-
_add_keys(job, ("metadata",), dict(name=_job_id_to_resource_name(job_id)))
123+
_add_keys(
124+
job,
125+
("metadata",),
126+
dict(name=_job_id_to_resource_name(job_id), namespace=namespace),
127+
)
123128
_add_keys(job, ("spec",), dict(mainClass=main_class))
124129
_add_keys(job, ("spec",), dict(mainApplicationFile=main_application_file))
125130
_add_keys(job, ("spec",), dict(arguments=arguments))
@@ -179,7 +184,7 @@ def _k8s_state_to_feast(k8s_state: str) -> SparkJobStatus:
179184

180185
def _resource_to_job_info(resource: Dict[str, Any]) -> JobInfo:
181186
labels = resource["metadata"]["labels"]
182-
sparkConf = resource["spec"].get("sparkConf")
187+
sparkConf = resource["spec"].get("sparkConf", {})
183188

184189
if "status" in resource:
185190
state = _k8s_state_to_feast(resource["status"]["applicationState"]["state"])

‎tests/e2e/conftest.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,9 @@ def pytest_addoption(parser):
99
parser.addoption("--job-service-url", action="store", default="localhost:6568")
1010
parser.addoption("--kafka-brokers", action="store", default="localhost:9092")
1111

12-
parser.addoption("--env", action="store", help="local|aws|gcloud", default="local")
12+
parser.addoption(
13+
"--env", action="store", help="local|aws|gcloud|k8s", default="local"
14+
)
1315
parser.addoption("--with-job-service", action="store_true")
1416
parser.addoption("--staging-path", action="store")
1517
parser.addoption("--dataproc-cluster-name", action="store")

‎tests/e2e/fixtures/client.py‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,19 @@ def feast_client(
7979
local_staging_path, "historical_output"
8080
),
8181
)
82+
elif pytestconfig.getoption("env") == "k8s":
83+
return Client(
84+
core_url=f"{feast_core[0]}:{feast_core[1]}",
85+
serving_url=f"{feast_serving[0]}:{feast_serving[1]}",
86+
spark_launcher="k8s",
87+
spark_staging_location=os.path.join(local_staging_path, "k8s"),
88+
spark_ingestion_jar=ingestion_job_jar,
89+
redis_host=pytestconfig.getoption("redis_url").split(":")[0],
90+
redis_port=pytestconfig.getoption("redis_url").split(":")[1],
91+
historical_feature_output_location=os.path.join(
92+
local_staging_path, "historical_output"
93+
),
94+
)
8295
else:
8396
raise KeyError(f"Unknown environment {pytestconfig.getoption('env')}")
8497

‎tests/e2e/test_historical_features.py‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
from datetime import datetime, timedelta
22
from typing import Union
3-
from urllib.parse import urlparse
3+
from urllib.parse import urlparse, urlunparse
44

55
import gcsfs
66
import numpy as np
@@ -24,11 +24,14 @@ def read_parquet(uri):
2424
files = ["gs://" + path for path in fs.glob(uri + "/part-*")]
2525
ds = parquet.ParquetDataset(files, filesystem=fs)
2626
return ds.read().to_pandas()
27-
elif parsed_uri.scheme == "s3":
27+
elif parsed_uri.scheme == "s3" or parsed_uri.scheme == "s3a":
28+
29+
s3uri = urlunparse(parsed_uri._replace(scheme="s3"))
30+
2831
import s3fs
2932

3033
fs = s3fs.S3FileSystem()
31-
files = ["s3://" + path for path in fs.glob(uri + "/part-*")]
34+
files = ["s3://" + path for path in fs.glob(s3uri + "/part-*")]
3235
ds = parquet.ParquetDataset(files, filesystem=fs)
3336
return ds.read().to_pandas()
3437
else:

0 commit comments

Comments
 (0)