Nessos.Streamを使用してクエリ出力を取得して `.CSV`に出力する時間を短縮

Question

これまで、クエリの全体を取得してメモリに結果を格納した後、シーケンスをタイププロバイダ.CSVに送信します。クエリの例：私はそれを理解したようクエリ結果の数は、このISN小さい場合Seqとしてクエリ自体からの出力を維持することはlazy.になりながら

let results = 
    query { 
     for row in db.ThisRow do 
     select row 
     } 
     |> Seq.toList 

、Seq.toList一部を実行するクエリを強制的に大したことじゃない。しかし、結果の数が多い場合（たとえば、1行以上の場合）、最終的にはSystem.OutOfMemoryExceptionとなります。その結果、友人はNessos.Streamライブラリを使用して見てください。

私の目標は、クエリから行を引き出し、その行にいくつかの操作を行い、次にその1行を.CSVに書き込んで、すべての行が最終的に同じ.CSVファイル内の各行に対して繰り返し実行することです。

だから、私は、データベースから照会きたすべての行を取得していて、これは限り動作しますが、これはまだ私はそれが何望むものをやっていないと私は」

open Nessos.Stream 

type StringInt = { 
    String: string option 
    Int: int 
    } 

type StringIntCsvType = CsvProvider<Sample = "item_number, num", 
            Schema = "item_number (string option), num (int)", 
            HasHeaders = true> 
let buildRowFromObject (obj: StringInt) = StringIntCsvType.Row(obj.String, 
                   obj.Int) 

let results = 
    query { 
     for row in db.ThisRow do 
     select row 
     } 
     |> Stream.ofSeq 
     |> Stream.groupBy (fun row -> row.ITEMNUMBER) 
     |> Stream.map (fun (itemString, seq) -> (itemString, (seq |> Seq.length))) 
     |> Stream.map (fun (str, num) -> {String = Some str; 
              Int = num}) 
     |> Stream.map buildRowFromObject 
     |> Stream.toSeq 

let ThisCsv= new StringIntCsvType(results) 
let ThisCsvLoc = "pathToFileLocation" 
let ThisCsv.Save(ThisCsvLoc) 

を試してみてくださいそれがどうなるかわからない。また、大規模なクエリの場合、私はまだSystem.OutOfMemoryExceptionになります。私が思うに、

は私が|> Stream.map buildRowFromObjectラインの下

|> Stream.map (fun x -> new StringIntCsvType(x)) 

を挿入することができるはずだと思いますが、私はこのエラーを与えられなかった場合でもxは

Type Constraint Mismatch. The type 
    CsvProvider<...>.Row 
is not compatible with type 
    Collections.Generic.IEnumerable<CsvProvider<...>.Row> 

と一緒にエラーが発生しましたさその行を追加すると、すべての行に対してという新しい.CSVが作成されます。

私はどのようにクエリを書くことができ、それぞれの異なるクエリを操作し、クエリの各行を同じ.CSVファイルに書き込むことができますか？私は上記のコードで終わっていますか？

Answer 1

中間のStringIntレコードタイプは必要ありません。また、すでにbuildRowFromObjectからCSVシーケンスを取得している場合は、そのファイルに直接書き込むことができます。私はNessos Streamsの大ファンでもありますが、私はそれがあなたのために多くの価値を追加しているのかどうかはわかりません（私は間違っているかもしれないし、もっと複雑なマップ/パイプラインの操作を減らしているかもしれません）。この場合、他のソリューションを調べる前に、まず単純なものを探します。

まず、FSIが64ビットに設定されていることと、exeファイルで64ビットにコンパイルされていることを確認してください。第二に、@ krontogiannisがコメントしているように、あなたの操作はいくつかのグループの項目を数えているようです。私はあなたが実際に各グループを反復したいとは思わないので、なぜDB側のカウントをしないのですか？いくつかのIDとカウントダウンがある一連のタプル（私が望む）を返すでしょう。 CSVTypeproviderに直接フィードできます。したがって：

type StringIntCsvType = CsvProvider<Sample = "item_number, num", 
            Schema = "item_number (string), num (int)", 
            HasHeaders = true> 

let buildRowFromObject (row:string * int) = StringIntCsvType.Row(row) 

let qry = query { 
      for row in tbl1 do 
      groupBy row.ItemNumber into g 
      select (g.Key,g.Count()) 
      } 

let csvout = qry |> Seq.map buildRowFromObject 
(new StringIntCsvType(csvout)).Save(@"C:\tmp\test.csv") 

これは、約700万行を9秒間でcaに書き込みます。 100MBのcsvファイル。あなたはこのような何かにファイルへの書き込みをくくり出す可能性がある、より複雑なアプリケーションで

は：

let writeFile csvout (path:string) = 
    use csvtype = new StringIntCsvType(csvout) 
    csvtype.Save(path) 
writeFile csvout @"C:\tmp\test2.csv" 

、またわからない大量の行での作業についてのマニュアル内のコメントがありますどのように関連しますそれはあなたのためですが、CsvProviderをインスタンス化するときにCacheRows=falseを設定することができます。

編集一部

これは適切ではないかもしれないが、ここにある場合：

#load @"..\..\FSLAB\packages\FsLab\FsLab.fsx" 

#r "System.Data.dll" 
#r "FSharp.Data.TypeProviders.dll" 
#r "System.Data.Linq.dll" 
#r @"..\packages\Streams.0.4.1\lib\net45\Streams.dll" 
#r @"..\packages\FileHelpers.3.1.5\lib\net45\FileHelpers.dll" 

open System 
open System.Diagnostics 
open System.IO 
open System.Collections 
open System.Collections.Generic 
open System.IO.Compression 

open System.Data 
open System.Data.Linq 
open System.Linq 
open Microsoft.FSharp.Data.TypeProviders 
open FSharp.Linq 
open FSharp.Data 
open Nessos.Streams 
open FileHelpers 

[<Literal>] 
let connectionString2 = @"Data Source=(LocalDB)\MSSQLLocalDB;AttachDbFilename=C:\Users\...\Documents\test.mdf;Connection Timeout = 60" 

type dbSchema = SqlDataConnection<connectionString2,StoredProcedures = true> 


type StringIntCsvType = CsvProvider<Sample = "item_number, num", 
            Schema = "item_number (string), num (int)", 
            HasHeaders = true,CacheRows=false> 



let getDbx() = 
    let dbx = dbSchema.GetDataContext() 
    dbx.DataContext.ObjectTrackingEnabled <- false // could impact memory consumption but so far has little effect on time 
    dbx.DataContext.CommandTimeout <- 90 
    dbx 

let dbx = getDbx() 

let tbl1 = dbx.MyTable