1

Cassandraの挿入を高速化するために、マルチスレッドを使用していますが、動作は問題ありませんが、スレッドを追加しても違いはありません。接続を生成していないと思います。おそらく、プールを使用する必要があると思います。 execute（f、* args、** kwargs）ですが、使用方法がわかりません。ドキュメントは非常に乏しいです。これが私のコードです。

import connect_to_ks_bp
from connect_to_ks_bp import ks_refs
import time
import pycassa
from datetime import datetime 
import json
import threadpool
pool = threadpool.ThreadPool(20)
count = 1
bench = open("benchCassp20_100000.txt", "w")

def process_tasks(lines):

    #let threadpool format your requests into a list
    requests = threadpool.makeRequests(insert_into_cfs, lines)

    #insert the requests into the threadpool
    for req in requests:
        pool.putRequest(req) 

    pool.wait()

def read(file):
    """read data from json and insert into keyspace"""
    json_data=open(file)
    lines = []
    for line in json_data:
        lines.append(line)
    print len(lines)
    process_tasks(lines)


def insert_into_cfs(line):
    global count
    count +=1
    if count > 5000:
            bench.write(str(datetime.now())+"\n")
            count = 1
    #print count
    #print kspool.checkedout()
    """
    user_tweet_cf = pycassa.ColumnFamily(kspool, 'UserTweet')
    user_name_cf = pycassa.ColumnFamily(kspool, 'UserName')
    tweet_cf = pycassa.ColumnFamily(kspool, 'Tweet')
    user_follower_cf = pycassa.ColumnFamily(kspool, 'UserFollower')
    """
    tweet_data = json.loads(line)
    """Format the tweet time as an epoch seconds int value"""
    tweet_time = time.strptime(tweet_data['created_at'],"%a, %d %b %Y %H:%M:%S +0000")
    tweet_time  = int(time.mktime(tweet_time))

    new_user_tweet(tweet_data['from_user_id'],tweet_time,tweet_data['id'])
    new_user_name(tweet_data['from_user_id'],tweet_data['from_user_name'])
    new_tweet(tweet_data['id'],tweet_data['text'],tweet_data['to_user_id'])

    if tweet_data['to_user_id'] != 0:
        new_user_follower(tweet_data['from_user_id'],tweet_data['to_user_id'])


""""4 functions below carry out the inserts into specific column families"""        
def new_user_tweet(from_user_id,tweet_time,id):
    ks_refs.user_tweet_cf.insert(from_user_id,{(tweet_time): id})

def new_user_name(from_user_id,user_name):
    ks_refs.user_name_cf.insert(from_user_id,{'username': user_name})

def new_tweet(id,text,to_user_id):
    ks_refs.tweet_cf.insert(id,{
    'text': text
    ,'to_user_id': to_user_id
    })  

def new_user_follower(from_user_id,to_user_id):
    ks_refs.user_follower_cf.insert(from_user_id,{to_user_id: 0})   

    read('tweets.json')
if __name__ == '__main__':

これは単なる別のファイルです。

import pycassa
from pycassa.pool import ConnectionPool
from pycassa.columnfamily import ColumnFamily

"""This is a static class I set up to hold the global database connection stuff,
I only want to connect once and then the various insert functions will use these fields a lot"""
class ks_refs():
    pool = ConnectionPool('TweetsKS',use_threadlocal = True,max_overflow = -1)

    @classmethod
    def cf_connect(cls, column_family):
        cf = pycassa.ColumnFamily(cls.pool, column_family)
        return cf

ks_refs.user_name_cfo = ks_refs.cf_connect('UserName')
ks_refs.user_tweet_cfo = ks_refs.cf_connect('UserTweet')
ks_refs.tweet_cfo = ks_refs.cf_connect('Tweet')
ks_refs.user_follower_cfo = ks_refs.cf_connect('UserFollower')

#trying out a batch mutator whihc is supposed to increase performance
ks_refs.user_name_cf = ks_refs.user_name_cfo.batch(queue_size=10000)
ks_refs.user_tweet_cf = ks_refs.user_tweet_cfo.batch(queue_size=10000)
ks_refs.tweet_cf = ks_refs.tweet_cfo.batch(queue_size=10000)
ks_refs.user_follower_cf = ks_refs.user_follower_cfo.batch(queue_size=10000)
4

1 に答える 1

1

いくつかの考え：

  • 10,000のバッチサイズは大きすぎます。100を試してください。
  • ConnectionPoolのサイズを、pool_sizeパラメーターを使用するスレッドの数と少なくとも同じ大きさにします。デフォルトは5です。プールオーバーフローは、アクティブなスレッドの数が時間の経過とともに変化する可能性がある場合にのみ使用する必要があります。スレッドの数が固定されている場合は使用しないでください。その理由は、新しい接続の不必要な開閉が多くなり、かなりコストのかかるプロセスになるためです。

これらの問題を解決したら、次を調べてください。

  • 私はあなたが使用しているスレッドプールライブラリに精通していません。写真からCassandraへの挿入を削除する場合は、スレッドの数を増やすとパフォーマンスが向上することを確認してください。
  • Python自体には、GILのために役立つスレッドの数に制限があります。通常は最大で20になることはありませんが、CPUを集中的に使用する場合や、Pythonの解釈を多く必要とする場合に発生する可能性があります。前のポイントで説明したテストでも、これについて説明します。モジュールの使用を検討する必要がある場合もありますがmultiprocessing、それを処理するにはコードを変更する必要があります（つまり、ConnectionPools、CF、またはその他のプロセス間でほとんど何も共有しない）。
于 2012-12-31T21:09:51.910 に答える