Django bulk_create z ignorowaniem wierszy, które powodują IntegrityError?

Question

Używam bulk_create do załadowania tysięcy lub wierszy do DB postgresql. Niestety niektóre z tych wierszy powodują IntegrityError i zatrzymują proces bulk_create. Zastanawiałem się, czy istnieje sposób, aby powiedzieć django, aby zignorował takie wiersze i zaoszczędzić jak najwięcej partii, jak to możliwe?

Answer 1

(Uwaga: nie używam Django, więc nie może być bardziej odpowiednie odpowiedzi ramowe specyficzne)

Nie jest możliwe dla Django to zrobić, po prostu ignorując INSERT awarie ponieważ PostgreSQL przerywa całą transakcję na pierwszy błąd.

Django musiałaby jedną z tych podejść:

1. INSERT każdy wiersz w oddzielnej transakcji i ignorować błędy (bardzo powoli);
2. Utwórz przed każdą wstawką SAVEPOINT (może mieć problemy z skalowaniem);
3. Użyj procedury lub kwerendy do wstawienia tylko wtedy, gdy wiersz jeszcze nie istnieje (skomplikowane i wolne); lub
4. Wstawianie danych lub (lepiej) COPY danych do tabeli TEMPORARY, a następnie scalenie ich z serwerem głównym.

Podejście przypominające upsert (3) wydaje się dobrym pomysłem, ale upsert and insert-if-not-exists are surprisingly complicated.

Osobiście biorę (4): Ja bym luzem-wkładkę do nowej osobnej tabeli, prawdopodobnie UNLOGGED lub TEMPORARY, wtedy bym uruchomić jakąś instrukcję SQL do:

LOCK TABLE realtable IN EXCLUSIVE MODE; 

INSERT INTO realtable 
SELECT * FROM temptable WHERE NOT EXISTS (
    SELECT 1 FROM realtable WHERE temptable.id = realtable.id 
); 

LOCK TABLE ... IN EXCLUSIVE MODE zapobiega współbieżnej wstawce, która powoduje, że wiersz powoduje konflikt z wstawką wykonaną przez powyższą instrukcję i nie działa. Nie ma to żadnego znaczenia, ponieważ zapobiega jednoczesnym SELECT s, tylko SELECT ... FOR UPDATE, INSERT, UPDATE i DELETE, więc odczyty z tabeli są normalne.

Jeśli nie możesz sobie pozwolić na blokowanie jednoczesnych zapisów przez zbyt długi czas, możesz zamiast tego użyć zapisywalnego CTE, aby skopiować zakresy wierszy od temptable do realtable, ponawiając próby wykonania każdego bloku, jeśli się nie powiodło.

Answer 2

lub 5. dziel i rządź

nie testowałem lub wyznacznikiem tego dokładnie, ale to działa całkiem dobrze dla mnie. YMMV, w zależności w szczególności od liczby błędów, jakie można spodziewać się w operacji masowej.

def psql_copy(records): 
    count = len(records) 
    if count < 1: 
     return True 
    try: 
     pg.copy_bin_values(records) 
     return True 
    except IntegrityError: 
     if count == 1: 
      # found culprit! 
      msg = "Integrity error copying record:\n%r" 
      logger.error(msg % records[0], exc_info=True) 
      return False 
    finally: 
     connection.commit() 

    # There was an integrity error but we had more than one record. 
    # Divide and conquer. 
    mid = count/2 
    return psql_copy(records[:mid]) and psql_copy(records[mid:]) 
    # or just return False 

Answer 3

Jeden szybki i-brudny obejście tego, że nie wiąże się instrukcja SQL i tabel tymczasowych jest po prostu próba luzem wstawić dane. Jeśli się nie powiedzie, wróć do wstawiania seryjnego.

objs = [(Event), (Event), (Event)...] 

try: 
    Event.objects.bulk_create(objs) 

except IntegrityError: 
    for obj in objs: 
     try: 
      obj.save() 
     except IntegrityError: 
      continue 

Jeśli masz wiele, wiele błędów, to może nie być tak skuteczne (będziesz spędzać więcej czasu niż robi seryjnie włożeniem tak luzem), ale pracuję za pośrednictwem zbioru danych o dużej liczności z niewielu duplikaty, więc rozwiązuje większość moich problemów.

Answer 4

Nawet w Django 1.11 nie można tego zrobić.Znalazłem lepszą opcję niż używanie Raw SQL. To przy użyciu djnago-query-builder. Ma upsert metody

from querybuilder.query import Query 
q = Query().from_table(YourModel) 
# replace with your real objects 
rows = [YourModel() for i in range(10)] 
q.upsert(rows, ['unique_fld1', 'unique_fld2'], ['fld1_to_update', 'fld2_to_update']) 

Uwaga: Biblioteka obsługuje tylko PostgreSQL