support for writing directly to stdout

cwensel · cwensel · commit e0b5e9c6bfe4 · 2025-05-19T14:02:15.000-07:00
this allows reading parquet to stdout directly from the command line via the --output option

--output-fields may also be provided to project only the declared fields
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/factory/TapFactories.java b/tessellate-main/src/main/java/io/clusterless/tessellate/factory/TapFactories.java
@@ -13,6 +13,7 @@
 import io.clusterless.tessellate.factory.hdfs.ParquetFactory;
 import io.clusterless.tessellate.factory.hdfs.TextFSFactory;
 import io.clusterless.tessellate.factory.local.LocalDirectoryFactory;
+import io.clusterless.tessellate.factory.local.StdOutFactory;
 import io.clusterless.tessellate.model.Sink;
 import io.clusterless.tessellate.model.Source;
 import io.clusterless.tessellate.options.PipelineOptions;
@@ -101,7 +102,7 @@ public static SinkFactory findSinkFactory(Sink sinkModel) {
         List<URI> inputUris = sinkModel.uris();
 
         if (inputUris.isEmpty()) {
-            return null;
+            return new StdOutFactory();
         }
 
         Format format = sinkModel.schema().format();
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/factory/local/StdOutFactory.java b/tessellate-main/src/main/java/io/clusterless/tessellate/factory/local/StdOutFactory.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright (c) 2023-2025 Chris K Wensel <chris@wensel.net>. All Rights Reserved.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+package io.clusterless.tessellate.factory.local;
+
+import cascading.nested.json.local.JSONTextLine;
+import cascading.scheme.Scheme;
+import cascading.scheme.local.CompressorScheme;
+import cascading.scheme.local.TextDelimited;
+import cascading.scheme.local.TextLine;
+import cascading.scheme.util.DelimitedParser;
+import cascading.tap.Tap;
+import cascading.tap.local.StdOutTap;
+import cascading.tuple.Fields;
+import io.clusterless.tessellate.factory.SinkFactory;
+import io.clusterless.tessellate.model.Dataset;
+import io.clusterless.tessellate.model.Schema;
+import io.clusterless.tessellate.model.Sink;
+import io.clusterless.tessellate.options.PipelineOptions;
+import io.clusterless.tessellate.util.Compression;
+import io.clusterless.tessellate.util.Format;
+import io.clusterless.tessellate.util.Models;
+import io.clusterless.tessellate.util.Protocol;
+import io.clusterless.tessellate.util.json.JSONUtil;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Properties;
+import java.util.Set;
+
+public class StdOutFactory implements SinkFactory {
+    @Override
+    public Set<Protocol> getSinkProtocols() {
+        return Set.of(Protocol.stdout);
+    }
+
+    @Override
+    public Set<Format> getFormats() {
+        return Set.of(Format.tsv, Format.csv, Format.json);
+    }
+
+    @NotNull
+    protected Scheme<Properties, InputStream, OutputStream, ?, ?> createScheme(Dataset dataset, Fields declaredFields) {
+        CompressorScheme.Compressor compressor = null;
+
+        Scheme<Properties, InputStream, OutputStream, ?, ?> scheme;
+
+        Schema schema = dataset.schema();
+        Format format = schema.format() == null ? Format.csv : schema.format();
+
+        switch (format) {
+            default:
+            case text:
+                Fields sinkFields = new Fields("line");
+                Fields sourceFields = dataset.schema().embedsSchema() ? new Fields("num", "line").applyTypes(Long.TYPE, String.class) : sinkFields;
+                scheme = new TextLine(sourceFields, sinkFields, compressor);
+                break;
+            case delimited:
+                DelimitedParser delimited = new DelimitedParser(schema.delimiterChar(), schema.quoteChar(), null, schema.strictParsing(), true);
+                return new TextDelimited(declaredFields, compressor, schema.embedsSchema(), schema.embedsSchema(), delimited);
+            case csv:
+                DelimitedParser csv = new DelimitedParser(",", schema.quoteChar(), null, schema.strictParsing(), true);
+                scheme = new TextDelimited(declaredFields, compressor, schema.embedsSchema(), csv);
+                break;
+            case tsv:
+                DelimitedParser tsv = new DelimitedParser("\t", schema.quoteChar(), null, schema.strictParsing(), true);
+                scheme = new TextDelimited(declaredFields, compressor, schema.embedsSchema(), tsv);
+                break;
+            case json:
+                scheme = new JSONTextLine(JSONUtil.DATA_MAPPER, declaredFields, compressor) {
+                    @Override
+                    public String getExtension() {
+                        return format.extension();
+                    }
+                };
+                break;
+        }
+
+        return scheme;
+    }
+
+    @Override
+    public Tap<Properties, ?, ?> getSink(PipelineOptions pipelineOptions, Sink sinkModel, Fields currentFields) throws IOException {
+        Fields declaredFields = Models.fieldAsFields(sinkModel.schema().declared(), String.class, Fields.ALL);
+
+        Fields resultFields = declaredFields.isAll() ? currentFields : declaredFields;
+
+        return new StdOutTap(createScheme(sinkModel, resultFields));
+    }
+
+    @Override
+    public Set<Compression> getCompressions() {
+        return Set.of(Compression.none);
+    }
+}
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/model/Field.java b/tessellate-main/src/main/java/io/clusterless/tessellate/model/Field.java
@@ -11,6 +11,7 @@
 import cascading.tuple.Fields;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import io.clusterless.tessellate.parser.FieldsParser;
 
 import java.util.Arrays;
@@ -27,8 +28,15 @@ public static List<Field> asField(String... declarations) {
         return Arrays.stream(declarations).map(Field::new).collect(Collectors.toList());
     }
 
+    // Constructor for deserialization from JSON string
     @JsonCreator
-    public Field(String declaration) {
+    public static Field fromString(String declaration) {
+        Objects.requireNonNull(declaration, "field may not be null");
+        return new Field(declaration);
+    }
+
+    @JsonCreator
+    public Field(@JsonProperty("declaration") String declaration) {
         Objects.requireNonNull(declaration, "field may not be null");
 
         this.declaration = declaration;
@@ -39,8 +47,24 @@ public Fields fields() {
         return fields;
     }
 
+    public String declaration() {
+        return declaration;
+    }
+
     @Override
     public String toString() {
         return declaration;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        Field field = (Field) o;
+        return Objects.equals(fields, field.fields) && Objects.equals(declaration, field.declaration);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(declaration);
+    }
 }
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/model/Sink.java b/tessellate-main/src/main/java/io/clusterless/tessellate/model/Sink.java
@@ -12,6 +12,7 @@
 
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 public class Sink implements Dataset, Model {
@@ -55,6 +56,10 @@ public boolean hasManifest() {
 
     @Override
     public List<URI> uris() {
+        if (output() == null) {
+            return Collections.emptyList();
+        }
+
         return List.of(output());
     }
 
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/options/FieldConverter.java b/tessellate-main/src/main/java/io/clusterless/tessellate/options/FieldConverter.java
@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) 2023-2025 Chris K Wensel <chris@wensel.net>. All Rights Reserved.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+package io.clusterless.tessellate.options;
+
+import io.clusterless.tessellate.model.Field;
+import picocli.CommandLine;
+
+public class FieldConverter implements CommandLine.ITypeConverter<Field> {
+    @Override
+    public Field convert(String value) throws Exception {
+        return new Field(value);
+    }
+}
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/options/OutputOptions.java b/tessellate-main/src/main/java/io/clusterless/tessellate/options/OutputOptions.java
@@ -8,13 +8,20 @@
 
 package io.clusterless.tessellate.options;
 
+import io.clusterless.tessellate.model.Field;
+import io.clusterless.tessellate.util.Format;
 import picocli.CommandLine;
 
 import java.net.URI;
+import java.util.List;
 
 public class OutputOptions implements AWSOptions {
     @CommandLine.Option(names = {"-o", "--output"}, description = "output uris")
     private URI output;
+    @CommandLine.Option(names = {"--output-fields"}, description = "output fields", converter = FieldConverter.class, split = "[\\+,]")
+    private List<Field> outputFields;
+    @CommandLine.Option(names = {"--output-format"}, description = "output format")
+    private Format outputFormat;
     @CommandLine.Option(names = {"-t", "--output-manifest-template"}, description = "output manifest uri template")
     private String outputManifestTemplate;
     @CommandLine.Option(names = {"-l", "--output-manifest-lot"}, description = "output lot")
@@ -37,6 +44,22 @@ public OutputOptions setOutput(URI output) {
         return this;
     }
 
+    public List<Field> outputFields() {
+        return outputFields;
+    }
+
+    public void setOutputFields(List<Field> outputFields) {
+        this.outputFields = outputFields;
+    }
+
+    public Format outputFormat() {
+        return outputFormat;
+    }
+
+    public void setOutputFormat(Format outputFormat) {
+        this.outputFormat = outputFormat;
+    }
+
     public String outputManifestTemplate() {
         return outputManifestTemplate;
     }
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/options/PipelineOptionsMerge.java b/tessellate-main/src/main/java/io/clusterless/tessellate/options/PipelineOptionsMerge.java
@@ -48,6 +48,8 @@ public class PipelineOptionsMerge {
             .putInto("inputManifestLot", "/source/manifestLot")
             .putInto("inputErrors", "/source/errorPath")
             .putInto("output", "/sink/output")
+            .putInto("outputFields", "/sink/schema/declared")
+            .putInto("outputFormat", "/sink/schema/format")
             .putInto("outputManifestTemplate", "/sink/manifestTemplate")
             .putInto("outputManifestLot", "/sink/manifestLot")
             .putInto("outputErrors", "/sink/errorPath");
@@ -68,6 +70,8 @@ public class PipelineOptionsMerge {
         argumentLookups.put("inputManifestLot", pipelineOptions -> nullOrNode(pipelineOptions.inputOptions().inputLot()));
         argumentLookups.put("inputErrors", pipelineOptions -> nullOrNode(pipelineOptions.inputOptions().inputErrors()));
         argumentLookups.put("output", pipelineOptions -> nullOrNode(pipelineOptions.outputOptions().output()));
+        argumentLookups.put("outputFields", pipelineOptions -> nullOrNode(pipelineOptions.outputOptions().outputFields()));
+        argumentLookups.put("outputFormat", pipelineOptions -> nullOrNode(pipelineOptions.outputOptions().outputFormat()));
         argumentLookups.put("outputManifestTemplate", pipelineOptions -> nullOrNode(pipelineOptions.outputOptions().outputManifestTemplate()));
         argumentLookups.put("outputManifestLot", pipelineOptions -> nullOrNode(pipelineOptions.outputOptions().outputLot()));
         argumentLookups.put("outputErrors", pipelineOptions -> nullOrNode(pipelineOptions.outputOptions().outputErrors()));
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/util/Protocol.java b/tessellate-main/src/main/java/io/clusterless/tessellate/util/Protocol.java
@@ -9,7 +9,7 @@
 package io.clusterless.tessellate.util;
 
 public enum Protocol {
-    file, hdfs, s3, http, https;
+    file, hdfs, s3, http, https, stdout;
 
     public static Protocol fromString(String protocol) {
         switch (protocol) {
@@ -23,6 +23,8 @@ public static Protocol fromString(String protocol) {
                 return http;
             case "https":
                 return https;
+            case "-":
+                return stdout;
             default:
                 throw new IllegalArgumentException("Unknown protocol: " + protocol);
         }
diff --git a/tessellate-main/src/test/java/io/clusterless/tessellate/pipeline/PipelineOptionsMergerTest.java b/tessellate-main/src/test/java/io/clusterless/tessellate/pipeline/PipelineOptionsMergerTest.java
@@ -10,11 +10,13 @@
 
 import com.adelean.inject.resources.junit.jupiter.GivenTextResource;
 import com.adelean.inject.resources.junit.jupiter.TestWithResources;
+import io.clusterless.tessellate.model.Field;
 import io.clusterless.tessellate.model.PipelineDef;
 import io.clusterless.tessellate.options.PipelineOptions;
 import io.clusterless.tessellate.options.PipelineOptionsMerge;
 import io.clusterless.tessellate.parser.ast.AssignmentStatement;
 import io.clusterless.tessellate.parser.ast.UnaryOperation;
+import io.clusterless.tessellate.util.Format;
 import io.clusterless.tessellate.util.json.JSONUtil;
 import org.junit.jupiter.api.Test;
 
@@ -50,17 +52,22 @@ void usingOptions(@GivenTextResource("/config/pipeline-mvel.json") String pipeli
     void usingOptionsMissing() throws IOException {
         List<URI> inputs = List.of(URI.create("s3://foo/input"));
         URI output = URI.create("s3://foo/output");
+        List<Field> declared = List.of(new Field("json"));
 
         PipelineOptions pipelineOptions = new PipelineOptions();
         pipelineOptions.inputOptions().setInputs(inputs);
         pipelineOptions.outputOptions().setOutput(output);
+        pipelineOptions.outputOptions().setOutputFields(declared);
+        pipelineOptions.outputOptions().setOutputFormat(Format.json);
 
         PipelineOptionsMerge merger = new PipelineOptionsMerge(pipelineOptions);
 
         PipelineDef merged = merger.merge();
 
         assertEquals(inputs, merged.source().inputs());
         assertEquals(output, merged.sink().output());
+        assertEquals(declared, merged.sink().schema().declared());
+        assertEquals(Format.json, merged.sink().schema().format());
     }
 
     @Test
