Clojurians Log v2

@dpsutton I'm crawling some data and generating ~ 20GB of uncompressed data per day (the watcher appends to gzipped version of the file directly). There are about 5000 files that I am writing to

> too many files to have one agent per file name > have the futures reset an agent...

I think we disagree about what agents are about? they are lightweight, in clojure terms at least, you can have a lot of them. and you send them actions - you wouldn't reset their file, you would close over the file handle and send them a function that makes them write. serializing writes is exactly the sort of thing agents are good at (as long as you aren't too worried about waiting for writes to finish before you move forward)

a quick example:

(require '[<http://clojure.java.io|clojure.java.io> :as io])

(defn writing-agent
  [fname]
  (let [handle (io/file fname)
        writer (io/writer handle)]
    (agent
     {:fresh (.createNewFile handle)
      :handle handle
      :writer writer
      :written 0})))

(defn append-to
  [the-agent the-string]
  (send the-agent
        (fn [{:keys [writer] :as m}]
          (.write writer the-string)
          (.flush writer)
          (update m :written + (count the-string)))))

(defn close
  [the-agent]
  (send the-agent
        (fn [{:keys [writer] :as m}]
          (.close writer)
          (assoc m :handle nil :writer nil))))

(cmd)user=> (.exists (io/file "foo.out"))
false
(cmd)user=> (def a (writing-agent "foo.out"))
#'user/a
(ins)user=> (pprint @a)
{:fresh false,
 :handle #object[java.io.File 0xff6077 "foo.out"],
 :writer
 #object[java.io.BufferedWriter 0x1280851e "java.io.BufferedWriter@1280851e"],
 :written 0}
nil
(ins)user=> (append-to a "hello")
#object[clojure.lang.Agent 0x12a160c2 {:status :ready, :val {:fresh false, :handle #object[java.io.File 0xff6077 "foo.out"], :writer #object[java.io.BufferedWriter 0x1280851e "java.io.BufferedWriter@1280851e"], :written 5}}]
(cmd)user=> (.exists (io/file "foo.out"))
true
(cmd)user=> (slurp "foo.out")
"hello"
(ins)user=> (append-to a ", world")
#object[clojure.lang.Agent 0x12a160c2 {:status :ready, :val {:fresh false, :handle #object[java.io.File 0xff6077 "foo.out"], :writer #object[java.io.BufferedWriter 0x1280851e "java.io.BufferedWriter@1280851e"], :written 5}}]
(cmd)user=> (slurp "foo.out")
"hello, world"
(ins)user=> (close a)
#object[clojure.lang.Agent 0x12a160c2 {:status :ready, :val {:fresh false, :handle #object[java.io.File 0xff6077 "foo.out"], :writer #object[java.io.BufferedWriter 0x1280851e "java.io.BufferedWriter@1280851e"], :written 12}}]
(ins)user=> (pprint @a)
{:fresh false, :handle nil, :writer nil, :written 12}
nil

I saw some weird behavior with .createNewFile btw- the docs say it creates a new fresh empty file only if it doesn't exist yet, but in my experiments it seems to be truncating the existing file before any writes

or maybe the writer I create is doing that - I should decouple for testing

i think you need {:append true} as options to the writer

Hum, ya it mixes weird with the (seq [1 2]) idiom though

Which made sense that you either have a seq with elements in it, or nil

But gets confusing when a seq can be empty

yes it used to be the case that (if s ... ...) was safe, but now(for years, the change was pretty close to clojure 1.0) you need (if (seq s) ... ...)

I guess in the taxonomy, what is the difference between s and what seq returns?

I'm trying to see if like there's a new word for a seq that can't be empty?

looking at it again, I think he is distinguishing between a seq and a lazy-seq (not a seq and a sequence), a seq is never empty (it is a seq of something or nil), a lazy-seq can be empty, and both of those are a sequence

but that doesn't help to make sense of the comment about (rest coll)

and like the type hint on rest says it returns an ISeq, the method call it bottoms out on (ISeq.more) says it returns an ISeq

My team would like to standardize on a Clojure library for API and/or function validation. • Clojure.spec can be used for these purposes but we've found difficulties in usage that make it unappealing for our org. • We're heavily considering Malli as our standardization target. I'm responsible for demo'ing this and other possibilities soon. Would anyone suggest other libraries or things to consider during this task?

I'm aware of Prismatic Schema but have never used it myself. We were previously having some success with Orchestra & Spec Tools, but we've generally seen low adoption into Clojure.spec.

I'm hoping spec2 will come out some time, I think the alpha status of spec1 and spec2 slows down momentum

Malli seems to be actively developed and embraced

That was our take on looking at the library. We're a larger enterprise and must be a little more risk adverse, but Malli seems very reasonable

I've been using malli for several of my projects (in tandem with Reitit) and it works very well imho.

In my own personal work, I've done some really hacky work on clojure.spec to get runtime specs working. I'm also really looking forward to spec2, but my fingers have been crossed for a long time there 🙂

Cognitect is committed to spec as part of the library. It’s inconceivable that spec won’t be supported. That’s not necessarily the case with malli

Even Schema has been maintained while the company behind it pivoted.

Good point, though I don't believe I'd have success trying to force adoption of spec1. I'll need to come to a conclusion by end of month, and I haven't checked but I'm assuming spec2 is still a ways out

As long as the community has a big interest in it, it will probably be maintained

And Metosin has a good track record I'd say

Yeah, I'll ➕1 that. Most of our team was pleasantly surprised to see Metosin was the publisher

If you decide go with Malli, there is #malli to get help with your demo. Also, 0.3.0 around the corner (parsers, function & sequence schemas).

Does there exist a ready Clojure solution for the following? I want to zip several very large files while streaming the result on the fly to a consumer. The ZIP compression algorithm allows sending chunks of the output even before the entire zip archives is complete, and I don't want to hold the entire thing in memory at any given time. Obviously, Java's ZipOutputStream is unsuitable for that. Alpakka has a solution for this problem in Java and Scala: https://doc.akka.io/docs/alpakka/current/file.html#zip-archive However, while I can certainly call Alpakka from Clojure, I don't want to drag in the dependency and Akka and have to initialize its actor systems just for this little thing. Any suggestions?

what makes you think ZipOutputStream requires holding the entire output in memory?

surely if the target it writes to isn't in memory, it doesn't need to hold the output in memory

Well, you cannot stream the results from its buffers as an input stream until it's done.

oh, that's counterintuitive, I'll read up, pardon my ignorance

Unless you can and it's me who's ignorant, of course. 🙂

I'd love to be proven wrong.

it takes an outputstream as a constructor argument (that part I did expect)

I want an input stream constructed out of several input streams, which simply zips their contents.

an input stream is something you read from :D

(simple typo I'm sure)

Yup. So I could read the zipped contents from even before the entire files are done being zipped.

Here's how Alpakka does it in Java:

Source<ByteString, NotUsed> source1 = ...
Source<ByteString, NotUsed> source2 = ...

Pair<ArchiveMetadata, Source<ByteString, NotUsed>> pair1 =
    Pair.create(ArchiveMetadata.create("akka_full_color.svg"), source1);
Pair<ArchiveMetadata, Source<ByteString, NotUsed>> pair2 =
    Pair.create(ArchiveMetadata.create("akka_icon_reverse.svg"), source2);

Source<Pair<ArchiveMetadata, Source<ByteString, NotUsed>>, NotUsed> source =
    Source.from(Arrays.asList(pair1, pair2));

Sink<ByteString, CompletionStage<IOResult>> fileSink = FileIO.toPath(Paths.get("logo.zip"));
CompletionStage<IOResult> ioResult = source.via(Archive.zip()).runWith(fileSink, mat);

oh - so you want data -> zipper -> (dupe) -> output, where the dupe step creates something you can read

that's a limit algorithmically - with that kind of buffering you always need to hold onto the unread backlog, no matter what you do

(unless I still misunderstand)

Yeah, kind of. If you look at how ZipOutputStream works, it reads a chunk at a time, deflates it, adds it to a buffer, updates the checksum, gets another chunk, etc. In the meantime, what's already processed can already be send downstream. That's what I want.

of course the (dupe) step can be implemented so it only holds backlog and doesn't keep read state

deflates?

That's their terminology for compression.

> The ZIP file format permits a number of compression algorithms, though DEFLATE is the most common.

So, something like this: Input: Files A.a, B.b, C.c. Zipped state: filename -> send right away downstream. first chunk -> compress -> send downsteam. second chunk -> compress -> send downstream. filename -> ...

Don't wait to send the entire thing downstream when everything is complete.

The files are too big for that.

it seems like the composition of output streams should allow this, if you use an IO based stream and don't use in-memory if I understand your sitution, the gotcha is that you want to be able to "tee" the output, you need something with sensible buffering behavior (if you control both producer and consumer you can make sure the buffer never goes too large in simple use cases) more generally you can get into backpressure and document that the stream will stall if the secondary consumer doesn't read fast enough or use a dropping buffer and document that the secondary reader will not see all the data if it reads too slowly

this is something that can be done with outputstreams

Well, I can get to solving the backpressure problem later. To begin with, what do you mean by "composition of output streams" in this particular scenario?

what I mean is that by chaining streams you can have two destinations for the zipped data - one you can read back in process, and one that gets sent to IO

if you don't need two destinations, your problem is simpler than I thought

I might have misunderstood your usage of the term "input stream" relating to the zipped data

based on your step by step, all you need is for the output stream you supply to ZipOutputStream to be an IO stream and not an in-memory buffer

Yes, I want something like ZipOutputStream (in the sense of building up a zipped data in its buffers) that can also be queried for available zipped data and read from--to wit, that also functions as an input stream.

I suppose that channels can be used for that, no?

channels?

core.async

this is totally unrelated to core.async, that will just make your problem more complicated

you can make an OutputStream that writes data to two other OutputStreams

you can use PipedInputStream and PipedOutputStream to turn your ZipOutputstream into an InputStream

you can make an OuputStream that sends to an InputStream

Thanks. I'll take a look at the piped streams.

the big caveat with the Piped*Streams is that you need two threads, whereas you might be able to get away with a single thread

@lkorogodski essentially, even without the Piped stuff, you can reify http://java.io.OutputStream and in the implementation simply delegate by writing to two other streams

this only needs one thread

(but will block if one reader blocks)

Ok, that will work, I think.

the reason I so glibly rejected core.async here is that the synchronization here is not actually complex and the mechanics of putting the problem into core.async leads to potential pitfalls (blocking IO should not be in go blocks for example)

and the core.async solution wouldn't simplify the management of the data consumption, just displace the complexity

also, solve one problem at a time

Thanks!

Yeah, backing a ZipOutputStream by an OutputStream that uses a CircularByteBuffer would do the trick, as the circular buffer has an associated input stream, too.

https://github.com/nedap/speced.def takes bit of a Schema-like approach (i.e. no instrumentation, using what essentially boils down to a pre/postcondition system) while using Spec1 as its backing validation system. The magic being that Spec1 is just an impl detail. I have a branch where I replaced it with Schema (because it's relevant for my day job). I could do the same for Spec2, Malli, etc. It's bit of an underdog but also a completely unique, robust proposition when considered carefully