Luigi在命令行中显示已完成任务的百分比
Luigi是一个Python编写的开源任务调度框架，用于构建复杂的数据管道和工作流。它提供了一种简单而强大的方式来定义和运行任务，并且可以在命令行中显示已完成任务的百分比。
Luigi的优势在于其易用性和可扩展性。它的任务调度功能使得并行处理任务变得更加简单，可以轻松地构建和管理复杂的数据管道。另外，Luigi还提供了丰富的工具和特性，如任务依赖关系的定义、任务状态的监控和报告、任务失败的重试等，使得开发人员能够更加高效地管理和调度任务。
应用场景方面，Luigi适用于各种需要进行任务调度和数据处理的场景。例如，在数据分析和机器学习领域，可以使用Luigi来构建数据管道和工作流，进行数据的抽取、转换、加载等操作。在大数据处理和数据工程领域，可以利用Luigi来进行任务的并行处理和调度，提高数据处理的效率和可靠性。
腾讯云相关产品中，与Luigi相匹配的是腾讯云的数据仓库产品TencentDB for PostgreSQL。TencentDB for PostgreSQL是一种高度可扩展的云原生关系型数据库，支持在腾讯云上构建和管理大规模数据仓库。它提供了强大的数据管理和查询功能，与Luigi搭配使用可以实现数据的高效处理和存储。
更多关于TencentDB for PostgreSQL的信息和产品介绍，请访问以下链接： https://cloud.tencent.com/product/tcdb-postgres
需要注意的是，本回答仅提供了一个腾讯云的产品作为示例，其他云计算品牌商也可能提供类似的产品和服务，读者可以根据实际需求选择合适的产品和解决方案。
相关·内容
在工作流管理系统中，当您有大量任务时，通常需要查看已完成任务的数量(或百分比)。 其他wfms系统，如Snakemake，在默认情况下会这样做。不过，我在Luigi的文档中找不到类似的功能。 我想在命令行上查看百分比信息。 有没有办法做到这一点？ 提前感谢！
浏览 30提问于2020-07-24得票数 1
1回答
我是Luigi的新手，我已经创建了一个管道，它从数据库中获取数据，转换数据，然后将其加载回数据库。我在其中创建了四个任务。但是，当我在cmd或Pycharm上执行任务时，它显示不能调度非任务。下面是我的流水线的伪代码。每个任务的参数不是输入，而是从其他文件中获取。 class Task1(luigi.Task):
# Some Parameters
def get_target():
def query():
def run():
class Task2(luigi.Task):
浏览 26提问于2019-08-28得票数 1
回答已采纳
1回答
我想定义一个Luigi任务，它将在AWS的Postgres数据库中创建一个表。我尝试过不同的导入，例如：
import luigi.contrib.postgres.CopyToTable
from luigi.contrib.postgres import CopyToTable
import luigi.contrib.postgres
from luigi.contrib import postgres
import luigi.postgres ( i thing this is depreceated)
我收到了一个不同的错误，但总是相似的，比如：
AttributeError: m
浏览 2提问于2017-05-15得票数 1
我在用python代码调用/triggering Luigi时遇到了麻烦。
基本上，我需要触发一个luigi任务，就像我们在命令行上所做的那样，但是要从python代码中触发。
我使用supbrocess.popen使用shell命令调用luigi任务
我有一个名为test.py的测试代码，并在模块task_scheduler.py中有一个测试类，它包含我的luigi任务(两个模块位于同一个位置/dir)
import luigi
class TestClass(luigi.Task):
# param = lui
浏览 1提问于2016-06-14得票数 1
我正在使用luigi.build方法尝试luigi的多处理能力。但是我在执行的时候遇到了一些库错误。
for next in self._add(item，is_complete)：文件请求行604，_add self._validate_dependency(d)文件异常行622，_validate_dependency raise Exception('requires() in is_complete objects')
这是我试图实现目标的一段代码。
import luigi
class TaskOne(luigi.Task):
custid= luigi
浏览 16提问于2017-08-11得票数 0
error: can't copy 'luigi\static\visualiser\lib\URI.js': doesn't exist or not a regular file
----------------------------------------
Failed building wheel for luigi
Running setup.py clean for luigi
Failed to build luigi
Installing collected packages: luigi
Running setup.py install f
浏览 2提问于2017-02-13得票数 3
回答已采纳
我正在构建一个luigi任务，我希望将一个numpy数组作为输入参数之一的缺省值：
import numpy as np
import luigi
class MyLuigiTask(luigi.Task):
my_array = luigi.Parameter(default=np.array([2, 3]))
然后在运行时得到以下错误：
Traceback (most recent call last):
File "/home/dalyag/MyModule/my_nodule.py", line 124, in <module>
exe
浏览 0提问于2019-12-31得票数 1
1回答
我需要担心luigi中的文件原子性吗?使用下面的代码，选择一个数据帧并将其作为任务的输出返回。我不明白原子性的部分，因为我希望luigi在声明任务完成之前，只需等待任务完成写入文件。
class readSQLtoPickle(luigi.Task):
sql = luigi.Parameter()
pickle = luigi.Parameter()
def output(self):
return luigi.LocalTarget(self.pickle,format=format.Nop)
def run(self):
浏览 6提问于2020-08-07得票数 0
2回答
我正在尝试使用Luigi-Jupyter教程链接：
我已经成功地设置了我的环境，但正在为将第一个任务添加到我的管道中而苦苦挣扎。
我使用nano file.py创建了一个python文件，并添加了以下代码：
import os
import luigi
output_path = '/Users/mattiaciollaro/Git/luigi_tutorial/output/'
class TellMeMyName(luigi.Task):
"""
An incredibly simple task that writes your
浏览 3提问于2019-07-24得票数 0
我在Python中使用luigi，试图理解luigi参数在编译时和运行时是如何工作的。
import luigi
import datetime
class HelloWorld(luigi.Task):
run_dt = luigi.DateSecondParameter(default = datetime.datetime.now())
def requires(self):
return(None)
def output(self):
return(luigi.LocalTarget('helloworld.txt
浏览 2提问于2017-07-18得票数 0
1回答
我有一个luigi预处理任务，它将我的原始数据分割成更小的文件。然后，这些文件将由实际管道处理。
因此，关于参数，我想要求每个管道都有一个预处理文件id作为参数。但是，此文件id仅在预处理步骤中生成，因此仅在运行时才知道。为了说明我的想法，我提供了以下不起作用的代码：
import luigi
import subprocess
import random
class GenPipelineFiles(luigi.Task):
input_file = luigi.Parameter()
def requires(self):
pass
de
浏览 1提问于2017-02-21得票数 3
1回答
我计划使用Luigi编写一个可重现的、抗故障的超参数调优任务。因此，我在“父”class HParamOptimizer中多次调用我的class TrainOneModel。 为了简单起见，这里是一个更简单的hello world版本： import luigi
# Child class
class HelloTask(luigi.Task):
name = luigi.parameter.Parameter(default='Luigi')
def run(self):
print(f'Luigi says: Hello {se
浏览 18提问于2020-01-09得票数 0
回答已采纳
3回答
我一直在尝试将Luigi整合为我们的工作流程处理程序。目前我们正在使用concourse，但是我们尝试做的许多事情在concourse中是一个麻烦，所以我们切换到Luigi作为我们的依赖管理器。到目前为止，没有任何问题，工作流触发和执行正常。
当一个任务由于某种原因而失败时，问题就出现了。这种情况特别是任务的requires块，但是所有的情况都需要处理。到目前为止，Luigi优雅地处理了这个错误，并将其写入STDOUT。它仍然发出并退出代码0，这意味着作业通过了。假阳性。
我一直在尝试让事件处理来解决这个问题，但是我不能让它触发，即使是一个非常简单的工作：
@luigi.Task.event_
浏览 20提问于2017-03-01得票数 6
1回答
我很难理解如何在Luigi中实现可重用的任务，然后在具体的情况下使用它们。
例如。我有两个一般任务，对一个文件做一些事情，然后输出结果：
class GffFilter(luigi.Task):
"Filters a GFF file to only one feature"
feature = luigi.Parameter()
out_file = luigi.Parameter()
in_file = luigi.Parameter()
...
class BgZip(luigi.Task):
"bgZips a
浏览 3提问于2017-09-28得票数 0
我有一条简单的管道。
我想用Id 2381启动一次，然后当第一个作业正在运行时，我想用Id 231启动第二次运行。第一次运行按预期完成。
第二次运行返回此响应。
Pid(s) set([10362]) already running
Process finished with exit code 0
我是这样开始跑步的
运行一个：
luigi.run(
cmdline_args=["--id='newId13822'", "--TaskTwo-id=2381"],
main_task_cls=TaskTwo()
)
运行二：
lu
浏览 5提问于2017-07-11得票数 0
回答已采纳
2回答
当我运行luigi任务时，有时会遇到框架崩溃，导致以下任务全部失败。在这里，错误日志信息：
2017-10-05 22:02:02,564 luigi-interface WARNING Failed pinging scheduler
2017-10-05 22:02:03,129 requests.packages.urllib3.connectionpool INFO Starting new HTTP connection (126): localhost
2017-10-05 22:02:03,130 luigi-interface ERROR Failed conn
浏览 3提问于2017-10-10得票数 2
我正在使用Luigi进行依赖解析，它工作得很好。
我是否也应该将Luigi用于“软”依赖项？
让我解释一下。假设我当前的任务下载并处理一系列日期的数据。假设在那之后，我想运行一个遍历所有数据并输出一个简单摘要的脚本。不是每个日期一个摘要，而是所有已下载数据的一个摘要。
我称之为软依赖，因为我希望我的最终脚本在所有日期的数据上运行，但如果有几个日期无法下载，我仍然希望为其他日期运行脚本。
我应该如何组织这个用例的任务，或者这不是Luigi的工作？
浏览 0提问于2019-01-31得票数 1
1回答
计算百分比的数学
、、
我希望有人能帮我找到一个简单的公式来计算完成任务的百分比。
我有三个变量：
$totaltasks
$remainingtasks
$completedtasks
让我们假设他们有这些价值：
$totaltasks = 100
$remainingtasks = 30
$completedtasks = 70
在本例中，70%的任务已经完成。但这很容易计算出来，因为$totaltasks是100，并且简单地使用了$totaltasks - $remainingtasks公式。
问题是，只有当$totaltasks总是100，而$remainingtasks总是100或更少时，这个方程才有效。
浏览 3提问于2013-10-20得票数 0
回答已采纳
如何将参数传递给Luigi？如果我有一个名为FileFinder.py的python文件，其中包含一个名为getFIles的类：
class getFiles(luigi.Task):
并且我想要将一个目录传递给这个类，例如：
C://Documents//fileName
然后在我的run方法中使用这个参数
def run(self):
如何在命令行中运行此命令，并添加要在代码中使用的参数？我习惯于在命令行中运行这个文件，如下所示：
python FileFinder.py getFiles --local-scheduler
我应该在代码中添加什么来使用参数，以及如何将该参数添加到命令行参
浏览 5提问于2017-01-26得票数 6
回答已采纳
1回答
我想显示正在读取的文件的进度，并将其放入数组列表中。在此过程中，在应用程序向用户显示列表视图中的数组列表之前，我想显示通过AsyncTask读取了多少文件。目前，一个白色的屏幕是简短地显示。正在读取的文件由大约20000行组成。
AsyncTask (MyTask)在onCreate()方法中执行，如下所示：
private class MyAsync extends AsyncTask<Void, Integer, ArrayList<Movie>> {
ProgressDialog mProgressDialog;
ArrayList<Mov
浏览 0提问于2018-10-28得票数 1
回答已采纳
当前跳过的测试显示在Allure报告中，并且百分比也是基于跳过的报告计算的？有没有任何方法可以将跳过的测试从诱惑力报告中删除？我们目前正在使用最新的版本。
浏览 12提问于2022-02-21得票数 1
5回答
我有一个数组：
array = ['mario','luigi','kong']
我调用它的剪接函数来删除索引之前的所有项()。
array.splice(1) //-> ['luigi','kong']
我只是想知道是否有一个类似于剪接的函数来删除索引之后的所有条目
伪码
array.mirrorsplice(1) //-> ['mario','luigi']
浏览 12提问于2014-10-26得票数 96
回答已采纳
我在用Luigi发射一些管道。让我们举一个简单的例子
task = myTask()
w = Worker(scheduler=CentralPlannerScheduler(), worker_processes=1)
w.add(task)
w.run()
现在，假设myTask在执行过程中引发异常。我所能得到的只是一个来自luigi的日志，显示了异常。
luigi是否可以传播它，或者至少返回一个failure状态？
这样，我就能够使我的程序对这种状态的功能作出反应。
谢谢。
编辑--我忘了在存储结果时指定luigi的输出正在存储数据库。如果引发异常，则不会存储任何结果，但该异常不会传播给
浏览 1提问于2015-09-15得票数 9
回答已采纳
2回答
如何在excel中显示负百分比？
我要找到最坏的情况。
电流输入
if A = 15/19 = 78.9%
B = 162/268 = 60.4%
C = 0/5 = 0.0%
D = 0/90 = 0.0%
需要输出.
FYI numerator and divisor is coming from another cell
if A = 15/19 = 78.9%
B = 162/268 = 60.4%
C = 0/5 = -5.0%
D = 0/90 = -90%
浏览 2提问于2013-03-25得票数 0
回答已采纳
1回答
在luigi.Task.run中，我们需要将对象序列化为文件/数据库/等等：
MyTask(luigi.Task):
param = luigi.Parameter()
def requires(self):
AnotherTask(self.param)
def output(self):
luigi.LocalTarget('out_{}'.format(self.param))
def run(self):
with self.input().open('r') as infi
浏览 1提问于2016-06-03得票数 0
回答已采纳
在Neo4j中，当我在google crome浏览器(最新版本，浏览器页面不是webadmin)中执行查询时，我经常看不到查询的结果。然后是一，二，三的力量...重新运行，它起作用了。这是怎么回事呢？谢谢Luigi
这是devtools的一些错误：
在firefox中，一切工作正常！为什么？
浏览 2提问于2014-09-26得票数 0
1回答
我安装了一个全新的Windows10，安装了一个带有luigi的miniconda env (py34)。当我在命令提示符下运行luigid时，它会启动调度器，当我在chrome中打开localhost:8082时，它会显示我的调度器的web前端。
但是，当我尝试在调度器中运行任务(在PyCharm中使用python )时，我似乎无法连接。我得到的错误是：
Traceback (most recent call last):
File "C:\MiniConda3\envs\py34\lib\site-packages\luigi\rpc.py", line 125, in
浏览 0提问于2017-05-10得票数 2
4回答
在我的sp中，我有一个百分比列名。在该列中，它将显示金额的百分比，但我想更改如果百分比列显示0%，它将仅显示为0。
浏览 1提问于2014-11-12得票数 0
我想知道在dataframe中哪些是不同的变量，这里是一个让您理解我的问题的例子。
dic = { 'a': ['pippo', 'giacomo', 'giacomo', 'francesco', 'luigi', 'francesco', 'luigi'] }
df = pd.DataFrame(dic)
数据文件是
a
pippo
giacomo
giacomo
francesco
luigi
francesco
luigi
francesco
浏览 1提问于2017-04-15得票数 3
回答已采纳
2回答
我在我的工作流程中使用Luigi。我的工作流程一般分为三个部分-导入、分析、导出。在每个部分中，都有多个Luigi任务。
我可以把所有的东西都放在一个文件里。但是如果我想把所有东西都分开，就像拥有data_import.py、analysis.py和export.py一样。
例如，如果data_import.py看起来像这样：
import luigi
class import_task_A(luigi.Task):
def requires(self):
return []
def output(self):
return luigi.Loc
浏览 0提问于2016-10-21得票数 5
1回答
我有一个表("gsheet.table")，其中有一列(“列”)，其中包含“是”或“否”。
YesNoYesNo 列
我想要这张桌子上有两列的图。
一个计算出现“是”的总数。第二个显示了“是”与列中的行总数相比的百分比。
id号百分比
1 *2* 50% (我不知道如何格式化它，但这应该是一个三列表(预期结果))
我对SQL非常陌生，很难做到这一点。你能帮帮我吗？
浏览 3提问于2021-07-22得票数 0
回答已采纳
我正在运行一个EMR星系团(使用纱线)，我正在运行Luigi任务直接从EMR主。我的工作链依赖于S3中的数据，在几个SparkSubmitTasks之后，最终将在Redshift中结束。
import luigi
import luigi.format
from luigi.contrib.spark import SparkSubmitTask
from luigi.contrib.redshift import RedshiftTarget
class SomeSparkTask(SparkSubmitTask):
# Stored in /etc/luigi/client.
浏览 2提问于2015-12-10得票数 0
回答已采纳
1回答
对于我的luigi工作流，我使用的配置文件类似于以下内容：
# Luigi logging configuration
[logging]
version = 1
disable_existing_loggers = false
[logging.formatters.simple]
format = "{levelname:8} {asctime} {module}:{lineno} {message}"
style = "{"
datefmt = "%Y-%m-%d %H:%M:%S"
# Luigi contrib configurat
浏览 1提问于2019-04-25得票数 4
我最近从LUIGI开始，有几个问题我无法用回答自己。
问题是关于路易吉的工作状况。
使用Luigi，我们可以设置一些全局配置(record_task_history)来跟踪作业的历史，并且需要设置mysql。然后，这个历史记录可以被luigi可视化器检索和显示。
这是否支持其他持久性存储？这能在文本文件中完成吗？(Visualizer可能不起作用，但如果我可以以编程方式获得历史状态，则可以接受)
luigi实际上什么时候将状态数据写入持久性存储？它是否将其保存在内存中，以便在作业完成后立即运行作业和转储数据？
--学习者
浏览 2提问于2016-06-29得票数 1
1回答
我想使用Luigi工作流并行加载spark data frame中的多个文件，并将它们存储在字典中。一旦所有的文件都被加载，我希望能够从main中的字典中访问这些数据帧，然后进一步执行processing.This进程。当我运行Luigi时，一个运行Luigi的worker.if与多个工作人员一起运行Luigi时，此变量在main方法中为空。
任何建议都会有帮助。
import Luigi
from Luigi import LocalTarget
from pyspark import SQLContext
from src.etl.SparkA
浏览 9提问于2021-05-18得票数 1
我尝试在名为test.py的命令提示符和名为Test的类中运行1Luigi任务，如下所示
luigi -m test Test --local-scheduler
如何传递参数并在ubuntu中运行
代码示例：
class Test:
def requires(self):
return class1(param1='path', param2='', param3=123)
def run(self):
# some logic with using those 3 params
def out
浏览 0提问于2018-07-24得票数 1
3回答
Java字符串池与反射结合在一起会在Java中产生一些难以想象的结果：
import java.lang.reflect.Field;
class MessingWithString {
public static void main (String[] args) {
String str = "Mario";
toLuigi(str);
System.out.println(str + " " + "Mario");
}
public static void toLu
浏览 33提问于2015-09-17得票数 85
回答已采纳
当我部署一个吊舱时，我受到了这个错误的折磨。
该图像位于与集群相同的项目中的google容器注册表上。
我可以从本地计算机上的注册表中提取图像。
如果我在实例中使用ssh，则无法提取图像。
从文档中可以看出，这应该是开箱即用的。我检查并存储读取访问确实在那里。
下面是配置：
apiVersion: v1
kind: ReplicationController
metadata:
name: luigi
spec:
replicas: 1
selector:
app: luigi
template:
metadata:
name: luigi
浏览 0提问于2018-02-09得票数 2
1回答
我正在测试Luigi在Docker容器中运行任务的能力，也就是说，我希望Luigi从给定的Docker映像中生成一个容器，并在其中执行一个任务。
据我所知，就这一目的而言，存在luigi.contrib.docker_runner.DockerTask。我试图修改它的命令并添加一个输出：
import luigi
from luigi.contrib.docker_runner import DockerTask
class Task(DockerTask):
def output(self):
return luigi.LocalTarget("bla.tx
浏览 4提问于2022-11-21得票数 0
回答已采纳
给出了一个对象列表，这些对象看起来有点像下面的伪代码：
class myObj:
self.init(i, s):
myInt = i
myString = j
我们列出了伪代码列表，其中一个具有如下重要属性：
Mario = [MyObj(0, 'foo1'), MyObj(0, 'foo2'), MyObj(0, 'foo3')]
Luigi = ['bar1', bar2']
然后尝试从MyObj中删除与Luigi中的字符串不匹配的任何Luigi。
for coin in Ma
浏览 2提问于2022-01-07得票数 -2
1回答
我正在使用luigi执行一系列任务，如下所示：
class Task1(luigi.Task):
stuff = luigi.Parameter()
def output(self):
return luigi.LocalTarget('test.json')
def run(self):
with self.output().open('w') as f:
f.write(stuff)
class Task2(luigi.Task):
stuff = luigi.P
浏览 0提问于2018-04-24得票数 5
我有一个简单的luigi管道。 import luigi
import subprocess
import row_count_test
class TaskOne(luigi.Task):
def requires(self):
return None
def run(self):
output = row_count_test()
if output:
with self.output().open('w') as open_file:
op
浏览 17提问于2019-01-29得票数 0
考虑以下任务：
import luigi
class YieldFailTaskInBatches(luigi.Task):
def run(self):
for i in range(5):
yield [
FailTask(i, j)
for j in range(2)
]
class YieldAllFailTasksAtOnce(luigi.Task):
def run(self):
yield [
浏览 37提问于2018-11-28得票数 3
回答已采纳
2回答
我有一个读取.sql文件并输出到BigQuery的luigi任务。
我的问题是，有没有方法可以在不同的.sql文件中重用相同的任务，而不必复制整个luigi任务，即我想要创建模板luigi任务的实例。
class run_sql(luigi.task):
sql_file = 'path/to/sql/file' # This is the only bit of code that changes
def complete(self):
...
def requires(self):
...
def run(
浏览 4提问于2018-08-23得票数 1
回答已采纳
我们试图运行一个简单的管道分布在一个码头群集群。luigi工人被部署为复制的码头服务。他们成功地开始工作，在向luigi服务器请求工作几秒钟之后，他们开始死亡，因为没有分配给他们的工作，所有的任务都分配给了一个工人。
我们不得不在keep_alive=True中设置工人的luigi.cfg，以迫使他们不死，但在输油管道完成后让工人留在这里似乎是个坏主意。有办法控制工作分配吗？
我们的测试管道：
class RunAllTasks(luigi.Task):
tasks = luigi.IntParameter()
sleep_time = luigi.IntParameter(
浏览 4提问于2017-05-08得票数 6
回答已采纳
1回答
我正在尝试配置Luigi的重试机制，以便将失败的任务重试几次。但是，在成功重新尝试任务时，Luigi未成功退出：
===== Luigi Execution Summary =====
Scheduled 3 tasks of which:
* 2 ran successfully:
- 1 FailOnceThenSucceed(path=/tmp/job-id-18.subtask)
- 1 MasterTask(path=/tmp/job-id-18)
* 1 failed:
- 1 FailOnceThenSucceed(path=/tmp/job-id-18
浏览 2提问于2016-10-24得票数 6
回答已采纳
我在通过Windows运行Luigi任务时遇到了问题。以下是事实：
运行安装在C：\ProgramData\ Anaconda 2(Python2.7)中的Anaconda
Anaconda已经将它的路径添加到PATH变量中，但是没有PYTHONPATH变量。
我试图运行的任务位于C:\....\tasks.py中。
试图按以下方式运行它：
C：.模块任务MyTask --dt 20170316
ImportError:没有命名任务的模块
我尝试创建一个PYTHONPATH变量并将确切的路径添加到包含tasks.py文件的目录中，但是它没有工作。我遇到的另一个问题(可能
浏览 4提问于2017-03-16得票数 10
回答已采纳
1回答
如果Luigi图中的某个任务需要在远程Hadoop集群上运行，这是否可行？运行Luigi的机器与Hadoop集群不同。luigi是否仍能检查远程集群中的HDFS文件是否存在？ 我试图找到相关的文档，但是找不到。
浏览 13提问于2019-03-12得票数 0
回答已采纳
在我的angular应用程序中，我使用firestore来存储我用户的个人资料。目前我有这样的东西： /profiles/{uid}/： {
displayName: "Luigi",//--> Only editable by Luigi
email: "luigi@mario.com",
favoriteIceCream: "cookie dough",
roles: {//--> Not editable by Luigi, only by people with admin role
浏览 9提问于2021-08-26得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
