Skip to content

Instantly share code, notes, and snippets.

Avatar
💻
Hacking away

Samuel Lampa samuell

💻
Hacking away
View GitHub Profile
View luigi_task_template_advanced.py
import luigi
import os
#from WorkflowUtils import WorkflowUtils
import logging as log
class ATask(luigi.Task):
# Task parameters
dataset_name = luigi.Parameter()
upstream_task = luigi.Parameter()
View advanced_luigi_workflow.py
'''Luigi Workflow
Usage:
run_luigi_workflow.py (--host=|--local-scheduler) --workers=<workers> run <taskname> -p <partitioning_parameter>
run_luigi_workflow.py (-h|--help)
run_luigi_workflow.py --version
Options:
-h, --help Show this screen
--version Show version
View 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
View execute_luigi_command.sh
python run_luigi_workflow.py --workers run some_task_c --partitioning-parameter Foo
View simple_luigi_workflow.py
# Some needed luigi imports
import luigi
import luigi.scheduler
import luigi.worker
# Here we are importing our own tasks, provided they are
# arranged in a python module (folder) named "components"
from components.SomeTaskA import SomeTaskA
from components.SomeTaskB import SomeTaskB
from components.SomeTaskC import SomeTaskC
@samuell
samuell / myconfig.cfg
Last active Aug 29, 2015
Query Semantic MediaWiki via MediaWiki REST API
View myconfig.cfg
[wiki]
user = MyWikiBotUser
pass = ********
View gc_rp_ptr_woderef.go
package main
import (
"bufio"
"fmt"
"log"
"os"
)
func main() {
@samuell
samuell / luigi_params_mixin.py
Created Jun 4, 2014
Snippet showing how to define (and pass on) workflow-wide parameters in the luigi workflow system. Courtesy of Joe Ennever ("stolen" from https://groups.google.com/forum/#!msg/luigi-user/7zIS_qseNOM/TWcYtRe5SVoJ )
View luigi_params_mixin.py
class FooParamsMixin(object):
param1 = luigi.Parameter()
param2 = luigi.Parameter()
...
def foo_params(self):
return { 'param1': self.param1, 'param2' : self.param2, ... }
class TaskA(FooParamsMixin, luigi.Task):
def requires(self):
View luigi_log_config_example.cfg
[formatters]
keys: default
[handlers]
keys: console, logfile
[loggers]
keys: root, luigi-interface
[formatter_default]
View luigi_multi_in_out.py
class TaskA1(luigi.Task):
def requires(self):
....
def output(self):
return { 'output1' : luigi.LocalTarget( self.input().path + ".output1" ),
'output2' : luigi.LocalTarget( self.input().path + ".output2" ) }
def run(self):
....
class TaskA2(luigi.Task):
You can’t perform that action at this time.