我有一个 python 函数,可以将数据从其他 2 个表加载到 sql server 表中。
def load_table(date1,date2):
strDate1 = date1.strftime('%m/%d/%Y')
strDate2 = date2.strftime('%m/%d/%Y')
stmt = "insert into Agent_Queue (ID) select distinct Send_Location_ID from Pretty_Txns where Send_Date >= '%s' and Send_Date <= '%s' and Send_Location_ID is not null union select distinct Pay_Location_ID from Pretty_Txns where Pay_Date >= '%s' and Pay_Date <= '%s' and Pay_Location_ID is not null" % (strDate1,strDate2,strDate1,strDate2)
cnx1= connection string
self.curs=cnx1.cursor()
self.curs.execute(stmt)
self.curs.commit()
我正在尝试将此功能转换为 luigi 任务
按照文档尝试了以下方法:
class Datetask(luigi.Task):
def output(self):
return luigi.DateParameter()
class loading(luigi.Task):
def requires(self):
return {'date1': DateTask(dt.date(2016,10,30)), 'date2': DateTask(dt.date(2016,11,29))}
def run(self):
date1 = dict['date1']
date2 = dict['date2']
strDate1 = date1.strftime('%m/%d/%Y')
strDate2 = date2.strftime('%m/%d/%Y')
stmt = "insert into Agent_Queue (ID) select distinct Send_Location_ID from Pretty_Txns where Send_Date >= '%s' and Send_Date <= '%s' and Send_Location_ID is not null union select distinct Pay_Location_ID from Pretty_Txns where Pay_Date >= '%s' and Pay_Date <= '%s' and Pay_Location_ID is not null" % (strDate1,strDate2,strDate1,strDate2)
curs=cnx1.cursor()
curs.execute(stmt)
curs.commit()
curs.close()
当我尝试运行它时,我收到错误:
python -m luigi --module load_task loading --local-scheduler
DEBUG: Checking if loading() is complete
/usr/local/lib/python2.7/dist-packages/luigi/worker.py:305: UserWarning: Task loading() without outputs has no custom complete() method
is_complete = task.complete()
WARNING: Will not run loading() or any dependencies due to error in deps() method:
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/luigi/worker.py", line 697, in _add
deps = task.deps()
File "/usr/local/lib/python2.7/dist-packages/luigi/task.py", line 572, in deps
return flatten(self._requires())
File "/usr/local/lib/python2.7/dist-packages/luigi/task.py", line 544, in _requires
return flatten(self.requires()) # base impl
File "load_task.py", line 19, in requires
return {'date1': DateTask(dt.date(2016,10,30)), 'date2': DateTask(dt.date(2016,11,29))}
NameError: global name 'DateTask' is not defined
我正在定义DateTask所以这个错误让我很困惑。
此外,是否所有任务都需要全部 3 个requires()、run、output?
另外,是否有必要始终将输出写入文件?在使用 luIgi 时是全新的,因此将不胜感激任何输入