Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
import luigi
class MyFirstTask(luigi.Task): # inherit from Luigi base class Task
def run(self):
with self.output().open('w') as f:
f.write("This is my first task.\n")
def output(self):
return luigi.LocalTarget('MyTask.txt')
class MySecondTask(luigi.Task):
# MyFirstTask() needs to run first
def requires(self):
return MyFirstTask()
def run(self):
# here we're going to count the words in the first file
# and output it to a second file
numwords = 0
# once MyFirstTask() has run, this file has the info we need
# it could just as easily be database output in here
for line in f:
words = line.split(None) # this splits on whitespace of any length
numwords += len(words)
# print our results
with open('words.txt', 'w') as out:
out.write(str(numwords)+"\n")
def output(self):
return luigi.LocalTarget('words.txt')
if __name__ == '__main__':
# since we are setting MySecondTask to be the main task,
# it will check for the requirements first, then run
luigi.run(["--local-scheduler"], main_task_cls=MySecondTask)

dfernan commented Apr 6, 2017

Hi, I was trying your demo, and it seems there're some mistakes in the code.
Wrong Indentiation:
# it could just as easily be database output in here
for line in f:
words = line.split(None) # this splits on whitespace of any length
numwords += len(words)
Also, it seems f is not defined in MySecondTask, shouldn't you define it somewhere?
Let me know if you can advise me here. Thanks!

mfharry commented May 2, 2017

If you fix the indentation on lines 23-25 and change
for line in f:
to
for line in self.input().open('r'):

The script should then run.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment