Skip to content

Commit d5ca4ea

Browse files
oavdeevoavdeev
committed
EMR launcher and configuration
Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
1 parent 39c9fbe commit d5ca4ea

11 files changed

Lines changed: 752 additions & 9 deletions

File tree

‎sdk/python/feast/cli.py‎

Lines changed: 39 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -364,11 +364,13 @@ def sync_offline_to_online(feature_table: str, start_time: str, end_time: str):
364364
"""
365365
Sync offline store to online.
366366
"""
367-
import feast.pyspark.aws.jobs
367+
from datetime import datetime
368368

369369
client = Client()
370370
table = client.get_feature_table(feature_table)
371-
feast.pyspark.aws.jobs.sync_offline_to_online(client, table, start_time, end_time)
371+
client.start_offline_to_online_ingestion(
372+
table, datetime.fromisoformat(start_time), datetime.fromisoformat(end_time)
373+
)
372374

373375

374376
@cli.command()
@@ -424,5 +426,40 @@ def list_emr_jobs():
424426
)
425427

426428

429+
@cli.command()
430+
@click.option(
431+
"--features",
432+
"-f",
433+
help="Features in feature_table:feature format, comma separated",
434+
required=True,
435+
)
436+
@click.option(
437+
"--entity-df-path",
438+
"-e",
439+
help="Path to entity df in CSV format. It is assumed to have event_timestamp column and a header.",
440+
required=True,
441+
)
442+
@click.option("--destination", "-d", help="Destination", default="")
443+
def get_historical_features(features: str, entity_df_path: str, destination: str):
444+
"""
445+
Get historical features
446+
"""
447+
import pandas
448+
449+
client = Client()
450+
451+
# TODO: clean this up
452+
entity_df = pandas.read_csv(entity_df_path, sep=None, engine="python",)
453+
454+
entity_df["event_timestamp"] = pandas.to_datetime(entity_df["event_timestamp"])
455+
456+
uploaded_df = client.stage_dataframe(
457+
entity_df, "event_timestamp", "created_timestamp"
458+
)
459+
460+
job = client.get_historical_features(features.split(","), uploaded_df,)
461+
print(job.get_output_file_uri())
462+
463+
427464
if __name__ == "__main__":
428465
cli()

‎sdk/python/feast/client.py‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@
7676
from feast.online_response import OnlineResponse, _infer_online_entity_rows
7777
from feast.pyspark.abc import RetrievalJob, SparkJob
7878
from feast.pyspark.launcher import (
79+
stage_dataframe,
7980
start_historical_feature_retrieval_job,
8081
start_historical_feature_retrieval_spark_session,
8182
start_offline_to_online_ingestion,
@@ -885,9 +886,16 @@ def _get_feature_tables_from_feature_refs(
885886
return feature_tables
886887

887888
def start_offline_to_online_ingestion(
888-
self,
889-
feature_table: Union[FeatureTable, str],
890-
start: Union[datetime, str],
891-
end: Union[datetime, str],
889+
self, feature_table: Union[FeatureTable, str], start: datetime, end: datetime,
892890
) -> SparkJob:
893891
return start_offline_to_online_ingestion(feature_table, start, end, self) # type: ignore
892+
893+
def stage_dataframe(
894+
self,
895+
df: pd.DataFrame,
896+
event_timestamp_column: str,
897+
created_timestamp_column: str,
898+
) -> FileSource:
899+
return stage_dataframe(
900+
df, event_timestamp_column, created_timestamp_column, self
901+
)

‎sdk/python/feast/constants.py‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,16 @@ class AuthProvider(Enum):
8080
CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_FORMAT = "historical_feature_output_format"
8181
CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_LOCATION = "historical_feature_output_location"
8282

83+
CONFIG_REDIS_HOST = "redis_host"
84+
CONFIG_REDIS_PORT = "redis_port"
85+
CONFIG_REDIS_SSL = "redis_ssl"
86+
87+
CONFIG_SPARK_EMR_REGION = "emr_region"
88+
CONFIG_SPARK_EMR_CLUSTER_ID = "emr_cluster_id"
89+
CONFIG_SPARK_EMR_CLUSTER_TEMPLATE_PATH = "emr_cluster_template_path"
90+
CONFIG_SPARK_EMR_STAGING_LOCATION = "emr_staging_location"
91+
CONFIG_SPARK_EMR_LOG_LOCATION = "emr_log_location"
92+
8393

8494
# Configuration option default values
8595
FEAST_DEFAULT_OPTIONS = {

‎sdk/python/feast/pyspark/abc.py‎

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,10 @@
55
from enum import Enum
66
from typing import Dict, List, Optional
77

8+
import pandas
9+
10+
from feast.data_source import FileSource
11+
812

913
class SparkJobFailure(Exception):
1014
"""
@@ -258,19 +262,31 @@ def __init__(
258262
start: datetime,
259263
end: datetime,
260264
jar: str,
265+
redis_host: str,
266+
redis_port: int,
267+
redis_ssl: bool,
261268
):
262269
self._feature_table = feature_table
263270
self._source = source
264271
self._start = start
265272
self._end = end
266273
self._jar = jar
274+
self._redis_host = redis_host
275+
self._redis_port = redis_port
276+
self._redis_ssl = redis_ssl
267277

268278
def get_name(self) -> str:
269279
return (
270-
f"BatchIngestion-{self._feature_table['name']}-"
280+
f"BatchIngestion-{self.get_feature_table_name()}-"
271281
f"{self._start.strftime('%Y-%m-%d')}-{self._end.strftime('%Y-%m-%d')}"
272282
)
273283

284+
def _get_redis_config(self):
285+
return dict(host=self._redis_host, port=self._redis_port, ssl=self._redis_ssl)
286+
287+
def get_feature_table_name(self) -> str:
288+
return self._feature_table["name"]
289+
274290
def get_main_file_path(self) -> str:
275291
return self._jar
276292

@@ -289,6 +305,8 @@ def get_arguments(self) -> List[str]:
289305
self._start.strftime("%Y-%m-%dT%H:%M:%S"),
290306
"--end",
291307
self._end.strftime("%Y-%m-%dT%H:%M:%S"),
308+
"--redis",
309+
json.dumps(self._get_redis_config()),
292310
]
293311

294312

@@ -334,3 +352,18 @@ def offline_to_online_ingestion(
334352
IngestionJob: wrapper around remote job that can be used to check when job completed.
335353
"""
336354
raise NotImplementedError
355+
356+
@abc.abstractmethod
357+
def stage_dataframe(
358+
self,
359+
df: pandas.DataFrame,
360+
event_timestamp_column: str,
361+
created_timestamp_column: str,
362+
) -> FileSource:
363+
"""
364+
Upload a pandas dataframe so it is available to the Spark cluster.
365+
366+
Returns:
367+
FileSource: representing the uploaded dataframe.
368+
"""
369+
raise NotImplementedError

‎sdk/python/feast/pyspark/historical_feature_retrieval_job.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -773,7 +773,7 @@ def _feature_table_from_dict(dct: Dict[str, Any]) -> FeatureTable:
773773
spark = SparkSession.builder.getOrCreate()
774774
args = _get_args()
775775
feature_tables_conf = json.loads(args.feature_tables)
776-
feature_tables_sources_conf = json.loads(args.feature_tables_source)
776+
feature_tables_sources_conf = json.loads(args.feature_tables_sources)
777777
entity_source_conf = json.loads(args.entity_source)
778778
destination_conf = json.loads(args.destination)
779779
start_job(

‎sdk/python/feast/pyspark/launcher.py‎

Lines changed: 41 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,18 @@
66

77
from feast.config import Config
88
from feast.constants import (
9+
CONFIG_REDIS_HOST,
10+
CONFIG_REDIS_PORT,
11+
CONFIG_REDIS_SSL,
912
CONFIG_SPARK_DATAPROC_CLUSTER_NAME,
1013
CONFIG_SPARK_DATAPROC_PROJECT,
1114
CONFIG_SPARK_DATAPROC_REGION,
1215
CONFIG_SPARK_DATAPROC_STAGING_LOCATION,
16+
CONFIG_SPARK_EMR_CLUSTER_ID,
17+
CONFIG_SPARK_EMR_CLUSTER_TEMPLATE_PATH,
18+
CONFIG_SPARK_EMR_LOG_LOCATION,
19+
CONFIG_SPARK_EMR_REGION,
20+
CONFIG_SPARK_EMR_STAGING_LOCATION,
1321
CONFIG_SPARK_HOME,
1422
CONFIG_SPARK_INGESTION_JOB_JAR,
1523
CONFIG_SPARK_LAUNCHER,
@@ -50,7 +58,27 @@ def _dataproc_launcher(config: Config) -> JobLauncher:
5058
)
5159

5260

53-
_launchers = {"standalone": _standalone_launcher, "dataproc": _dataproc_launcher}
61+
def _emr_launcher(config: Config) -> JobLauncher:
62+
from feast.pyspark.launchers import aws
63+
64+
def _get_optional(option):
65+
if config.exists(option):
66+
return config.get(option)
67+
68+
return aws.EmrClusterLauncher(
69+
region=config.get(CONFIG_SPARK_EMR_REGION),
70+
existing_cluster_id=_get_optional(CONFIG_SPARK_EMR_CLUSTER_ID),
71+
new_cluster_template_path=_get_optional(CONFIG_SPARK_EMR_CLUSTER_TEMPLATE_PATH),
72+
staging_location=config.get(CONFIG_SPARK_EMR_STAGING_LOCATION),
73+
emr_log_location=config.get(CONFIG_SPARK_EMR_LOG_LOCATION),
74+
)
75+
76+
77+
_launchers = {
78+
"standalone": _standalone_launcher,
79+
"dataproc": _dataproc_launcher,
80+
"emr": _emr_launcher,
81+
}
5482

5583

5684
def resolve_launcher(config: Config) -> JobLauncher:
@@ -177,5 +205,17 @@ def start_offline_to_online_ingestion(
177205
feature_table=_feature_table_to_argument(client, feature_table),
178206
start=start,
179207
end=end,
208+
redis_host=client._config.get(CONFIG_REDIS_HOST),
209+
redis_port=client._config.getint(CONFIG_REDIS_PORT),
210+
redis_ssl=client._config.getboolean(CONFIG_REDIS_SSL),
180211
)
181212
)
213+
214+
215+
def stage_dataframe(
216+
df, event_timestamp_column: str, created_timestamp_column: str, client: "Client"
217+
) -> FileSource:
218+
launcher = resolve_launcher(client._config)
219+
return launcher.stage_dataframe(
220+
df, event_timestamp_column, created_timestamp_column,
221+
)

‎sdk/python/feast/pyspark/launchers/aws/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
from .emr import EmrClusterLauncher, EmrIngestionJob, EmrRetrievalJob
2+
3+
__all__ = ["EmrRetrievalJob", "EmrIngestionJob", "EmrClusterLauncher"]

0 commit comments

Comments
 (0)