Bulk-Insertion in couchbase über Python

Question

Ich versuche, einige Bulk-Insertion in Couch-Basis zu tun. Ich habe versucht, Beispiele über SO und Google zu suchen, aber ich konnte keine Ahnung bekommen. Hier erwähnt jemand, dass es nicht möglich ist.

How to insert a documents in bulk in Couchbase?

aber ich denke, diese Frage vor 3 Jahren gefragt wurde. Ich suche und wenn ich von unten angegebenen Link richtig verstehe, ist es möglich, das Dokument in großen Mengen einzufügen.

https://developer.couchbase.com/documentation/server/current/sdk/batching-operations.html

https://pythonhosted.org/couchbase/api/couchbase.html#batch-operation-pipeline

Hier ist mein Code, auf dem ich möchte Bulk-Insertion in Couchbase implementieren

import time 
import csv 
from couchbase import Couchbase 
from couchbase.bucket import Bucket 
from couchbase.exceptions import CouchbaseError 
c = Bucket('couchbase://localhost/bulk-load') 
from couchbase.exceptions import CouchbaseTransientError 
BYTES_PER_BATCH = 1024 * 256 # 256K 

with open('/home/royshah/Desktop/bulk_try/roy.csv') as csvfile: 
    lines = csvfile.readlines()[4:] 
for k, line in enumerate(lines): 
    data_tmp = line.strip().split(',') 
    strDate = data_tmp[0].replace("\"", "") 
    timerecord = datetime.datetime.strptime(strDate, 
              '%Y-%m-%d %H:%M:%S.%f') 
    microsecs = timerecord.microsecond 
    strDate = "\"" + strDate + "\"" 
    ts = calendar.timegm(timerecord.timetuple())*1000000 + microsecs 
    datastore = [ts] + data_tmp[1:] 

    stre = {'col1 ': datastore[1], # I am making key-values on the fly from csv file 
      'col2': datastore[2], 
      'col3': datastore[3], 
      'col4': datastore[4], 
      'col5': datastore[5], 
      'col6': datastore[6]} 
    cb.upsert(str(datastore[0]), (stre)) # datastore[0] is used as document 
             id and (stre) is used as key-value to be 
             inserted for respective id. 

cb.upsert (str (Datenspeicher [0]), (stre)) macht Single-Insertion und ich möchte es Bulk-Insertion machen, um es schneller zu machen. Ich hatte keine Ahnung, wie man das in Bulk-Insertion in Couchbase drehen soll. Ich finde dieses Beispiel aber nicht sicher, wie es zu implementieren ist.

https://developer.couchbase.com/documentation/server/current/sdk/batching-operations.html

Wenn jemand einige Beispiele für Bulk-Ladung in Couchbase weist darauf hin, oder mir helfen, herauszufinden, wie kann ich bulk-Insertion über meinen Code zu tun. Ich wäre wirklich sehr dankbar. .thanx viel für irgendeine Idee oder Hilfe.

Answer 1

Ich habe versucht, das Beispiel von der docs an Ihren Anwendungsfall anzupassen. Sie müssen vielleicht ein oder zwei Details ändern, aber Sie sollten die Idee bekommen.

c = Bucket('couchbase://localhost/bulk-load') 
from couchbase.exceptions import CouchbaseTransientError 
BYTES_PER_BATCH = 1024 * 256 # 256K 

batches = [] 
cur_batch = {} 
cur_size = 0 
batches.append(cur_batch) 

with open('/home/royshah/Desktop/bulk_try/roy.csv') as csvfile: 
    lines = csvfile.readlines()[4:] 
for key, line in enumerate(lines): 
    #Format your data 
    data_tmp = line.strip().split(',') 
    strDate = data_tmp[0].replace("\"", "") 
    timerecord = datetime.datetime.strptime(strDate, 
              '%Y-%m-%d %H:%M:%S.%f') 
    microsecs = timerecord.microsecond 
    strDate = "\"" + strDate + "\"" 
    timestamp = calendar.timegm(timerecord.timetuple())*1000000 + microsecs 

    #Build kv 
    datastore = [ts] + data_tmp[1:] 
    value = {'col1 ': datastore[1], # I am making key-values on the fly from csv file 
      'col2': datastore[2], 
      'col3': datastore[3], 
      'col4': datastore[4], 
      'col5': datastore[5], 
      'col6': datastore[6]} 

    key = str(datastore[0] 
    cur_batch[key] = value 
    cur_size += len(key) + len(value) + 24 

    if cur_size > BYTES_PER_BATCH: 
     cur_batch = {} 
     batches.append(cur_batch) 
     cur_size = 0 

print "Have {} batches".format(len(batches)) 
num_completed = 0 
while batches: 
    batch = batches[-1] 
    try: 
     cb.upsert_multi(batch) 
     num_completed += len(batch) 
     batches.pop() 
    except CouchbaseTransientError as e: 
     print e 
     ok, fail = e.split_results() 
     new_batch = {} 
     for key in fail: 
      new_batch[key] = all_data[key] 
     batches.pop() 
     batches.append(new_batch) 
     num_completed += len(ok) 
     print "Retrying {}/{} items".format(len(new_batch), len(ok))