csvresumablegithub.comはじめにCSVを消費する処理を再開可能にしたいという気持ちになりました。具体的には、1つ1つの処理にそこそこ時間が掛かる(30秒から1分)ものをそこそこ多く(104件くらい)処理しないといけないことがあったのですが。DBとか用意したり使ったりするの面倒だなと思ったときのことです。CSVを消費したい(再開したい)例えば以下の様なイメージです(実際の処理とは異なります)。input.csvid,x,y1,10,202,100,100このようなcsvがあって、これらの各行に対して処理を行う(例えば和を求める)必要があるとします。このあたりで止めたい(再開したい)2,200実際、重いと言っても計算的なものではなく主に帯域制限的なものが原因です。並列化とかほぼ意味がない状態なのですが。途中で失敗したら辛いという感じの状況です。随時終わるたびに書き出していき、終わったところまで入力を削るみたいな作業をして手動で入力となるファイルを書き換えていっても良いのですが。csvresumableまぁそんなわけでだるかったので。したライブラリを作ることにしました(まだ開発途中なのでAPIの変更は普通にあると思います)。具体的には以下の様な形で動きます。通常のCSVのDictReaderと同様に動くどこまで終わったのかを別途記録する(history.csv(実際には別の場所に記録されます))(再開時には、記録していたところまでの入力はスキップする)状態などを管理するのは面倒だったので、完全にinsertだけで済むようにしました。ここで止めたい2,100,100以下の様な履歴(history.csv)を用意し(csvである必要はない)id1再開(resume)の際はこれとzipしたiterator(概念上)に対して処理を行えば良いということになります。履歴に残ったものはスキップしてしまえば良いということです。ちなみに、pandasなどのinterfaceを用意しなかった理由は、そもそも計算や集計が目的ではなかったためです。streamとしてCSVがあれば良いというだけだったので(つまりCSVである理由も特にない可能性があります。そのあたりも込みでinterfaceが変わりうるという感じです)。以下の様な形で書けます。sys.stdout.flush()CSVから1行ずつ取得していき、それを入力として何らかの処理を行うという形です。この処理がそこそこ重めの処理(と言っても先程言った通りにほぼほぼ流量制限が原因で高速化できない)になっており、数十秒程度掛かると言ったものだとします。(再掲)に対して実行し、id,x,y1,10,202,100,100実行を途中で止めます(id=2の部分は計算が終わらない。KeyboardInterrupt途中で止まったので、途中から再開したいはずです。いう環境変数と一緒に実行すると再開(resume)できます。main.pyちなみにRESUMEをつけないとはじめからやり直しです(つまり何も知らない人にとってはただのcsv.DictReaderとして動く)。"100"}環境変数以外の方法ところで環境変数で設定というのが、設定より規約(CoC)っぽい感じがして嫌という人いると思います。そんな人は真面目にオプションを与えてください。"100"}idとして扱う値を変えたい場合idとして使う値を変えたくなることがあるかもしれません。その場合にはkeyオプションがあります。これはsorted()関数と同様のイメージで考えてもらえれば良いです。渡されるCSVというのはかならずしも常に自分の意図した通りの構造で渡されるということがなかったりしますし。以下の様な形かもしれません。groupId,userId,name,age,cache1,1,foo,8,10001,2,bar,10,2002,3,boo,2,03,4,bar,2,100デフォルトでは左端をidとして扱いますが、上の例では左端のgroupIdではなくuserIdをidとして消費したくなると思います。このような場合には以下の様に書けば良いです。sys.stdout.flush()テキトウに年齢(勤続年数?)を貯金で割って、1万円?を稼ぐのに何年掛かるのかというような値でも計算してみましょう(これはテキトーな例です)。zeroおや、エラーになってしまいましたね。0除算を気にしてませんでした(まぁこういう感じでたまに考慮漏れのエラーがあったりします)。sys.stdout.flush()再開します。0.02途中から再開できてますね。複数のCSVを合成した結果を元に消費したい場合ところで、今までは入力がひとつだけでしたが。複数の入力が必要になることもあると思います。そのような場合にも一応対応はしています。直列につなぐ場合(concat)単純に複数に分割されたファイルを入力だとしましょう。input.csvgroupId,userId,name,age,cache1,1,foo,8,10001,2,bar,10,200input2.csvgroupId,userId,name,age,cache2,3,boo,2,03,4,bar,2,100そのような場合はつなげるだけです。sys.stdout.flush()注意点としてはDictReaderのiteratorをリストなどにして消費しないようにしてください。0.02ファイルの切れ目など関係なくresumeできています。これは当然と言えば当然なのですが。順序であるなど順序が不定の場合にはおかしくなります)。並列につなぐ場合(groupby)今度は並列につなぐ場合を考えてみます。先程のcsvについてgroupIdでgroupingされた結果に対する何らかの処理をしてみるということにしてみましょう。そのような場合でも考え方は同様です。streamのようなものが構成されていればそれで十分です(入力がCSVである必要も特にありません)。"100"}]例を見てわかる通り、実は入力がCSVである必要はありません。一定の順序を保った何らかのeventのstreamであれば大丈夫です(pythonで言えばiterator)。xs[0])、もちろんkeyオプションがとれます。chainしてsortしてとやっているので、原理的には与えられたファイルを全部見ているわけですが。そもそも冒頭で触れたように元となる入力の数自体はせいぜい104程度しかありません。そこまでコストというわけでもないです。途中で止めてRESUMEで再開できます。100おまけちなみにchainしてsortしてgroupbyというのは結構よくやる処理なのですが。毎回書くのもめんどくさいのでconcat_groupbyという関数を用意しています。row["groupId"])ところで先頭N件だけ取りたいという場合にはitertools.isliceが使えます。streamは消費しないように気をつけてください(消費というのはlist(gen(files))のようなことを指してます)。再開時に過去の出力を覚えておきたい場合さて、いままでは処理の中断・再開を扱ってきましたが。出力を全て通して行いつつ、実際の処理自体は中断・再開したいということがあります。先程のスクリプトが以下のようなMakefileに書かれていたタスクだったとします。output.csvここで、処理を再開したときには、過去分も含めた全ての実行結果が渡されて欲しいはずです(もちろん、呼び出すスクリプト側でファイル入出力を行い、追記でやるという案もあります)。このようなときにちょっと一手間を加えると良い感じにできます。呼び出し方によってはbufferingされてしまう場合もあるcaptureという名前が良いかはまだ微妙ですが。このコンテキストマネージャでくるんであげるとその間の出力を記録しておけます(ちなみに引数で記録したいstreamは変更できます。デフォルトが標準出力)。再開時には記録していた出力を再度出力してくれるため、再開(resume)時にも過去も含めた全てを出力をしてれるようになります。100そんなわけでteeを使っていても、再開後のファイル中に全ての結果が残ります。100最後に裏側の話はまた今度。手触りの良いscriptというものについて最近考えている。componentという機能を追加した月別アーカイブはてなブログをはじめよう!podhmoさんは、はてなブログを使っています。ブログを報告するスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。