Skip to content

Commit 5db62d3

Browse files
terryyylimterryyylim
authored
Cleanup CLI and Python dependencies (#1062)
* Cleanup cli and python dependencies Signed-off-by: Terence <terencelimxp@gmail.com> * Cleanup cli Signed-off-by: Terence <terencelimxp@gmail.com>
1 parent 69d3f47 commit 5db62d3

2 files changed

Lines changed: 51 additions & 22 deletions

File tree

‎sdk/python/feast/cli.py‎

Lines changed: 51 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323

2424
from feast.client import Client
2525
from feast.config import Config
26+
from feast.constants import CONFIG_SPARK_LAUNCHER
2627
from feast.entity import Entity
2728
from feast.feature_table import FeatureTable
2829
from feast.loaders.yaml import yaml_loader
@@ -351,18 +352,27 @@ def project_list():
351352
print(tabulate(table, headers=["NAME"], tablefmt="plain"))
352353

353354

354-
@cli.command()
355+
@cli.group(name="jobs")
356+
def job():
357+
"""
358+
Create and manage jobs
359+
"""
360+
pass
361+
362+
363+
@job.command(name="start-offline-to-online")
355364
@click.option(
356365
"--feature-table",
357366
"-t",
358-
help="Feature table name to ingest data into",
367+
help="Feature table name of data to be synced",
368+
type=click.STRING,
359369
required=True,
360370
)
361371
@click.option("--start-time", "-s", help="Interval start", required=True)
362372
@click.option("--end-time", "-e", help="Interval end", required=True)
363373
def sync_offline_to_online(feature_table: str, start_time: str, end_time: str):
364374
"""
365-
Sync offline store to online.
375+
Sync offline store data to online store
366376
"""
367377
from datetime import datetime
368378

@@ -373,56 +383,77 @@ def sync_offline_to_online(feature_table: str, start_time: str, end_time: str):
373383
)
374384

375385

376-
@cli.command()
386+
@job.command(name="start-stream-to-online")
377387
@click.option(
378388
"--feature-table",
379389
"-t",
380-
help="Feature table name to ingest data into",
390+
help="Feature table name of job to be started",
391+
type=click.STRING,
381392
required=True,
382393
)
383394
@click.option(
384-
"--jar", "-j", help="Feature table name to ingest data into", default="",
395+
"--jar",
396+
"-j",
397+
help="The file path to the uber jar for offline to online ingestion spark job",
398+
default="",
385399
)
386400
def start_stream_to_online(feature_table: str, jar: str):
387401
"""
388-
Start stream to online sync job.
402+
Start stream to online sync job
389403
"""
390404

391405
client = Client()
392406
table = client.get_feature_table(feature_table)
393407
client.start_stream_to_online_ingestion(table, [jar] if jar else [])
394408

395409

396-
@cli.command()
410+
@job.command(name="stop-stream-to-online")
397411
@click.option(
398412
"--feature-table",
399413
"-t",
400-
help="Feature table name to ingest data into",
414+
help="Feature table name of job to be stopped",
415+
type=click.STRING,
401416
required=True,
402417
)
403418
def stop_stream_to_online(feature_table: str):
404419
"""
405-
Start stream to online sync job.
420+
Stop stream to online sync job
406421
"""
407-
import feast.pyspark.aws.jobs
408422

409-
feast.pyspark.aws.jobs.stop_stream_to_online(feature_table)
423+
spark_launcher = Config().get(CONFIG_SPARK_LAUNCHER)
410424

425+
if spark_launcher == "emr":
426+
import feast.pyspark.aws.jobs
411427

412-
@cli.command()
413-
def list_emr_jobs():
428+
feast.pyspark.aws.jobs.stop_stream_to_online(feature_table)
429+
else:
430+
raise NotImplementedError(
431+
f"Feast currently does not provide support for the specified spark launcher: {spark_launcher}"
432+
)
433+
434+
435+
@job.command()
436+
def list_jobs():
414437
"""
415-
List jobs.
438+
List jobs
416439
"""
417440
from tabulate import tabulate
418441

419-
import feast.pyspark.aws.jobs
442+
spark_launcher = Config().get(CONFIG_SPARK_LAUNCHER)
420443

421-
jobs = feast.pyspark.aws.jobs.list_jobs(None, None)
444+
if spark_launcher == "emr":
445+
import feast.pyspark.aws.jobs
422446

423-
print(
424-
tabulate(jobs, headers=feast.pyspark.aws.jobs.JobInfo._fields, tablefmt="plain")
425-
)
447+
jobs = feast.pyspark.aws.jobs.list_jobs(None, None)
448+
print(
449+
tabulate(
450+
jobs, headers=feast.pyspark.aws.jobs.JobInfo._fields, tablefmt="plain"
451+
)
452+
)
453+
else:
454+
raise NotImplementedError(
455+
f"Feast currently does not provide support for the specified spark launcher: {spark_launcher}"
456+
)
426457

427458

428459
@cli.command()

‎sdk/python/setup.py‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,14 +39,12 @@
3939
"protobuf>=3.10",
4040
"PyYAML==5.1.*",
4141
"fastavro>=0.22.11,<0.23",
42-
"kafka-python==1.*",
4342
"tabulate==0.8.*",
4443
"toml==0.10.*",
4544
"tqdm==4.*",
4645
"pyarrow<0.16.0,>=0.15.1",
4746
"numpy",
4847
"google",
49-
"confluent_kafka",
5048
]
5149

5250
# README file from Feast repo root directory

0 commit comments

Comments
 (0)