example_luigi_task.py

import luigi
from SomePossibleUpstreamTask import SomePossibleUpstreamTask
from SomeOtherPossibleUpstreamTask import SomeOtherPossibleUpstreamTask

class ExampleTask(luigi.Task):
    upstream_task = luigi.Parameter()
    
    def requires(self):
        # This will return the upstream task object
        return self.upstream_task

    def output(self):
        # A simple file naming scheme where the input file name is used, 
        # and just combined with an additional extension.
        return luigi.LocalTarget(self.input().path + ".newextension")

    def run(self):
        # Loop over the input file, do something to each line, and 
        # write to the output file.
        with self.input().open() as infile, self.output().open() as outfile:
            for line in infile:
                outfile.writeline(do_something(line))

    def do_something(self, line):
        pass # TODO: implement function