MongoDBDestination
class MongoDBDestination(
uri: str,
collections_with_ids: tuple[str, str | None] | List[tuple[str, str | None]],
credentials: UserPasswordCredentials = None,
connection_options: dict = None,
if_collection_exists: Literal['append', 'replace'] = 'append',
use_trxs: bool = False,
docs_per_trx: int = 1000,
maintain_order: bool = False,
update_existing: bool = True,
fail_on_duplicate_key: bool = True,
log_intermediate_files: bool = False,
**kwargs,
)
Bases: DestinationPlugin
Categories: destination
MongoDB-based data outputs.
Initializes the MongoDBDestination with the configuration desired to store the data.
Parameters
uriThe URI of the MongoDB database.
collections_with_idsA tuple or list of tuples with the collection and the name of the field that will be used as the unique identifier. For example, if you want to store the data in a collection called 'my_collection' in database 'my_database' and use the field 'username' as the unique identifier, you would provide the following tuple: ( 'my_database.my_collection', 'username'). If you wanted MongoDB to autogenerate the id, you would provide the following tuple: ('my_database.my_collection', None).
credentialsThe credentials to connect with the database. If None, no credentials will be used.
connection_optionsA dictionary with the options to pass to the pymongo.MongoClient constructor. For example, if you want to set the timeout to 1000 milliseconds, you would provide the following dictionary: {'serverSelectionTimeoutMS': 1000}.
if_collection_existsThe action to take if the collection already exists. If 'append', the data will be appended to the existing collection. If 'replace', the existing collection will be replaced with the new data. Defaults to 'append'.
use_trxsWhether to use a transaction when storing the data in the database. If True, the data will be stored in a transaction, which will ensure that all the data is stored or none of it is (requires that the database is configured with a replica set). If False, the data will be stored without a transaction, which may lead to inconsistent data in the database. Defaults to False.
docs_per_trxThe maximum number of documents to store in a single transaction. If the number of documents to store exceeds this number, the data will be stored in multiple transactions.
maintain_orderWhether to maintain the order of the documents when storing them in the database. If True, the documents will be stored in the same order as they are in the TableFrame. If False, the documents will be stored in the order that they are processed. Defaults to False.
update_existingWhether to update the existing documents in the database. If True, the documents will be updated if they already exist in the database. If False, the documents will be inserted without updating the existing documents, and if a document with the same id already exists execution will fail. Defaults to True.
fail_on_duplicate_keyWhether to raise an exception if a document with the same id already exists in the collection. If True, an exception will be raised. If False, the operation will continue without raising an exception. Defaults to True.
log_intermediate_filesWhether to log when each batch of data is stored in the database. If True, a message will be logged for each batch of data stored. If False, no message will be logged until all the data for a single collection has been stored. Defaults to False.
Properties
collections_with_idsList[tuple[str, str | None]]connection_optionsdictcredentialsUserPasswordCredentials | Noneif_collection_existsLiteral['append', 'replace']uristrMethods
chunkdef chunk(working_dir: str, *results: LazyFrame | None) -> List[List[str] | None]
Trigger the exporting of the data to local parquet chunks. This method will receive the resulting data from the user function and must store it in the local system as parquet files, using the working_dir. Note: This method should not materialize the data, it should only store it in the local system.
Parameters:
working_dirstrThe folder where any files generated must be stored (this refers to temporary files that will be deleted after the execution of the plugin, not the final destination of the data)
resultsThe data to be exported. It is a list of polars LazyFrames or None.
Returns:
A list of the intermediate files created
streamdef stream(working_dir: str, *results: List[LazyFrame | None] | LazyFrame | NoneType)
Trigger the exporting of the data. This method will receive the resulting data from the user function and must store it in the desired location. Note: this method might materialize the data provided in a single chunk generated by the chunk function if invoked, so chunks should be of an appropriate size.
Parameters:
working_dirstrThe folder where any intermediate files generated must be stored (this refers to temporary files that will be deleted after the execution of the plugin, not the final destination of the data)
resultsThe data to be exported. It is a list of polars LazyFrames or None.
Returns:
None
writedef write(files)
This method is used to write the files to the database. It is called from the stream method, and it is not intended to be called directly.