python csvのループ処理の高速化

Question

前提

Python3.6で、csvを読み込み、条件に該当する時刻から10分以内を条件として、さらに該当する条件のものを抽出するスクリプトを作成しています。
具体的には、以下のようなcsvを読み込み、値が「none」の場合に、①当該時刻から10分以内の範囲を検索し、②同じユーザーかつ同じ値の個数をカウントし、③10個以上あれば当該ユーザーを別のcsvに吐き出すという処理をするものです。

以下のcsvであれば、2022/05/26 20:52:15から2022/05/26 21:00:22の間に、ユーザーAの値が11回noneとなっているので、Aが新たなcsvに吐き出されるようにしようとしています。

しかし、自分の書いたスクリプトでは最初の「none」(ヘッダを含め4行目)を検出し、①から③の条件を確認後、次の処理に進んだ際に読み込んでいるcsvの行が意図しない箇所(ヘッダを含め18行目、条件確認後、処理を終了した行の続き)に進んでいました。
本来であれば、最初の「none」を検出した後、条件に一致しないので次の行からの処理に進み、2回目のnone(ヘッダを含め8行目)を検出して条件の確認、次は3回目のnoneを検出とやっていきたいのですが、ループ処理が上手くいかず詰まっているところです。(自己解決済み)

しかし、対象のcsvが50万行程あるため自分の書いたスクリプトでは物凄い時間が掛かってしまいます。
大量の行があるcsvに対してfor文を3回も使用しているのが原因だと思っているのですが、解決方法が分からず困っているため、お知恵を拝借できればと思っております。
どうぞよろしくお願いいたします。

date,user,atai
2022/05/26 20:41:03,A,koko
2022/05/26 20:42:46,B,atgk
2022/05/26 20:44:08,C,none
2022/05/26 20:47:44,D,moge
2022/05/26 20:48:17,E,yuik
2022/05/26 20:49:06,F,aaaa
2022/05/26 20:50:19,G,none
2022/05/26 20:50:56,H,123
2022/05/26 20:51:33,I,abc
2022/05/26 20:52:00,J,aaa
2022/05/26 20:52:15,A,none
2022/05/26 20:53:15,A,none
2022/05/26 20:53:30,A,none
2022/05/26 20:53:37,K,999
2022/05/26 20:54:05,A,none
2022/05/26 20:55:37,A,none
2022/05/26 20:57:44,A,none
2022/05/26 20:57:49,L,111
2022/05/26 20:58:11,A,none
2022/05/26 20:59:02,A,none
2022/05/26 20:59:58,A,none
2022/05/26 21:00:01,A,none
2022/05/26 21:00:11,A,none
2022/05/26 21:00:22,A,none
2022/05/26 21:01:16,M,111
2022/05/26 21:02:00,N,222
2022/05/26 21:03:22,O,333
2022/05/26 21:04:22,P,444
2022/05/26 21:05:22,Q,555

該当のソースコード

import csv
from datetime import datetime,timedelta

# CSVファイルのパス設定
source_file_path = "hoge.csv" 
result_file_path = "result.csv"

# CSV読み込み
def load_csv(filename, encoding="utf-8"):
	with open(filename, encoding=encoding) as input_file:
		yield from csv.reader(input_file)

# CSV保存
def save_csv(filename, rows, newline=""):
	with open(filename, "w", newline=newline) as output_file:
		csv_writer = csv.writer(output_file)
		csv_writer.writerows(rows)

#　条件(条件①～③)
def process_2(day, us, at, csv):
	count = 0
	for date, user, atai in csv:
		aaa = datetime.strptime(date,"%Y/%m/%d %H:%M:%S")

		if aaa - day < timedelta(minutes = 10) and user == us and atai ==  at:
			count += 1
		elif aaa - day > timedelta(minutes = 11):
			break
		else:
			pass
	return count

#　条件(特定の値を検出)			
def process_1(csv):
	for date, user, atai in csv:
			if atai == "none":
				aa = datetime.strptime(date,"%Y/%m/%d %H:%M:%S")
				bb = user
				cc = atai
				csv2 = load_csv(source_file_path)
				for date, user, atai in csv2
				     dd = process_2(aa,bb,cc,csv)
				     if dd >= 10:
					yield [date, user]
			else :
				pass
	
def main():
	csv = load_csv(source_file_path)
	extract_csv = process_1(csv)
	save_csv(result_file_path, extract_csv)

if __name__ == "__main__":
    main()

ご質問への回答

■tatsu99様
ご質問ありがとうございます。
回答ですが、1については「基準のレコードを含めて10件以上」となります。
2につきましては、ご質問のとおり「基準となるレコードの時刻とユーザのみ」を想定していましたが、私の環境で実行したところ、該当する条件のものがすべて出力されていました。。
しかし、この部分に関しては全て出力されたとしても、後で簡単に処理できるのでこのままでも良いと思っています。

Comments

[tatsu99]

１．１０分以内に１０件以上ということですが、
基準のレコード（例では2022/05/26 20:52:15,A,none）を含めて１０件以上ですか、含めないで１０件以上ですか。
２．出力イメージはどうなりますか。
（基準となるレコードの時刻とユーザのみですか）
例では、以下の１行が出力
2022/05/26 20:52:15,A

[kokoro01234]

ご質問ありがとうございます。
回答ですが、1については「基準のレコードを含めて10件以上」となります。
2につきましては、ご質問のとおり「基準となるレコードの時刻とユーザのみ」を想定していましたが、私の環境で実行したところ、該当する条件のものがすべて出力されていました。。
しかし、この部分に関しては全て出力されたとしても、後で簡単に処理できるのでこのままでも良いと思っています。

Answer 1

最初にCSVファイルを全件読み込み、値がnoneのレコードのみをリストに取り込み、そのリストを検索するようにしました。不明点があれば補足してください。
テストテータのパターンが少ないので、問題があるかもしれません。
（1ユーザーが出力対象となるケース、２ユーザーが出力対象となるケースのみ試験しています）
50万件の（全レコードがnone）のデータで、１０秒程度でしたので、処理速度は問題ないかと思います。

import csv
from datetime import datetime,timedelta

#global変数の設定
# noneのレコードのみのリスト
non_array = []  # 0:date 1:=user 2:datetime,3:flag(True/Flase)
# 出力レコードのリスト
out_array = []  # 0:date 1:user

# CSVファイルのパス設定
source_file_path = "hoge.csv"
result_file_path = "result.csv"

# CSV読み込み
def load_csv(filename, encoding="utf-8"):
    global non_array
    with open(filename, encoding=encoding) as input_file:
        reader = csv.reader(input_file)
        #ヘッダはスキップ
        header = next(input_file)
        for date,user,atai in reader:
            # noneのデータのみ格納
            # date,user,dateのdatetime型,True をnon_arrayへ登録
            if atai == 'none':
                wdate =datetime.strptime(date,"%Y/%m/%d %H:%M:%S")
                non_array.append([date,user,wdate,True])

# CSV保存
def save_csv(filename,newline=""):
    global out_array
    with open(filename, "w", newline=newline) as output_file:
        csv_writer = csv.writer(output_file)
        csv_writer.writerows(out_array)

#　sx～最後までnon_arrayを検索する
#　但しendtimeを超えた場合は打ち切る
def process_2(sx,user,endtime):
    global non_array
    count = 0
    #検索にマッチしたレコード番号のリスト
    outix = []
    for i in range(sx,len(non_array)):
        #時刻がendtimeを超えた場合は終了
        if non_array[i][2] > endtime:
            break
        #userが一致するレコードの場合、カウントアップ、及びレコード番号を追加
        if non_array[i][1] == user:
            count += 1
            outix.append(i)
    return count,outix

#　先頭から順に処理する(Falseのレコードは処理済みのためスキップする)
def process_1():
    global non_array
    global out_array
    band_time = timedelta(minutes=10)
    for i in range(0,len(non_array)):
        # True(未処理)のレコードのみ処理
        if non_array[i][3] == True:
            #終了時刻
            wendtime = non_array[i][2] + band_time
            #user
            user = non_array[i][1]
            #当該ユーザーで10分以内のレコード検索
            count,outix = process_2(i+1,user,wendtime)
            #合計10件以上なら
            if count >= 9:
                #出力リストへ追加
                out_array.append([non_array[i][0],non_array[i][1]])
                #non_arrayの検索対象となったレコードを処理済みに設定する
                for j in outix:
                    non_array[j][3] = False

def main():
    load_csv(source_file_path)
    process_1()
    save_csv(result_file_path)

if __name__ == "__main__":
    main()

Comments

[kokoro01234]

tatsu99様
回答していただきありがとうございました。
スクリプトを修正していただいた上、丁寧な説明まで付けて頂いたので、少し調べるだけでどのような処理をしているのか簡単に理解することができました。
特にグローバル変数を扱うようなスクリプトは扱うことがほとんどなかったので、このように使えば良いのかと大変勉強になりましたし、検索対象となったレコードを処理済みに設定するためにTrue・Falseを使う発想は、私には到底思いつくことができず、ただただ感服いたしました。
処理についても、私が想定していた通りの結果となっていたことに加え、自身で書いたスクリプトが10時間以上経っても終了していなかったのに、提案頂いたスクリプトでは数秒で処理されて大変感激いたしました。
以上のように、私にとって大変勉強になったご回答でしたのでベストアンサーにさせていただきます。
また機会がありましたらアドバイスの程よろしくお願いいたします。

Answer 2

以下のようなアルゴリズムではどうでしょう。少しは早いのではないかと。

○ CSVから各user毎のatai==noneの時刻のリストを作る。
ユーザをキーにしてdict に入れるとこんな感じ。

 { 'A': [2022/05/26 20:52:15, 2022/05/26 20:53:15, 2022/05/26 20:53:30...],
    'B': [... ]}

csvを一回走査するだけで作れます。

○ 作ったdictで、ユーザ毎のリストを見て

• 10個以下なら、そのユーザは対象外
• 10個以上あれば、10個起きに見て差が10分以内なら対象
  0個めと9個めの差が8分なら、10分以内に10個以上あることになる。

こんな感じです。

Comments

[kokoro01234]

TakaiY様
回答していただきありがとうございました。
ご提案頂いた通り、リストを辞書化した上で精査するスクリプトを書いてみました。
このようなスクリプトを今までに書いたことがなかったので、色々と調べる上で大変勉強になりました。
今回は別の方をベストアンサーとさせていただきますが、また機会がありましたらアドバイスの程よろしくお願いいたします。