Sx 中的 Oberorite Lougi

Question

我有一项例行任务是上传和共享 S3 存储桶上的转储。虽然下面的代码有效，但出于某种原因它不想覆盖文件。

根据文档，我需要 1）为两个并行执行定义解决方案： path = luigi.Parameter(default=glob(DATA_DIR)[-2], batch_method=max)

2) 添加resources = {'overwrite_resource': 1}

虽然它适用于本地文件 - 但不适用于 S3。

class report_to_S3(luigi.Task):
    client = S3Client()
    path = luigi.Parameter(default=glob(DATA_DIR)[-2], batch_method=max)
    local_dump_path = '../../../data/local_db_dump.csv'
    resources = {'overwrite_resource': 1}

    def requires(self):
        return upload_tweets(path=self.path)

    def output(self):
        self.s3_path = "s3://qclm-nyc-ct/dump/dump.csv"
        return S3Target(self.s3_path, client=self.client)

    def run(self):
        c = sqa.create_engine('postgresql:///qc_soc_media')
        df = pd.read_sql_query('SELECT id, user_id, timestamp, lat, lon, ct FROM tweets WHERE ct IS NOT NULL', c)
        N = len(df)
        df.to_csv(self.local_dump_path, index=None)
        self.client.put(self.local_dump_path, self.output().path,
                        headers={'Content-Type': 'application/csv'})
        send_S3_report(N)


if __name__ == '__main__':
    luigi.run(local_scheduler=True, main_task_cls=report_to_S3)

Answer 1

如果output()方法中指定的目标已经存在，运行()方法将不会执行。您可能希望将时间戳添加到文件名中，或者创建另一个 sentinel/flag 来指示工作已完成。

Sx 中的 Oberorite Lougi

Luigi overwrite to S3

amazon-s3

luigi