luigi - 路易吉中的 MongoDB

Question

我试图用 luigi 建立一个管道。首先通过从 API 获取数据，转换然后将其保存到 mongo db。我还是 luigi 的新手，我的问题是如何实现 output() 函数，该函数指定输出到 mongo db。我将如何为后续任务创建 require() 函数？

第一个，我正在尝试演示 here ，但它使用的是 MySql 而不是 mongodb。所以我尝试了

from luigi.contrib.mongodb import MongoTarget
from pymongo import MongoClient

def output(self):
    # connect to db
    connection = MongoClient(self.host, self.port)
    db_client = connection[self.db_name]
    collection_name = 'myCollection'

    return MongoTarget(db_client, '_id', collection_name)

但它给了我这样的错误:

TypeError: Can't instantiate abstract class MongoTarget with abstract methods exists

快速搜索错误似乎是由于 pyMongo ，但该解决方案仍然无法解决它。

对于要求部分，我也不知道如何处理它，我想检查记录是否已经存在，所以我不会复制它们。但是我的 API 数据没有唯一索引，所以我想我必须以某种方式扫描所有记录以确保没有重复。

没有很多关于将 mongo 与 luigi 一起使用的文档或示例，感谢您提供任何帮助。

Answer 1

我自己从未使用过 mongodb 包，但您的第一个问题似乎是由于滥用了 MongoTarget 接口(interface)。
如果您查看此处的代码:

https://github.com/spotify/luigi/blob/master/luigi/contrib/mongodb.py#L25

您看到您需要传递 MongoClient 的实例(在您的情况下为 connection 而不是 db_client )。不过你是对的，几乎没有任何关于非核心 Luigi 包的文档。我养成了阅读代码库的习惯，以了解在使用 Luigi 时如何使用给定的包(现在这已成为我使用的任何库的习惯)

话虽如此，我不认为 MongoTarget 是一个实际的目标。它没有实现 exists方法。您应该改用模块中提供的其他目标，即 MongoCellTarget , MongoRangeTarget , MongoCollectionTarget ETC
阅读他们的文档字符串以获取有关他们所做工作的更多信息!