Skip to content

Optional IngestionJob parameters passed by Spark Launcher #1130

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

Merged
pyalex merged 22 commits into from
Nov 4, 2020
Merged

Optional IngestionJob parameters passed by Spark Launcher #1130

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
6c7a404
optional job parameters
pyalex Nov 4, 2020
15b750b
default values
pyalex Nov 4, 2020
ba1481e
default values
pyalex Nov 4, 2020
5c2e162
add job service to docker compose
pyalex Nov 4, 2020
f93289b
remove jobcontroller from makefile
pyalex Nov 4, 2020
6ef85a7
pass project to job
pyalex Nov 4, 2020
ecb1337
helm update: jobservice secrets
pyalex Nov 4, 2020
3f00e19
run feast cli directly
pyalex Nov 4, 2020
a87e4ce
fix jobservice host name
pyalex Nov 4, 2020
e26a167
statsd enabled
pyalex Nov 4, 2020
e0d3e54
shared directory
pyalex Nov 4, 2020
4ca5beb
override staging path in local env
pyalex Nov 4, 2020
3aa6e03
override staging path in local env
pyalex Nov 4, 2020
b6d20b3
debug
pyalex Nov 4, 2020
ef24730
use root
pyalex Nov 4, 2020
2f694ee
cleanup
pyalex Nov 4, 2020
d985845
ingestion & jobcontroller module deleted from pom
pyalex Nov 4, 2020
a6a1894
ingestion & jobcontroller module deleted from pom
pyalex Nov 4, 2020
fe14419
--with-job-service
pyalex Nov 4, 2020
a375d8e
make feast_jobservice fixture scope - session
pyalex Nov 4, 2020
934e359
local jar path
pyalex Nov 4, 2020
e5903d0
stage only local files
pyalex Nov 4, 2020
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
Prev Previous commit
Next Next commit
--with-job-service 
Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
  • Loading branch information
@pyalex
pyalex committed Nov 4, 2020
commit fe14419bd65f31dfaf6245a6e8d86ead50382912
1 change: 1 addition & 0 deletions infra/scripts/test-end-to-end-gcp.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,5 +14,6 @@ su -p postgres -c "PATH=$PATH HOME=/tmp pytest -v tests/e2e/ \
--feast-version develop --env=gcloud --dataproc-cluster-name feast-e2e \
--dataproc-project kf-feast --dataproc-region us-central1 \
--staging-path gs://feast-templocation-kf-feast/ \
--with-job-service \
--redis-url 10.128.0.105:6379 --redis-cluster --kafka-brokers 10.128.0.103:9094 \
--bq-project kf-feast"
16 changes: 7 additions & 9 deletions sdk/python/feast/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,7 +387,7 @@ def version(self):
return result

@property
def project(self) -> Union[str, None]:
def project(self) -> str:
"""
Retrieve currently active project

Expand Down Expand Up @@ -1003,7 +1003,7 @@ def get_historical_features(
else:
return start_historical_feature_retrieval_job(
client=self,
project=self.project or FEAST_DEFAULT_OPTIONS[CONFIG_PROJECT_KEY],
project=self.project,
entity_source=entity_source,
feature_tables=feature_tables,
output_format=output_format,
Expand Down Expand Up @@ -1050,7 +1050,7 @@ def get_historical_features_df(
)
return start_historical_feature_retrieval_spark_session(
client=self,
project=self.project or FEAST_DEFAULT_OPTIONS[CONFIG_PROJECT_KEY],
project=self.project,
entity_source=entity_source,
feature_tables=feature_tables,
)
Expand Down Expand Up @@ -1090,15 +1090,14 @@ def start_offline_to_online_ingestion(
if not self._use_job_service:
return start_offline_to_online_ingestion(
client=self,
project=self.project or FEAST_DEFAULT_OPTIONS[CONFIG_PROJECT_KEY],
project=self.project,
feature_table=feature_table,
start=start,
end=end,
)
else:
request = StartOfflineToOnlineIngestionJobRequest(
project=self.project or FEAST_DEFAULT_OPTIONS[CONFIG_PROJECT_KEY],
table_name=feature_table.name,
project=self.project, table_name=feature_table.name,
)
request.start_date.FromDatetime(start)
request.end_date.FromDatetime(end)
Expand All @@ -1113,14 +1112,13 @@ def start_stream_to_online_ingestion(
if not self._use_job_service:
return start_stream_to_online_ingestion(
client=self,
project=self.project or FEAST_DEFAULT_OPTIONS[CONFIG_PROJECT_KEY],
project=self.project,
feature_table=feature_table,
extra_jars=extra_jars or [],
)
else:
request = StartStreamToOnlineIngestionJobRequest(
project=self.project or FEAST_DEFAULT_OPTIONS[CONFIG_PROJECT_KEY],
table_name=feature_table.name,
project=self.project, table_name=feature_table.name,
)
response = self._job_service.StartStreamToOnlineIngestionJob(request)
return RemoteStreamIngestionJob(
Expand Down
1 change: 1 addition & 0 deletions tests/e2e/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ def pytest_addoption(parser):
parser.addoption("--kafka-brokers", action="store", default="localhost:9092")

parser.addoption("--env", action="store", help="local|aws|gcloud", default="local")
parser.addoption("--with-job-service", action="store_true")
parser.addoption("--staging-path", action="store")
parser.addoption("--dataproc-cluster-name", action="store")
parser.addoption("--dataproc-region", action="store")
Expand Down
5 changes: 2 additions & 3 deletions tests/e2e/fixtures/feast_services.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,17 +144,16 @@ def feast_serving(
process.terminate()


@pytest.fixture(params=["jobservice_disabled", "jobservice_enabled"])
@pytest.fixture(scope="session")
def feast_jobservice(
request,
pytestconfig,
ingestion_job_jar,
redis_server: RedisExecutor,
feast_core: Tuple[str, int],
feast_serving: Tuple[str, int],
local_staging_path,
):
if request.param == "jobservice_disabled":
if not pytestconfig.getoption("with_job_service"):
yield None
else:
env = os.environ.copy()
Expand Down