Python経由でcouchbaseに一括挿入

Question

私はソファベースでバルク挿入をしようとしています。私はSOやgoogleで例を検索しようとしましたが、何の手がかりも得られませんでした。ここで誰かがそれは不可能と言います。

How to insert a documents in bulk in Couchbase?

が、私はこの質問は3年前に頼まれたと思います。私は検索し、私は下のリンクから正しく理解している場合は、一括して文書を挿入することが可能です。ここで

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

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

私はCouchbaseの中

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（データストア[0]）、（STREを一括挿入を実装したい私のコードです）） は1回の挿入を行っています。より高速にするために一括挿入する必要があります。私はcouchbaseで一括挿入する方法を知らなかった。私はこの例を見つけましたが、実装方法は不明です。

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

誰かがCouchbaseのバルクロードのいくつかの例を指摘したり、私は私のコードを経由して一括挿入を行うことができます方法を見つけ出すために私を助けている場合。私は本当に本当に感謝しています。 .thanxはどんなアイデアや助けのためにもたくさんあります。

Answer 1

例をdocsから使用例に変更しようとしました。あなたはおそらく1つまたは2つの詳細を変更する必要がありますが、あなたはその考えを得るべきです。

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))