pythonを使って1つの配列に対して並列処理させたい

Question

itemsは配列でその下に多数のオブジェクトが格納されています。
現在は、シングルスレッドで一つずつ処理していますが
マルチスレッドで並列処理させたいです。

私の方で100件ずつ処理するプログラムを考えましたが
いかんせん、初心者のためヘンテコな構造となっております。
1つの配列に対してマルチスレッドで処理する方法をご教示頂けないでしょうか

items =[
			{'blog': 'blog1', 'count': '111', 'user_name': 'user1'}, 
			{'blog': 'blog2', 'count': '112', 'user_name': 'user2'}, 
			{'blog': 'blog3', 'count': '113', 'user_name': 'user3'},
		　　・
		　　・
		　　・
			{'blog': 'blog1000', 'count': '11113', 'user_name': 'user1000'}
	   ]

    table = dynamodb.Table('Movies')
    for item in items:
    	response = table.put_item(Item=item)

items =[
			{'blog': 'blog1', 'count': '111', 'user_name': 'user1'}, 
			{'blog': 'blog2', 'count': '112', 'user_name': 'user2'}, 
			{'blog': 'blog3', 'count': '113', 'user_name': 'user3'},
		　　・
		　　・
		　　・
			{'blog': 'blog1000', 'count': '11113', 'user_name': 'user1000'}
	   ]
	   
       process_list = [] 
       with ThreadPoolExecutor() as executor: 
       i = 0
	 　table = dynamodb.Table('Movies')
	 　for item in items:
		 　if i > 100:
		 　  process_list.append(executor.submit(table.put_item(Item=item)))
		 　elif i > 200:
		 　  process_list.append(executor.submit(table.put_item(Item=item)))
			　　・
			　　・
			　　・
		 　elif i > 900:
		 　  process_list.append(executor.submit(table.put_item(Item=item)))
		 　else:
		 　  process_list.append(executor.submit(table.put_item(Item=item)))
	 　  
	 　  i+=1
                for process in process_list:
                    print(process.result())

↓

items =[
            {'blog': 'blog1', 'count': '111', 'user_name': 'user1'}, 
            {'blog': 'blog2', 'count': '112', 'user_name': 'user2'}, 
            {'blog': 'blog3', 'count': '113', 'user_name': 'user3'},
        　　・
        　　・
        　　・
            {'blog': 'blog1000', 'count': '11113', 'user_name': 'user1000'}
       ]

#リスト分ける
newList = split_list(items,100)
#スレッドプールを作る
tpe = ThreadPoolExecutor(max_workers=100)

for _ in range(newList.len):
    tpe.submit(put_to_db)
    tpe.shutdown()
    print("完了")


#リストを分ける処理
def split_list(l, n):
    for idx in range(0, len(l), n):
        return l[idx:idx + n]

#DBに登録する処理
def put_to_db(Items):

    table = dynamodb.Table('Movies')
    for item in Items:
        response = table.put_item(Item=item)

Answer 1

この書き方では「100個ずつ処理する」にはならず、単に1つずつ別のスレッドで処理するようになってますね。 同じ行で呼んだからといって、同じスレッドで処理されるわけではなく、それは別の登録になります。

何通りかやりかたはありますが、基本的なのはこんな感じでしょう。
・ 値のリストをうけとって、それをループで順にDBに登録する処理する関数(put_to_dbとか)を作る。
・全体のリストを、適当なサイズの塊に分ける。この場合は100個ずつでしょうか。
・スレッドプールを作る。
・ループで、作った関数と、100個ずつ分けたデータをsubmitする。
・結果をうけとる

---

コメントの処理をみてみました。 コメントはインデントなくなっているので、インデントはあつずっぽいです。間にコメントいれます。

#リスト分ける
newList = split_list(items,100)
#スレッドプールを作る
tpe = ThreadPoolExecutor(max_workers=100)

for _ in range(newList.len):
    tpe.submit(put_to_db)
tpe.shutdown()
print("完了")

submitが返すfutureを保存していませが、このfutureで結果を取得できるので、結果が必要なら、保存が必要です。、

#リストを分ける処理
def split_list(l, n):
    for idx in range(0, len(l), n):
        return l[idx:idx + n]

for文で回しても、1つでリターンしてしまったら、1つしか取得できません。
分けたリストを取っておくリストを作って、最後にそれを返すようにします。

#DBに登録する処理
def put_to_db(Items):
    table = dynamodb.Table('Movies')
    for item in Items:
        response = table.put_item(Item=item)

responsで受けた結果をどこにも保存していないので、返してもいません。
結果返却用のリストを作って、そこにresponseを保存して、最後に返すようにします。

---

で、ざっと直してみました。動かしていないので、間違えているところあるはずですが、正解に近付いているとは思います。
もっと効率のいい書き方などあるのですが、元の方式を残す形にしました。
また、読み易さを考えて冗長に書いたところもあります。

#リスト分ける
newList = split_list(items,100)
#スレッドプールを作る
tpe = ThreadPoolExecutor(max_workers=100)

future_list = []
for l in newList:
    f = tpe.submit(put_to_db, l)
    future_list.append(f)
tpe.shutdown()

for r in future_list:
    print(r.result)

print("完了")


#リストを分ける処理
def split_list(l, n):
    new_list
    for idx in range(0, len(l), n):
        new_list.append(l[idx:idx + n])
    return nuw_list


#DBに登録する処理
def put_to_db(item_list):
    result_list = []
    table = dynamodb.Table('Movies')
    for item in Items:
        result = table.put_item(Item=item)
        result_list.append(result)
    return result_list

Comments

[pokemonta]

ご回答ありがとうございます。
教えて頂いたイメージに近い感じで
実装したつもりなんですけど、
どこが決定的に違うんでしょうか？
（リストを１００個ずつに分割して並列に処理させたつもりです）

[TakaiY]

だいぶかけはなれてますよ。
・関数作っていない
・リストを100個ずつにわける処理がない
・よって、スレッドに適切に処理が渡せていない。
です。

process_list.append(executor.submit(table.put_item(Item=item)))

この記述だと、
table.put_item(Item=item)は、渡されるのでなくここで実行されます。
なので、executer.submit()にはその返り値(何かは不明)が渡されています。
で、その結果がprocess_listに格納されています。

まずは、最初の2つをやってみてください。

[pokemonta]

手探りでご教示頂いた方法を作ってみました。

#リスト分ける
newList = split_list(items,100)
#スレッドプールを作る
tpe = ThreadPoolExecutor(max_workers=100)

for _ in range(newList.len):
tpe.submit(put_to_db)
tpe.shutdown()
print("完了")

#リストを分ける処理
def split_list(l, n):
for idx in range(0, len(l), n):
return l[idx:idx + n]

#DBに登録する処理
def put_to_db(Items):

table = dynamodb.Table('Movies')
for item in Items:
response = table.put_item(Item=item)

以下の件がよくわからなかったのですが、どのように実装すればよいのでしょうか？
・ループで、作った関数と、100個ずつ分けたデータをsubmitする。
・結果をうけとる

[pokemonta]

内容確認させて頂います。ありがとうございます。

[pokemonta]

items の数が100でスレッド1で処理した場合、
改修前の8倍場合遅くなります。
一方でitemsの数が500でスレッド5で処理した場合、
itemsの数が1000でスレッド10で処理した場合より遅くなります。
あとは、チューニングになってくるのでしょうか？(分割=スレッド数)

[TakaiY]

うごいたみたいですね。
そうですね、後はチューニングですね。 分割するのにも、タスクを登録するのにも、結果を集めるのにもそれぞれ時間がかかります。最適なところは探ってみる必要があります。
今は、threadですがprocessにしてみるとまた結果は変わります。
受ける DBの処理能力にもよります。

[pokemonta]

ありがとうございました。