Frames-streamly 0.1.1.0 → 0.1.1.1
raw patch · 6 files changed
+23/−437 lines, 6 filesdep −binarydep −bytestringdep −bytestring-strict-builderdep ~Framesdep ~basedep ~primitivePVP ok
version bump matches the API change (PVP)
Dependencies removed: binary, bytestring, bytestring-strict-builder, cereal, clock, fast-builder, foldl, mtl, streamly-bytestring, vector
Dependency ranges changed: Frames, base, primitive, relude, streamly, strict, text, vinyl
API changes (from Hackage documentation)
Files
- Frames-streamly.cabal +1/−37
- Readme.md +22/−17
- examples/DemoPaths.hs +0/−1
- examples/Serialize.hs +0/−140
- examples/Strictness.hs +0/−222
- examples/StrictnessPaths.hs +0/−20
Frames-streamly.cabal view
@@ -7,7 +7,7 @@ -- hash: f129113aa17f26529351746660f5b3ada35ea09fec7866ca91d25902f08ac8be name: Frames-streamly-version: 0.1.1.0+version: 0.1.1.1 synopsis: A streamly layer for Frames I/O description: More information is available in the <https://github.com/adamConnerSax/Frames-streamly/blob/master/Readme.md readme>.' category: Data@@ -72,40 +72,4 @@ , streamly , text , vinyl- default-language: Haskell2010--test-suite Strictness- type: exitcode-stdio-1.0- main-is: Strictness.hs- other-modules:- Serialize- StrictnessPaths- Paths_Frames_streamly- hs-source-dirs:- examples- ghc-options: -Wall -O2 -fdicts-strict -fmax-worker-args=16 -fspec-constr-recursive=16--- ghc-options: -Wall -prof -fprof-auto -eventlog -rtsopts -fexternal-interpreter -fdicts-strict -fmax-worker-args=16 -fspec-constr-recursive=16- build-depends:- Frames- , Frames-streamly- , base- , bytestring- , cereal- , binary- , fast-builder- , bytestring-strict-builder- , mtl- , foldl- , primitive- , relude- , streamly- , streamly-bytestring- , strict- , clock- , text- , vector- , vinyl- mixins: base hiding (Prelude)- , relude (Relude as Prelude- , Relude.Extra) default-language: Haskell2010
Readme.md view
@@ -1,26 +1,31 @@-# Frames-streamly- v 0.1.0.2+# Frames-streamly- v 0.1.1.1 [![Build Status][travis-badge]][travis] [![Hackage][hackage-badge]][hackage] [![Hackage Dependencies][hackage-deps-badge]][hackage-deps] -This library contains some useful functions for using the -[Frames](https://hackage.haskell.org/package/Frames) +* Breaking Change (from 0.1.0.2): Some streaming functions+use the ```StrictReadRec``` class, a stricter version of+```ReadRec``` from Frames. This class is located in+Frames.Streamly.CSV.++This library contains some useful functions for using the+[Frames](https://hackage.haskell.org/package/Frames) package with [streamly](https://hackage.haskell.org/package/streamly). -Frames has some built-in dependencies on the +Frames has some built-in dependencies on the [Pipes](https://hackage.haskell.org/package/pipes) package, a few of which--primarily file I/O-- require users of Frames to use-the Pipes package explicitly. Streamly provides much of the same -functionality as Pipes and may be some users preferred streaming +the Pipes package explicitly. Streamly provides much of the same+functionality as Pipes and may be some users preferred streaming interface. This package replicates all the external-facing bits of Frames that rely on Pipes and uses streamly instead. It also fleshes out the Frames API in a couple of places: -1. It adds some flexibility to the functions to write CSV files. -Frames supported formatting of fields for CSV +1. It adds some flexibility to the functions to write CSV files.+Frames supported formatting of fields for CSV via a typeclass ```ShowCSV```. That is supported here as well. But this package also supports using the ```Show``` instance, and, for more customized formatting,@@ -28,30 +33,30 @@ a [Vinyl](https://hackage.haskell.org/package/vinyl) record of functions. Helpful combinators are provided for formatting any single field with a ```Show``` instance or a ```ShowCSV``` instance or a user provided-function from the field type to ```Text```. +function from the field type to ```Text```. -2. It adds some (experimental) support for ```Frame``` transformations +2. It adds some (experimental) support for ```Frame``` transformations using Streamly streams as an intermediate state for transformations which may benefit from the concurrency available in streamly. Such-transformations first make any foldable of Records +transformations first make any foldable of Records (including a ```Frame```) into a stream, apply a streamly transformation to a stream of some other records and then transforms those into a frame. So the result is a frame -> frame function but one that can take advantage of streamly's features at the cost of the transformation into a stream and then back in to a Frame.-This allows use of the concurrent features of -Streamly for functions like ```mapM``` or ```mapMaybeM```. +This allows use of the concurrent features of+Streamly for functions like ```mapM``` or ```mapMaybeM```. -3. It adds Streamly folds for the various stream to in-core -transformations in case users want to use them directly +3. It adds Streamly folds for the various stream to in-core+transformations in case users want to use them directly in stream to Frame transformations. Frames exposed only the functions-to transform an entire stream (a pipe producer) into Frames "AoS" +to transform an entire stream (a pipe producer) into Frames "AoS" structure. This library provides that functionality as well, in this case using streamly streams as the input. But here we also expose streamly folds from streams of Records to Frames so that more complex stream to Frame transformations can be done by the user. For example, suppose you are doing-a map/reduce on a large data set and you want to store the +a map/reduce on a large data set and you want to store the grouped subsets as Frames for memory-efficiency. These folds make that simpler.
examples/DemoPaths.hs view
@@ -1,6 +1,5 @@ module DemoPaths where -import qualified Frames.TH as Frames import qualified Paths_Frames_streamly as Paths forestFiresPath :: FilePath
− examples/Serialize.hs
@@ -1,140 +0,0 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE DerivingVia #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE TypeApplications #-}-{-# LANGUAGE TypeOperators #-}-{-# LANGUAGE UndecidableInstances #-}-{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}-module Serialize- (- -- * Types- SElField(..)- , RecSerialize- , RecBinary- -- * Record coercions- , toS- , fromS- -- * Wrapper for serializable frames- , SFrame (..)- )-where--import qualified Control.Monad.ST as ST-import qualified Data.Vinyl as V-import qualified Data.Vinyl.TypeLevel as V--import Data.Binary as B-import Data.Binary.Put as B-import Data.Binary.Get as B-import Data.Serialize as S-import qualified Frames as F-import qualified Frames.InCore as FI-import qualified Frames.Streamly.InCore as FS--import GHC.Generics (Generic,Rep)-import GHC.TypeLits (KnownSymbol)--import qualified Streamly-import qualified Streamly.Prelude as Streamly-import qualified Streamly.Internal.Data.Fold as Streamly.Fold---newtype SElField t = SElField { unSElField :: V.ElField t }-deriving via (V.ElField '(s,a)) instance (KnownSymbol s, Show a) => Show (SElField '(s,a))-deriving via (V.ElField '(s,a)) instance (KnownSymbol s) => Generic (SElField '(s,a))-deriving via (V.ElField '(s,a)) instance Eq a => Eq (SElField '(s,a))-deriving via (V.ElField '(s,a)) instance Ord a => Ord (SElField '(s,a))---toS :: V.RMap rs => V.Rec V.ElField rs -> V.Rec SElField rs-toS = V.rmap coerce-{-# INLINE toS #-}--fromS :: V.RMap rs => V.Rec SElField rs -> V.Rec V.ElField rs-fromS = V.rmap coerce-{-# INLINE fromS #-}---- those generic instances allow us to derive instances for the serialization libs--- instance (S.Serialize (V.Snd t), V.KnownField t) => S.Serialize (V.ElField t)-instance (S.Serialize (V.Snd t), V.KnownField t) => S.Serialize (SElField t)-instance (B.Binary (V.Snd t), V.KnownField t) => B.Binary (SElField t)--type RecSerialize rs = (GSerializePut (Rep (V.Rec SElField rs))- , GSerializeGet (Rep (V.Rec SElField rs))- , Generic (V.Rec SElField rs))--instance RecSerialize rs => S.Serialize (V.Rec SElField rs)--type RecBinary rs = (GBinaryPut (Rep (V.Rec SElField rs))- , GBinaryGet (Rep (V.Rec SElField rs))- , Generic (V.Rec SElField rs))--instance RecBinary rs => B.Binary (V.Rec SElField rs)--newtype SFrame a = SFrame { unSFrame :: F.Frame a }--type SFrameRec rs = SFrame (F.Record rs)---- Cereal-instance (V.RMap rs, FI.RecVec rs, RecSerialize rs) => S.Serialize (SFrame (F.Record rs)) where- put = streamlyPutC . Streamly.map toS . Streamly.fromFoldable . unSFrame- {-# INLINEABLE put #-}- get = sframeGetC- {-# INLINEABLE get #-}---- we use only one fold to get the length and build the bytestream-streamlyPutC :: S.Serialize a => S.Putter (Streamly.SerialT Identity a)-streamlyPutC s = do- let lengthF = Streamly.Fold.length- putF = Streamly.Fold.Fold (\b a -> return $ b <> S.put a) mempty return- (l, streamPut) = runIdentity $ Streamly.fold (Streamly.Fold.tee lengthF putF) s- S.putWord64be $ fromIntegral l- streamPut-{-# INLINEABLE streamlyPutC #-}---- the ST monad is...tricky! In some sense, is each invocation of @go@ using a "different" s?-sframeGetC :: forall rs. (FI.RecVec rs, V.RMap rs, RecSerialize rs) => S.Get (SFrameRec rs)-sframeGetC = go Streamly.nil =<< S.getWord64be where- go :: (forall s.Streamly.SerialT (ST.ST s) (F.Rec SElField rs)) -> Word64 -> S.Get (SFrameRec rs)- go s nLeft =- if nLeft == 0- then return $ SFrame $ ST.runST $ FS.inCoreAoS $ Streamly.map fromS s- else do- a <- S.get- go (Streamly.cons a s) (nLeft - 1)-{-# INLINEABLE sframeGetC #-}---- Binary-instance (V.RMap rs, FI.RecVec rs, RecBinary rs) => B.Binary (SFrame (F.Record rs)) where- put = streamlyPutB . Streamly.map toS . Streamly.fromFoldable . unSFrame- {-# INLINEABLE put #-}- get = sframeGetB- {-# INLINEABLE get #-}---- we use only one fold to get the length and build the bytestream-streamlyPutB :: B.Binary a => Streamly.SerialT Identity a -> B.Put-streamlyPutB s = do- let lengthF = Streamly.Fold.length- putF = Streamly.Fold.Fold (\b a -> return $ b <> B.put a) mempty return- (l, streamPut) = runIdentity $ Streamly.fold (Streamly.Fold.tee lengthF putF) s- B.putWord64be $ fromIntegral l- streamPut-{-# INLINEABLE streamlyPutB #-}--sframeGetB :: forall rs. (FI.RecVec rs, V.RMap rs, RecBinary rs) => B.Get (SFrameRec rs)-sframeGetB = go Streamly.nil =<< B.getWord64be where- go :: (forall s.Streamly.SerialT (ST.ST s) (F.Rec SElField rs)) -> Word64 -> B.Get (SFrameRec rs)- go s nLeft =- if nLeft == 0- then return $ SFrame $ ST.runST $ FS.inCoreAoS $ Streamly.map fromS s- else do- a <- B.get- go (Streamly.cons a s) (nLeft - 1)-{-# INLINEABLE sframeGetB #-}
− examples/Strictness.hs
@@ -1,222 +0,0 @@-{-# LANGUAGE AllowAmbiguousTypes #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE QuasiQuotes #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE TypeApplications #-}-{-# LANGUAGE TypeOperators #-}---{-# OPTIONS_GHC -O0 #-}-module Main where--import StrictnessPaths--import qualified Data.Text as T-import qualified Data.Serialize as S-import qualified Data.ByteString as BS-import qualified Data.ByteString.Lazy as BL-import qualified Data.ByteString.Builder as BB-import qualified ByteString.StrictBuilder as BSB-import qualified Control.Foldl as FL-import qualified Control.Monad.State as ST-import qualified Data.Word as Word-import qualified Frames as F-import qualified Frames.Streamly.CSV as FStreamly-import qualified Frames.Streamly.InCore as FStreamly-import qualified Serialize as FS-import qualified Frames.CSV as Frames-import qualified Streamly.Data.Fold as Streamly.Fold-import qualified Streamly.Internal.Data.Fold.Types as Streamly.Fold-import qualified Streamly.Internal.Data.Fold as Streamly.Fold-import qualified Streamly.Prelude as Streamly-import qualified Streamly.Internal.Prelude as Streamly-import qualified Streamly as Streamly-import qualified Streamly.Internal.FileSystem.File- as Streamly.File-import qualified Streamly.External.ByteString as Streamly.ByteString--import qualified Streamly.Internal.Data.Array as Streamly.Data.Array-import qualified Streamly.Internal.Memory.Array as Streamly.Memory.Array--import qualified System.Clock--import qualified Data.Vector as V-import qualified Data.Vector.Mutable as VM-import qualified Control.Monad.Primitive as Prim--import qualified Data.Strict.Either as Strict-import qualified Data.Strict.Maybe as Strict--F.tableTypes' pumsACS1YrRowGen--main :: IO ()-main= do- testInIO--encodeOne :: S.Serialize a => a -> BB.Builder-encodeOne !x = S.execPut $ S.put x--bldrToCT = Streamly.ByteString.toArray . BL.toStrict . BB.toLazyByteString--encodeBSB :: S.Serialize a => a -> BSB.Builder-encodeBSB !x = BSB.bytes $! encodeBS x--encodeBS :: S.Serialize a => a -> BS.ByteString-encodeBS !x = S.runPut $! S.put x---bsbToCT = Streamly.ByteString.toArray . BSB.builderBytes--data Accum b = Accum { count :: !Int, bldr :: !b }--streamlySerializeF :: forall c bldr m a ct.(Monad m, Monoid bldr, c a, c Word.Word64)- => (forall b. c b => b -> bldr)- -> (bldr -> ct)- -> Streamly.Fold.Fold m a ct-streamlySerializeF encodeOne bldrToCT = Streamly.Fold.Fold step initial extract where- step (Accum n b) !a = return $ Accum (n + 1) (b <> encodeOne a)- initial = return $ Accum 0 mempty- extract (Accum n b) = return $ bldrToCT $ encodeOne (fromIntegral @Int @Word.Word64 n) <> b-{-# INLINEABLE streamlySerializeF #-}--toCT :: BSB.Builder -> Int -> Streamly.Memory.Array.Array Word8-toCT bldr n = bsbToCT $ encodeBSB (fromIntegral @Int @Word.Word64 n) <> bldr--streamlySerializeF2 :: forall c bldr m a ct.(Monad m, Monoid bldr, c a, c Word.Word64)- => (forall b. c b => b -> bldr)- -> (bldr -> ct)- -> Streamly.Fold.Fold m a ct-streamlySerializeF2 encodeOne bldrToCT =- let fBuilder = Streamly.Fold.Fold step initial return where- step !b !a = return $ b <> encodeOne a- initial = return mempty- toCT' bldr n = bldrToCT $ encodeOne (fromIntegral @Int @Word.Word64 n) <> bldr- in toCT' <$> fBuilder <*> Streamly.Fold.length--- extract (Accum n b) = return $ bldrToCT $ encodeOne (fromIntegral @Int @Word.Word64 n) <> b-{-# INLINEABLE streamlySerializeF2 #-}--type SmallRow = [PUMSYEAR, PUMSPERWT, PUMSSTATEFIP, PUMSREGION, PUMSPUMA, PUMSMETRO, PUMSDENSITY, PUMSAGE]--testInIO :: IO ()-testInIO = do- let pumsCSV = "example_data/acs100k.csv"- putTextLn "Testing File.toBytes..."- let rawBytesS = Streamly.File.toBytes pumsCSV- rawBytes <- Streamly.fold Streamly.Fold.length rawBytesS- putTextLn $ "raw PUMS data has " <> show rawBytes <> " bytes."- putTextLn "Testing readTable..."- let sPUMSRawRows :: Streamly.SerialT IO PUMS_Raw- = FStreamly.readTableOpt Frames.defaultParser pumsCSV- iRows <- Streamly.fold Streamly.Fold.length sPUMSRawRows- putTextLn $ "raw PUMS data has " <> (T.pack $ show iRows) <> " rows."- putTextLn "Testing Frames.Streamly.inCoreAoS:"- fPums <- FStreamly.inCoreAoS sPUMSRawRows- putTextLn $ "raw PUMS frame has " <> show (FL.fold FL.length fPums) <> " rows."-- putTextLn "frame to vec"- vPostFrame :: V.Vector PUMS_Raw <- FL.foldM FL.vectorM fPums- putTextLn $ "vPostFrame has " <> show (V.length vPostFrame) <> " rows"-- putTextLn "frame to array"- aPostFrame :: Streamly.Data.Array.Array PUMS_Raw <- Streamly.fold Streamly.Data.Array.write $ Streamly.fromFoldable fPums- putTextLn $ "aPostFrame has " <> show (Streamly.Data.Array.length aPostFrame) <> " rows"--{-- -- Previous goes up to 28MB, looks like via doubling. Then to 0 (collects fPums after counting?)- -- This one then climbs to 10MB, rows are smaller. No large leaks.- let f :: PUMS_Raw -> F.Record SmallRow- f !x = F.rcast x- putTextLn "Testing Frames.Streamly.inCoreAoS with row transform:"- fPums' :: F.FrameRec SmallRow <- FStreamly.inCoreAoS $ Streamly.map f sPUMSRawRows- putTextLn $ "transformed PUMS frame has " <> show (FL.fold FL.length fPums') <> " rows."- putTextLn "v1"--- sDict = KS.cerealStreamlyDict- let countFold = runningCountF "reading..." (\n -> "read " <> show (250000 * n) <> " rows") "finished"- sPUMSRunningCount = Streamly.map f- $ Streamly.tapOffsetEvery 250000 250000 countFold sPUMSRawRows- sPUMSRCToS = Streamly.map FS.toS sPUMSRunningCount-- serializedBytes :: Streamly.Memory.Array.Array Word.Word8 <- Streamly.fold (streamlySerializeF2 @S.Serialize encodeBSB bsbToCT) sPUMSRCToS- print $ Streamly.Memory.Array.length serializedBytes-- putTextLn "v4"- bldr <- Streamly.foldl' (\acc !x -> let b = S.runPut (S.put x) in b `seq` (acc <> BSB.bytes b)) mempty sPUMSRCToS- print $ BS.length $ BSB.builderBytes bldr-- putTextLn "In stages"- putTextLn "Word8: "- w8Array <- Streamly.Data.Array.fromStream $ Streamly.File.toBytes pumsCSV- putTextLn $ "w8 array is " <> show (Streamly.Data.Array.length w8Array) <> " long."-- putTextLn "To Vector"- putTextLn "Word8: "- vWord8 <- streamToVector $ FStreamly.streamWord8 pumsCSV- putTextLn $ "w8 vector is " <> show (V.length vWord8) <> " long."-- putTextLn "text lines (via vector)"- vTextLine <- streamToVector $ FStreamly.streamTextLines pumsCSV- putTextLn $ "vTextLine has " <> show (V.length vTextLine) <> " elements."-- putTextLn "text lines (via mutable vector)"- mvTextLine <- streamToVector2 100 $ FStreamly.streamTextLines pumsCSV- putTextLn $ "mvTextLine has " <> show (V.length mvTextLine) <> " elements."-- putTextLn "tokenized (via vector)"- vTokenized <- streamToVector $ FStreamly.streamTokenized pumsCSV- putTextLn $ "vTokenized has " <> show (V.length vTokenized) <> " elements."-- putTextLn "parsed (via vector)"- vParsed :: V.Vector (F.Rec (Strict.Either Text F.:. F.ElField) (F.RecordColumns PUMS_Raw)) <- streamToVector $ FStreamly.streamParsed pumsCSV- putTextLn $ "vParsed has " <> show (V.length vParsed) <> " elements."-- putTextLn "parsedMaybe (via vector)"- vParsedMaybe :: V.Vector (F.Rec (Maybe F.:. F.ElField) (F.RecordColumns PUMS_Raw)) <- streamToVector $ FStreamly.streamParsedMaybe pumsCSV- putTextLn $ "vParsedMaybe has " <> show (V.length vParsedMaybe) <> " elements."--}-{-- putTextLn "Copy to boxed array"- array <- Streamly.Data.Array.fromStream $ sPUMSRunningCount- putTextLn $ "array has " <> show (Streamly.Data.Array.length array) <> " elements."--}--streamToVector :: Monad m => Streamly.SerialT m a -> m (V.Vector a)-streamToVector = V.unfoldrM strictUncons--strictUncons :: Monad m => Streamly.SerialT m a -> m (Maybe (a, Streamly.SerialT m a))-strictUncons s = do- lu <- Streamly.uncons s- case lu of- Nothing -> return Nothing- Just (!a, s') -> return $ Just (a, s')-{-# INLINE strictUncons #-}--streamToVector2 :: (Monad m, Prim.PrimMonad m) => Int -> Streamly.SerialT m a -> m (V.Vector a)-streamToVector2 initialSize s0 = do- let go curLength curN s v = do- mNext <- Streamly.uncons s- case mNext of- Nothing -> V.unsafeFreeze (VM.unsafeTake curN v)- Just (!a, s') -> do- (curLength', v') <- if curN < curLength- then return (curLength, v)- else VM.grow v (2 * curLength) >>= \v' -> return (2 * curLength, v')- VM.write v' curN a- go curLength' (curN + 1) s' v'- v0 <- VM.new initialSize- go initialSize 0 s0 v0---runningCountF :: ST.MonadIO m => T.Text -> (Int -> T.Text) -> T.Text -> Streamly.Fold.Fold m a ()-runningCountF startMsg countMsg endMsg = Streamly.Fold.Fold step start done where- start = ST.liftIO (putText startMsg) >> return 0- step !n _ = ST.liftIO $ do- t <- System.Clock.getTime System.Clock.ProcessCPUTime- putStr $ show t ++ ": "- putTextLn $ countMsg n- return (n+1)- done _ = ST.liftIO $ putTextLn endMsg
− examples/StrictnessPaths.hs
@@ -1,20 +0,0 @@-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeOperators #-}-module StrictnessPaths where--import qualified Frames as F-import qualified Frames.TH as F-import qualified Data.Text as T--pumsACS1YrCSV :: FilePath-pumsACS1YrCSV = "example_data/acs100k.csv"---pumsACS1YrRowGen = (F.rowGen pumsACS1YrCSV) { F.tablePrefix = "PUMS"- , F.separator = ","- , F.rowTypeName = "PUMS_Raw"- }--type PUMSSTATEFIP = "STATEFIP" F.:-> Int