Skip to content

#192: (WIP: Needs Work) Replace PythonDataflowTask With BeamDataflowJobTask #199

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

Open
rclough wants to merge 5 commits into
base: master
Choose a base branch
from
Open

#192: (WIP: Needs Work) Replace PythonDataflowTask With BeamDataflowJobTask #199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
cdde56b
#192: Replace PythonDataflowTask With BeamDataflowTask
rclough Jun 18, 2019
e43d2e2
#192: Fix spelling mistake
rclough Jun 18, 2019
232f649
#192: Add python script/dataflow executable info to TFTransformTask
rclough Jun 18, 2019
6449ee1
#192: Update luigi dependency to version that supports the new beam job
rclough Jun 21, 2019
7bb29e7
#192: move beam script references up one level of inheritance
rclough Jun 21, 2019
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
2 changes: 1 addition & 1 deletion setup.cfg
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ packages =

[extras]
luigi =
luigi>=2.3.3,<3.0.0
luigi>=2.8.5,<3.0.0
examples =
scipy
sklearn
Expand Down
297 changes: 0 additions & 297 deletions spotify_tensorflow/luigi/python_dataflow_task.py
View file
Open in desktop

This file was deleted.

12 changes: 10 additions & 2 deletions spotify_tensorflow/luigi/tfx_task.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,19 @@

from abc import abstractmethod

from spotify_tensorflow.luigi.python_dataflow_task import PythonDataflowTask
from luigi.contrib.beam_dataflow import BeamDataflowJobTask


class TFXBaseTask(PythonDataflowTask):
class TFXBaseTask(BeamDataflowJobTask):
python_script = None # type: str

def __init__(self, *args, **kwargs):
super(TFXBaseTask, self).__init__(*args, **kwargs)

def dataflow_executable(self):
""" Must be overwritten from the BeamDataflowTask """
return ["python", self.python_script]

def tfx_args(self):
""" Extra arguments that will be passed to your tfx dataflow job.

Expand All @@ -44,6 +50,8 @@ def _mk_cmd_line(self):


class TFTransformTask(TFXBaseTask):
# Required dataflow arg

def __init__(self, *args, **kwargs):
super(TFTransformTask, self).__init__(*args, **kwargs)

Expand Down
Loading