Skip to content

Refactor Spark Job launcher API #1060

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 2 commits into from
Oct 15, 2020
Merged

Refactor Spark Job launcher API #1060

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
68272a0
simplify launcher API
pyalex Oct 15, 2020
cedc246
uniq destination property with others
pyalex Oct 15, 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
238 changes: 119 additions & 119 deletions sdk/python/feast/pyspark/abc.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,8 +86,113 @@ def __init__(
feature_tables_sources: List[Dict],
entity_source: Dict,
destination: Dict,
**kwargs,
):
"""
Args:
entity_source (Dict): Entity data source configuration.
feature_tables_sources (List[Dict]): List of feature tables data sources configurations.
feature_tables (List[Dict]): List of feature table specification.
The order of the feature table must correspond to that of feature_tables_sources.
destination (Dict): Retrieval job output destination.

Examples:
>>> # Entity source from file
>>> entity_source = {
"file": {
"format": "parquet",
"path": "gs://some-gcs-bucket/customer",
"event_timestamp_column": "event_timestamp",
"options": {
"mergeSchema": "true"
} # Optional. Options to be passed to Spark while reading the dataframe from source.
"field_mapping": {
"id": "customer_id"
} # Optional. Map the columns, where the key is the original column name and the value is the new column name.

}
}

>>> # Entity source from BigQuery
>>> entity_source = {
"bq": {
"project": "gcp_project_id",
"dataset": "bq_dataset",
"table": "customer",
"event_timestamp_column": "event_timestamp",
}
}

>>> feature_tables_sources = [
{
"bq": {
"project": "gcp_project_id",
"dataset": "bq_dataset",
"table": "customer_transactions",
"event_timestamp_column": "event_timestamp",
"created_timestamp_column": "created_timestamp" # This field is mandatory for feature tables.
}
},

{
"file": {
"format": "parquet",
"path": "gs://some-gcs-bucket/customer_profile",
"event_timestamp_column": "event_timestamp",
"created_timestamp_column": "created_timestamp",
"options": {
"mergeSchema": "true"
}
}
},
]


>>> feature_tables = [
{
"name": "customer_transactions",
"entities": [
{
"name": "customer
"type": "int32"
}
],
"features": [
{
"name": "total_transactions"
"type": "double"
},
{
"name": "total_discounts"
"type": "double"
}
],
"max_age": 86400 # In seconds.
},

{
"name": "customer_profile",
"entities": [
{
"name": "customer
"type": "int32"
}
],
"features": [
{
"name": "is_vip"
"type": "bool"
}
],

}
]

>>> destination = {
"format": "parquet",
"path": "gs://some-gcs-bucket/retrieval_output"
}

"""
self._feature_tables = feature_tables
self._feature_tables_sources = feature_tables_sources
self._entity_source = entity_source
Expand All @@ -114,6 +219,9 @@ def get_arguments(self) -> List[str]:
json.dumps(self._destination),
]

def get_destination_path(self) -> str:
return self._destination["path"]
Comment thread
pyalex marked this conversation as resolved.


class RetrievalJob(SparkJob):
"""
Expand Down Expand Up @@ -150,7 +258,6 @@ def __init__(
start: datetime,
end: datetime,
jar: str,
**kwargs,
):
self._feature_table = feature_table
self._source = source
Expand Down Expand Up @@ -198,139 +305,32 @@ class JobLauncher(abc.ABC):

@abc.abstractmethod
def historical_feature_retrieval(
self,
entity_source_conf: Dict,
feature_tables_sources_conf: List[Dict],
feature_tables_conf: List[Dict],
destination_conf: Dict,
**kwargs,
self, retrieval_job_params: RetrievalJobParameters
) -> RetrievalJob:
"""
Submits a historical feature retrieval job to a Spark cluster.

Args:
entity_source_conf (Dict): Entity data source configuration.
feature_tables_sources_conf (List[Dict]): List of feature tables data sources configurations.
feature_tables_conf (List[Dict]): List of feature table specification.
The order of the feature table must correspond to that of feature_tables_sources.
destination_conf (Dict): Retrieval job output destination.

Raises:
SparkJobFailure: The spark job submission failed, encountered error
during execution, or timeout.

Examples:
>>> # Entity source from file
>>> entity_source_conf = {
"file": {
"format": "parquet",
"path": "gs://some-gcs-bucket/customer",
"event_timestamp_column": "event_timestamp",
"options": {
"mergeSchema": "true"
} # Optional. Options to be passed to Spark while reading the dataframe from source.
"field_mapping": {
"id": "customer_id"
} # Optional. Map the columns, where the key is the original column name and the value is the new column name.

}
}

>>> # Entity source from BigQuery
>>> entity_source_conf = {
"bq": {
"project": "gcp_project_id",
"dataset": "bq_dataset",
"table": "customer",
"event_timestamp_column": "event_timestamp",
}
}

>>> feature_table_sources_conf = [
{
"bq": {
"project": "gcp_project_id",
"dataset": "bq_dataset",
"table": "customer_transactions",
"event_timestamp_column": "event_timestamp",
"created_timestamp_column": "created_timestamp" # This field is mandatory for feature tables.
}
},

{
"file": {
"format": "parquet",
"path": "gs://some-gcs-bucket/customer_profile",
"event_timestamp_column": "event_timestamp",
"created_timestamp_column": "created_timestamp",
"options": {
"mergeSchema": "true"
}
}
},
]


>>> feature_tables_conf = [
{
"name": "customer_transactions",
"entities": [
{
"name": "customer
"type": "int32"
}
],
"features": [
{
"name": "total_transactions"
"type": "double"
},
{
"name": "total_discounts"
"type": "double"
}
],
"max_age": 86400 # In seconds.
},

{
"name": "customer_profile",
"entities": [
{
"name": "customer
"type": "int32"
}
],
"features": [
{
"name": "is_vip"
"type": "bool"
}
],

}
]

>>> destination_conf = {
"format": "parquet",
"path": "gs://some-gcs-bucket/retrieval_output"
}

Returns:
str: file uri to the result file.
RetrievalJob: wrapper around remote job that returns file uri to the result file.
"""
raise NotImplementedError

@abc.abstractmethod
def offline_to_online_ingestion(
self,
jar_path: str,
source_conf: Dict,
feature_table_conf: Dict,
start: datetime,
end: datetime,
self, ingestion_job_params: IngestionJobParameters
) -> IngestionJob:
"""
Submits a batch ingestion job to a Spark cluster.

Raises:
SparkJobFailure: The spark job submission failed, encountered error
during execution, or timeout.

Returns:
IngestionJob: wrapper around remote job that can be used to check when job completed.
"""
raise NotImplementedError
42 changes: 26 additions & 16 deletions sdk/python/feast/pyspark/launcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,13 @@
)
from feast.data_source import BigQuerySource, DataSource, FileSource
from feast.feature_table import FeatureTable
from feast.pyspark.abc import IngestionJob, JobLauncher, RetrievalJob
from feast.pyspark.abc import (
IngestionJob,
IngestionJobParameters,
JobLauncher,
RetrievalJob,
RetrievalJobParameters,
)
from feast.staging.storage_client import get_staging_client
from feast.value_type import ValueType

Expand Down Expand Up @@ -129,16 +135,18 @@ def start_historical_feature_retrieval_job(
) -> RetrievalJob:
launcher = resolve_launcher(client._config)
return launcher.historical_feature_retrieval(
entity_source_conf=_source_to_argument(entity_source),
feature_tables_sources_conf=[
_source_to_argument(feature_table.batch_source)
for feature_table in feature_tables
],
feature_tables_conf=[
_feature_table_to_argument(client, feature_table)
for feature_table in feature_tables
],
destination_conf={"format": output_format, "path": output_path},
RetrievalJobParameters(
entity_source=_source_to_argument(entity_source),
feature_tables_sources=[
_source_to_argument(feature_table.batch_source)
for feature_table in feature_tables
],
feature_tables=[
_feature_table_to_argument(client, feature_table)
for feature_table in feature_tables
],
destination={"format": output_format, "path": output_path},
)
)


Expand All @@ -163,9 +171,11 @@ def start_offline_to_online_ingestion(
local_jar_path = _download_jar(client._config.get(CONFIG_SPARK_INGESTION_JOB_JAR))

return launcher.offline_to_online_ingestion(
jar_path=local_jar_path,
source_conf=_source_to_argument(feature_table.batch_source),
feature_table_conf=_feature_table_to_argument(client, feature_table),
start=start,
end=end,
IngestionJobParameters(
jar=local_jar_path,
source=_source_to_argument(feature_table.batch_source),
feature_table=_feature_table_to_argument(client, feature_table),
start=start,
end=end,
)
)
33 changes: 4 additions & 29 deletions sdk/python/feast/pyspark/launchers/gcloud/dataproc.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import os
import uuid
from datetime import datetime
from typing import Dict, List, cast
from typing import cast
from urllib.parse import urlparse

from google.api_core.operation import Operation
Expand Down Expand Up @@ -144,37 +143,13 @@ def dataproc_submit(self, job_params: SparkJobParameters) -> Operation:
)

def historical_feature_retrieval(
self,
entity_source_conf: Dict,
feature_tables_sources_conf: List[Dict],
feature_tables_conf: List[Dict],
destination_conf: Dict,
**kwargs,
self, job_params: RetrievalJobParameters
) -> RetrievalJob:
job_params = RetrievalJobParameters(
feature_tables=feature_tables_conf,
feature_tables_sources=feature_tables_sources_conf,
entity_source=entity_source_conf,
destination=destination_conf,
)

return DataprocRetrievalJob(
self.dataproc_submit(job_params), destination_conf["path"]
self.dataproc_submit(job_params), job_params.get_destination_path()
)

def offline_to_online_ingestion(
self,
jar_path: str,
source_conf: Dict,
feature_table_conf: Dict,
start: datetime,
end: datetime,
self, job_params: IngestionJobParameters
) -> IngestionJob:
job_params = IngestionJobParameters(
feature_table=feature_table_conf,
source=source_conf,
start=start,
end=end,
jar=jar_path,
)
return DataprocIngestionJob(self.dataproc_submit(job_params))
Loading