Django批量加载

使用Postgres COPY命令将大量Django模型加载到数据库中。这个库是Django中bulk_create和bulk_update的更高性能的替代方案。

注意：目前，此库仅支持Postgres。将来可能会添加其他数据库。

安装

pip install django-bulk-load

基准测试

bulk_update_models与Django的bulk_update与django-bulk-update的比较

结果

count: 1,000
bulk_update (Django):             0.45329761505126953
bulk_update (django-bulk-update): 0.1036691665649414
bulk_update_models:               0.04524850845336914

count: 10,000
bulk_update (Django):             6.0840747356414795
bulk_update (django-bulk-update): 2.433042049407959
bulk_update_models:               0.10899758338928223

count: 100,000
bulk_update (Django):             647.6648473739624
bulk_update (django-bulk-update): 619.0643970966339
bulk_update_modelsL               0.9625072479248047

count: 1,000,000
bulk_update (Django):             Does not complete
bulk_update (django-bulk-update): Does not complete
bulk_update_models:               14.923949003219604

有关Django性能问题的信息，请参阅此线程。https://groups.google.com/g/django-updates/c/kAn992Fkk24

代码

models = [TestComplexModel(id=i, integer_field=i, string_field=str(i)) for i in range(count)]

def run_bulk_update_django():
  start = time()
  TestComplexModel.objects.bulk_update(models, fields=["integer_field", "string_field"])
  print(time() - start)
  
def run_bulk_update_library():
  start = time()
  TestComplexModel.objects.bulk_update(models, update_fields=["integer_field", "string_field"])
  print(time() - start)
  
def run_bulk_update_models():
  start = time()
  bulk_update_models(models)
  print(time() - start)

bulk_insert_models与Django的bulk_create的比较

结果

count: 1,000
bulk_create:        0.048630714416503906
bulk_insert_models: 0.03132152557373047

count: 10,000
bulk_create:        0.45952868461608887
bulk_insert_models: 0.1908433437347412

count: 100,000
bulk_create:        4.875206708908081
bulk_insert_models: 1.764514684677124

count: 1,000,000
bulk_create:        59.16990399360657
bulk_insert_models: 18.651455640792847

代码

models = [TestComplexModel(integer_field=i, string_field=str(i)) for i in range(count)]

def run_bulk_create():
  start = time()
  TestComplexModel.objects.bulk_create(models)
  print(time() - start)
  
def run_bulk_insert_models():
  start = time()
  bulk_insert_models(models)
  print(time() - start)

API

只需导入并使用以下函数。无需更改settings.py

bulk_insert_models()

INSERT一批模型。它使用Postgres COPY命令来提高速度。如果某行已存在，则整个插入将失败。有关所有参数的描述，请参阅bulk_load.py。

from django_bulk_load import bulk_insert_models

bulk_insert_models(
    models: Sequence[Model],
    ignore_conflicts: bool = False,
    return_models: bool = False,
)

bulk_upsert_models()

UPSERT一批模型。它复制UPSERTing。默认情况下，它通过模型pk匹配现有模型，但您可以使用pk_field_names指定匹配其他字段。有关所有参数的描述，请参阅bulk_load.py。

from django_bulk_load import bulk_upsert_models

bulk_upsert_models(
    models: Sequence[Model],
    pk_field_names: Sequence[str] = None,
    insert_only_field_names: Sequence[str] = None,
    model_changed_field_names: Sequence[str] = None,
    update_if_null_field_names: Sequence[str] = None,
    update_where: Callable[[Sequence[Field], str, str], Composable] = None,
    return_models: bool = False,
)

bulk_update_models()

UPDATE一批模型。默认情况下，它通过模型pk匹配现有模型，但您可以使用pk_field_names指定匹配其他字段。如果在数据库中找不到模型，则忽略它。有关所有参数的描述，请参阅bulk_load.py。

from django_bulk_load import bulk_update_models

bulk_update_models(
    models: Sequence[Model],
    update_field_names: Sequence[str] = None,
    pk_field_names: Sequence[str] = None,
    model_changed_field_names: Sequence[str] = None,
    update_if_null_field_names: Sequence[str] = None,
    update_where: Callable[[Sequence[Field], str, str], Composable] = None,
    return_models: bool = False,
)

bulk_insert_changed_models()

当模型字段在任一compare_field_names中发生变化时，相对于其最新状态，数据库中插入一条新记录。其中，“最新”定义为按order_field_name指定的列以降序对具有给定主键的记录进行排序。如果最新记录没有变化，则不会插入新记录。有关所有参数的描述，请参阅bulk_load.py。

from django_bulk_load import bulk_insert_changed_models

bulk_insert_changed_models(
    models: Sequence[Model],
    pk_field_names: Sequence[str],
    compare_field_names: Sequence[str],
    order_field_name=None,
    return_models=None,
)

bulk_select_model_dicts()

通过filter_field_names选择/获取模型字典。出于性能原因，它返回字典而不是Django模型。当查询大量模型或多个字段IN子句时非常有用。

from django_bulk_load import bulk_select_model_dicts

bulk_select_model_dicts(
    model_class: Type[Model],
    filter_field_names: Iterable[str],
    select_field_names: Iterable[str],
    filter_data: Iterable[Sequence],
    select_for_update=False,
    skip_filter_transform=False,
)

贡献

目前不接受Cedar员工以外的任何人的拉取请求。所有拉取请求都将被关闭。

提交语法

所有PR必须是单个提交，并遵循以下语法https://github.com/angular/angular/blob/master/CONTRIBUTING.md#-commit-message-format

测试

您需要安装并运行以下命令Docker

./test.sh