1

pycassa を使用して Cassandra への挿入を高速化しようとしています。マルチスレッドを使用して複数の接続を開くと、速度が大幅に向上すると聞きました。大量のツイートを json 形式で挿入しています。ここでの私のコードは少しの間動作しますが、スレッドが例外をスローし始めて停止します。スレッドが多いほど動作が速く停止するようです...問題はカサンドラへの接続であると推測しています。接続プーリングで行います。何か案は？

編集: すべてのスレッドが「スレッド Thread-3 で例外をスローします (インタープリターのシャットダウン中に発生する可能性が最も高い):」

import time
import pycassa
from pycassa.pool import ConnectionPool
from pycassa.columnfamily import ColumnFamily
from datetime import datetime 
import json
import threadpool
pool = threadpool.ThreadPool(4)
kspool = ConnectionPool('TweetsKS',use_threadlocal = True)

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) 


def read(file):
    bench = open("bench.txt", "w")
    bench.write(str(datetime.now())+"\n")
    """read data from json and insert into keyspace"""
    json_data=open(file)
    lines = []
    for line in json_data:
        lines.append(line)
    process_tasks(lines)


def insert_into_cfs(line):

    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(user_tweet_cf,tweet_data['from_user_id'],tweet_time,tweet_data['id'])
    new_user_name(user_name_cf,tweet_data['from_user_id'],tweet_data['from_user_name'])
    new_tweet(tweet_cf,tweet_data['id'],tweet_data['text'],tweet_data['to_user_id'])

    if tweet_data['to_user_id'] != 0:
        new_user_follower(user_follower_cf,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(user_tweet_cf,from_user_id,tweet_time,id):
    user_tweet_cf.insert(from_user_id,{(tweet_time): id})

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

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

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

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

1 に答える 1

1

ここでの問題は、スレッドプールの使用にありました。pool.putRequest(req) (ループ外) の後に pool.wait が必要でした。

2 つのスレッドを使用すると、Cassandra の挿入が約 2 倍高速になりますが、何だと思いますか!? それでもMySQLより遅いです!! 6 スレッドでもほぼ同じです...もっといじくり回す必要があると思います!

于 2012-12-12T10:04:39.850 に答える