Parametrización (2)#

Última modificación: Mayo 13, 2022 | YouTube

[1]:

%cd /tmp
!rm -rf output/ all_sales.csv
!mkdir -p input/

/tmp

[2]:

%%writefile input/sales1.csv
Fance,May,100
NA
Germany,May,200

Overwriting input/sales1.csv

[3]:

%%writefile input/sales2.csv
Fance,June,140
Germany,June,180
UK,June,180

Overwriting input/sales2.csv

[4]:

%%writefile input/sales3.csv
France,June,240
None
Germany,June,180

Overwriting input/sales3.csv

[5]:

%%writefile pipeline.py

import os

import luigi
from luigi import IntParameter, LocalTarget, Parameter, Task

INPUT_FOLDER = "input"
OUTPUT_FOLDER = "output"


class DownloadFile(Task):

    file_name = Parameter()
    index = IntParameter()

    def output(self):
        output_path = os.path.join(
            OUTPUT_FOLDER,
            str(self.index),
            self.file_name,
        )
        return LocalTarget(output_path)

    def run(self):
        input_path = os.path.join(INPUT_FOLDER, self.file_name)
        with open(input_path) as in_file:
            with self.output().open("w") as out_file:
                for line in in_file:
                    if "," in line:
                        out_file.write(line)


class DownloadSalesData(Task):
    def output(self):
        return LocalTarget("all_sales.csv")

    def run(self):
        processed_files = []
        counter = 1
        for file in sorted(os.listdir(INPUT_FOLDER)):
            target = yield DownloadFile(file, counter)
            counter += 1
            processed_files.append(target)

        with self.output().open("w") as out_file:
            for file in processed_files:
                with file.open() as in_file:
                    for line in in_file:
                        out_file.write(line)


if __name__ == "__main__":
    luigi.run(["DownloadSalesData", "--local-scheduler"])

Overwriting pipeline.py

[6]:

!python3 pipeline.py

DEBUG: Checking if DownloadSalesData() is complete
INFO: Informed scheduler that task   DownloadSalesData__99914b932b   has status   PENDING
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) running   DownloadSalesData()
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) new requirements      DownloadSalesData()
DEBUG: 1 running tasks, waiting for next task to finish
DEBUG: Checking if DownloadFile(file_name=sales3.csv, index=1) is complete
INFO: Informed scheduler that task   DownloadFile_sales3_csv_1_3c9fa6db4d   has status   PENDING
INFO: Informed scheduler that task   DownloadSalesData__99914b932b   has status   PENDING
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) running   DownloadFile(file_name=sales3.csv, index=1)
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) done      DownloadFile(file_name=sales3.csv, index=1)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   DownloadFile_sales3_csv_1_3c9fa6db4d   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) running   DownloadSalesData()
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) new requirements      DownloadSalesData()
DEBUG: 1 running tasks, waiting for next task to finish
DEBUG: Checking if DownloadFile(file_name=sales2.csv, index=2) is complete
INFO: Informed scheduler that task   DownloadFile_sales2_csv_2_a64b62af8e   has status   PENDING
INFO: Informed scheduler that task   DownloadSalesData__99914b932b   has status   PENDING
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) running   DownloadFile(file_name=sales2.csv, index=2)
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) done      DownloadFile(file_name=sales2.csv, index=2)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   DownloadFile_sales2_csv_2_a64b62af8e   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) running   DownloadSalesData()
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) new requirements      DownloadSalesData()
DEBUG: 1 running tasks, waiting for next task to finish
DEBUG: Checking if DownloadFile(file_name=sales1.csv, index=3) is complete
INFO: Informed scheduler that task   DownloadFile_sales1_csv_3_8ffb46dea7   has status   PENDING
INFO: Informed scheduler that task   DownloadSalesData__99914b932b   has status   PENDING
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) running   DownloadFile(file_name=sales1.csv, index=3)
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) done      DownloadFile(file_name=sales1.csv, index=3)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   DownloadFile_sales1_csv_3_8ffb46dea7   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) running   DownloadSalesData()
INFO: [pid 475] Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) done      DownloadSalesData()
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   DownloadSalesData__99914b932b   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Done
DEBUG: There are no more tasks to run at this time
INFO: Worker Worker(salt=466330766, workers=1, host=b148b7d2dd75, username=root, pid=475) was stopped. Shutting down Keep-Alive thread
INFO:
===== Luigi Execution Summary =====

Scheduled 4 tasks of which:
* 4 ran successfully:
    - 3 DownloadFile(file_name=sales1.csv, index=3) ...
    - 1 DownloadSalesData()

This progress looks :) because there were no failed tasks or missing dependencies

===== Luigi Execution Summary =====

[7]:

!cat all_sales.csv

France,June,240
Germany,June,180
Fance,June,140
Germany,June,180
UK,June,180
Fance,May,100
Germany,May,200

[8]:

!ls -1 /tmp/output

1
2
3

[9]:

!ls -1 /tmp/output/1/

sales3.csv

[10]:

!ls -1 /tmp/output/2/

sales2.csv

[11]:

!ls -1 /tmp/output/3/

sales1.csv

[12]:

!cat /tmp/output/1/sales3.csv

France,June,240
Germany,June,180