Parquet output first version

project.clj

@@ -7,6 +7,7 @@
                  [com.boundary/high-scale-lib "1.0.6"]
                  [com.google.protobuf/protobuf-java "3.22.0"]
                  [com.google.protobuf/protobuf-java-util "3.22.0"]
+                 [com.taoensso/nippy "3.4.2"]
                  [org.hdrhistogram/HdrHistogram "2.2.2"]
                  [com.taoensso/nippy "3.4.2"]
                  [io.netty/netty-all "4.1.89.Final"]
@@ -18,6 +19,12 @@
                  [io.netty/netty-resolver "4.1.89.Final"]
                  [fr.mcorbin/corbihttp "0.35.0"]
                  [org.apache.httpcomponents.client5/httpclient5 "5.3.1"]
+                 [org.apache.parquet/parquet-common "1.14.1"]
+                 [org.apache.parquet/parquet-encoding "1.14.1"]
+                 [org.apache.parquet/parquet-column "1.14.1"]
+                 [org.apache.parquet/parquet-hadoop "1.14.1"]
+                 ;[org.apache.hadoop/hadoop-common "3.4.0"]
+                 ;[org.apache.hadoop/hadoop-mapreduce-client-core "3.4.0"]
                  [org.apache.kafka/kafka-clients "3.4.0"]
                  [org.clj-commons/byte-streams "0.3.2"]
                  [ch.qos.logback/logback-classic "1.2.11"]

src/clojure/mirabelle/action.clj

@@ -2119,7 +2119,7 @@
     (info))
   ```
 
-  In this example, events will be forwarded of the value of the `:state` key
+  In this example, events will be forwarded if the value of the `:state` key
   is the same for at least 10 seconds.
 
   Support nested fields:

src/clojure/mirabelle/output/.parquet.poc

@@ -0,0 +1,68 @@
+
+
+;; this was fun to write
+
+(def riemann
+  {:tags [{:tag :binary}]
+   :host :binary
+   :service :binary
+   :state :binary
+   :metric :double
+   :time :required/double
+   :attributes [{:key :binary
+                 :value :binary}]})
+
+(defn repetition
+  [k]
+  (cond
+    (= "optional" (namespace k)) Type$Repetition/OPTIONAL
+    (= "required" (namespace k)) Type$Repetition/REQUIRED
+    :default Type$Repetition/OPTIONAL))
+
+(defn rm-ns
+  [k]
+  (when (keyword? k)
+      (-> k name keyword)))
+
+(defn definition->schema
+  [def state]
+  (for [[k v] def]
+    (cond
+      (= (rm-ns v) :int64) (let [type (PrimitiveType. (repetition v)
+                                              PrimitiveType$PrimitiveTypeName/INT64
+                                              (name k))]
+                     (swap! state assoc k type)
+                     type)
+
+      (= (rm-ns v) :binary) (let [type (PrimitiveType. (repetition v)
+                                               PrimitiveType$PrimitiveTypeName/BINARY
+                                               (name k))]
+                      (swap! state assoc k type)
+                      type)
+
+      (= (rm-ns v) :float) (let [type (PrimitiveType. (repetition v)
+                                              PrimitiveType$PrimitiveTypeName/FLOAT
+                                              (name k))]
+                     (swap! state assoc k type)
+                     type)
+
+      (= (rm-ns v) :double) (let [type (PrimitiveType. (repetition v)
+                                               PrimitiveType$PrimitiveTypeName/DOUBLE
+                                               (name k))]
+                      (swap! state assoc k type)
+                      type)
+
+      (map? v)
+      (let [type (GroupType. Type$Repetition/OPTIONAL ^String (name k)
+                             ^"[Lorg.apache.parquet.schema.Type;" (into-array Type (definition->schema v) state))]
+        (swap! state assoc (-> k name (str "-map") keyword) type)
+        type)
+
+      (sequential? v)
+      (let [type (GroupType. Type$Repetition/REPEATED
+                             ^String (name k)
+                             ^"[Lorg.apache.parquet.schema.Type;"
+                             (into-array Type
+                                         (definition->schema (first v) state)))]
+        (swap! state assoc (-> k name (str "-list") keyword) type)
+        type))))