https://linen.dev logo
Join Slack
Powered by
Hi, I need to write a custom input row parser so t...
# general
k
Hi, I need to write a custom input row parser so that i can create multiple rows from single input row...something similar to this - https://github.com/apache/druid/issues/8446 I am trying to follow this guide to create a ExampleByteBufferInputRowParser - https://github.com/implydata/druid-example-extension But not sure on how to use it...when i try to paste the ingestion spec mentioned in Edit Spec console, i get errors...has anyone tried the same...would be helpful if i can get a sample data, ingestion spec and the steps to follow
g
the parser example there is pretty old; it implements a 
ByteBufferInputRowParser
but these days the preferred interface is 
InputFormat
in the 
apache/druid
repo you could look at 
AvroStreamInputFormat
or 
ParquetInputFormat
as some examples of how to write an input format
the 
implydata/druid-example-extension
is a good example still of how to structure an extension, i just wouldn't follow the 
ByteBufferInputRowParser
part specifically.
k
Copy code
package io.imply.druid.example.indexer;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.JsonParser.Feature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.druid.data.input.InputEntity;
import org.apache.druid.data.input.InputEntityReader;
import org.apache.druid.data.input.InputRowSchema;
import org.apache.druid.data.input.impl.NestedInputFormat;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.parsers.JSONPathSpec;
import org.apache.druid.utils.CompressionUtils;

import javax.annotation.Nullable;
import java.io.File;
import java.util.Collections;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;

public class TelemetryInputFormat extends NestedInputFormat
{
    public static final String TYPE_KEY = "telemetry";
    private final Map<String, Boolean> featureSpec;
    private final ObjectMapper objectMapper;
    private final boolean keepNullColumns;

    /**
     * Indicates whether or not the given InputEntity should be split by lines before parsing it.
     * If it is set to true, the InputEntity must be split by lines first.
     * If it is set to false, unlike what you could imagine, it means that the InputEntity doesn't have to be split by lines first, but it can still contain multiple lines.
     * A created InputEntityReader from this format will determine by itself if line splitting is necessary.
     *
     * This parameter should always be true for batch ingestion and false for streaming ingestion.
     * For more information, see: <https://github.com/apache/druid/pull/10383>.
     */
    private final boolean lineSplittable;

    /**
     * If true, overrides the behavior controlled by lineSplittable and the parsing will assume that the input
     * is newline delimited. This is to allow streaming ingestion to use the newline delimited parser when the
     * input is known to follow that format, since this allows for more flexible handling of parse errors (i.e.,
     * an invalid JSON event on one line will not prevent other events on different lines from being ingested).
     */
    private final boolean assumeNewlineDelimited;

    /**
     * If true, use the JsonNodeReader when parsing non-newline delimited JSON. This parser splits the input string
     * into JsonNode objects, parsing each JsonNode into a separate InputRow, instead of parsing the input string
     * into several InputRow at the same time. This allows for more flexible handling of parse errors, where timestamp
     * parsing errors do not prevent other events in the same string from being ingested, and allows valid events prior to
     * encountering invalid JSON syntax to be ingested as well.
     */
    private final boolean useJsonNodeReader;

    @JsonCreator
    public TelemetryInputFormat(
            @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec,
            @JsonProperty("featureSpec") @Nullable Map<String, Boolean> featureSpec,
            @JsonProperty("keepNullColumns") @Nullable Boolean keepNullColumns,
            @JsonProperty("assumeNewlineDelimited") @Nullable Boolean assumeNewlineDelimited,
            @JsonProperty("useJsonNodeReader") @Nullable Boolean useJsonNodeReader
    )
    {
        this(flattenSpec, featureSpec, keepNullColumns, true, assumeNewlineDelimited, useJsonNodeReader);
    }

    public TelemetryInputFormat(
            @Nullable JSONPathSpec flattenSpec,
            Map<String, Boolean> featureSpec,
            Boolean keepNullColumns,
            boolean lineSplittable,
            Boolean assumeNewlineDelimited,
            Boolean useJsonNodeReader
    )
    {
        super(flattenSpec);
        this.featureSpec = featureSpec == null ? Collections.emptyMap() : featureSpec;
        this.objectMapper = new ObjectMapper();
        if (keepNullColumns != null) {
            this.keepNullColumns = keepNullColumns;
        } else {
            this.keepNullColumns = flattenSpec != null && flattenSpec.isUseFieldDiscovery();
        }
        for (Entry<String, Boolean> entry : this.featureSpec.entrySet()) {
            Feature feature = Feature.valueOf(entry.getKey());
            objectMapper.configure(feature, entry.getValue());
        }
        this.lineSplittable = lineSplittable;
        this.assumeNewlineDelimited = assumeNewlineDelimited != null && assumeNewlineDelimited;
        this.useJsonNodeReader = useJsonNodeReader != null && useJsonNodeReader;
        if (this.assumeNewlineDelimited && this.useJsonNodeReader) {
            throw new IAE("useJsonNodeReader cannot be set to true when assumeNewlineDelimited is true.");
        }
    }

    @JsonProperty
    @JsonInclude(JsonInclude.Include.NON_EMPTY)
    public Map<String, Boolean> getFeatureSpec()
    {
        return featureSpec;
    }

    @JsonProperty // No @JsonInclude, since default is variable, so we can't assume false is default
    public boolean isKeepNullColumns()
    {
        return keepNullColumns;
    }

    @JsonProperty
    public boolean isAssumeNewlineDelimited()
    {
        return assumeNewlineDelimited;
    }

    @JsonProperty
    public boolean isUseJsonNodeReader()
    {
        return useJsonNodeReader;
    }

    @Override
    public boolean isSplittable()
    {
        return false;
    }

    @Override
    public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity source, File temporaryDirectory)
    {
//        System.out.println("inside createReader - " + this.toString());

        if (this.lineSplittable || this.assumeNewlineDelimited) {
//            System.out.println("inside TelemetryLineReader TelemetryLineReader TelemetryLineReader");
            return new TelemetryLineReader(inputRowSchema, source, getFlattenSpec(), objectMapper, keepNullColumns);
        } else if (this.useJsonNodeReader) {
//            System.out.println("inside TelemetryNodeReader TelemetryNodeReader TelemetryNodeReader");
            return new TelemetryNodeReader(inputRowSchema, source, getFlattenSpec(), objectMapper, keepNullColumns);
        } else {
//            System.out.println("inside TelemetryReader TelemetryReader TelemetryReader");
            return new TelemetryReader(inputRowSchema, source, getFlattenSpec(), objectMapper, keepNullColumns);
        }
    }

    @Override
    public long getWeightedSize(String path, long size)
    {
        CompressionUtils.Format compressionFormat = CompressionUtils.Format.fromFileName(path);
        if (CompressionUtils.Format.GZ == compressionFormat) {
            return size * CompressionUtils.COMPRESSED_TEXT_WEIGHT_FACTOR;
        }
        return size;
    }

    /**
     * Create a new TelemetryInputFormat object based on the given parameter
     *
     * sub-classes may need to override this method to create an object with correct sub-class type
     */
    public TelemetryInputFormat withLineSplittable(boolean lineSplittable)
    {
        return new TelemetryInputFormat(this.getFlattenSpec(),
                this.getFeatureSpec(),
                this.keepNullColumns,
                lineSplittable,
                assumeNewlineDelimited,
                useJsonNodeReader
        );
    }

    @Override
    public boolean equals(Object o)
    {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        if (!super.equals(o)) {
            return false;
        }
        TelemetryInputFormat that = (TelemetryInputFormat) o;
        return keepNullColumns == that.keepNullColumns
                && lineSplittable == that.lineSplittable
                && assumeNewlineDelimited == that.assumeNewlineDelimited
                && useJsonNodeReader == that.useJsonNodeReader
                && Objects.equals(featureSpec, that.featureSpec);
    }

    @Override
    public int hashCode()
    {
        return Objects.hash(
                super.hashCode(),
                featureSpec,
                keepNullColumns,
                lineSplittable,
                assumeNewlineDelimited,
                useJsonNodeReader
        );
    }

    @Override
    public String toString()
    {
        return "TelemetryInputFormat{" +
                "featureSpec=" + featureSpec +
                ", keepNullColumns=" + keepNullColumns +
                ", lineSplittable=" + lineSplittable +
                ", assumeNewlineDelimited=" + assumeNewlineDelimited +
                ", useJsonNodeReader=" + useJsonNodeReader +
                '}';
    }
}
Copy code
package io.imply.druid.example.indexer;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.druid.data.input.InputEntity;
import org.apache.druid.data.input.InputRow;
import org.apache.druid.data.input.InputRowSchema;
import org.apache.druid.data.input.TextReader;
import org.apache.druid.data.input.impl.MapInputRowParser;
import org.apache.druid.java.util.common.parsers.JSONFlattenerMaker;
import org.apache.druid.java.util.common.parsers.JSONPathSpec;
import org.apache.druid.java.util.common.parsers.ObjectFlattener;
import org.apache.druid.java.util.common.parsers.ObjectFlatteners;
import org.apache.druid.java.util.common.parsers.ParseException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class TelemetryLineReader extends TextReader {
    private final ObjectFlattener<JsonNode> flattener;
    private final ObjectMapper mapper;

    TelemetryLineReader(
            InputRowSchema inputRowSchema,
            InputEntity source,
            JSONPathSpec flattenSpec,
            ObjectMapper mapper,
            boolean keepNullColumns
    ) {
        super(inputRowSchema, source);
        this.flattener = ObjectFlatteners.create(
                flattenSpec,
                new JSONFlattenerMaker(
                        keepNullColumns,
                        inputRowSchema.getDimensionsSpec().useSchemaDiscovery()
                )
        );
        this.mapper = mapper;
    }

    @Override
    public List<InputRow> parseInputRows(String line) throws IOException, ParseException {
        try {
            final JsonNode document = mapper.readValue(line, JsonNode.class);
            final JsonNode telemetryNode = document.get("Telemetry");
            final JsonNode rowsNode = document.get("Rows");

            if (telemetryNode == null || rowsNode == null || !rowsNode.isArray()) {
                throw new RuntimeException("Invalid input JSON format: " + line);
            }

            final List<InputRow> inputRows = new ArrayList<>();

            for (JsonNode rowNode : rowsNode) {
                final ObjectNode outputNode = mapper.createObjectNode();
                outputNode.set("Source", document.get("Source"));
                outputNode.set("Telemetry", telemetryNode);
                outputNode.set("Row", rowNode);

                inputRows.add(MapInputRowParser.parse(getInputRowSchema(), flattener.flatten(outputNode)));
            }
            return inputRows;
        } catch (IOException e) {
            throw new IOException("Error parsing input line: " + line, e);
        }
    }

    @Override
    public List<Map<String, Object>> toMap(String intermediateRow) throws IOException {
        try {
            final JsonNode document = mapper.readTree(intermediateRow);
            final List<Map<String, Object>> outputRows = new ArrayList<>();

            final String source = document.has("Source") ? document.get("Source").asText() : null;
            final JsonNode telemetryNode = document.get("Telemetry");
            final JsonNode rowsNode = document.get("Rows");


            if (source != null && telemetryNode != null && rowsNode != null) {
                for (JsonNode rowNode : rowsNode) {
                    final Map<String, Object> outputRow = new HashMap<>();
                    outputRow.put("Source", source);
                    outputRow.put("Telemetry", mapper.treeToValue(telemetryNode, Map.class));
                    outputRow.put("Row", mapper.treeToValue(rowNode, Map.class));
                    outputRows.add(outputRow);
                }
            } else {
                throw new IOException("Invalid intermediate row format");
            }

            return outputRows;
        } catch (IOException e) {
            throw new IOException("Error converting intermediate row to map: " + intermediateRow, e);
        }
    }

    @Override
    public int getNumHeaderLinesToSkip() {
        return 0;
    }

    @Override
    public boolean needsToProcessHeaderLine() {
        return false;
    }

    @Override
    public void processHeaderLine(String line) {
        // No header line processing required
    }

}
Hi...i tried the above mentioned approach to explode single input row into multiple rows by just following the JsonInputFormat: 1. created a TelemetryInputFormat.java 2. created a TelemetryLineReader.java Worked once for input file with few lines of JSON...but getting 
Too many open files
exception repeatedly now Any pointers on how to fix this
g
is it possible your dataset has a very large # of columns? i think the 
AutoTypeColumnMerger
is opening some number of files per column
you might also have a too-small number of max open files on your system
you can raise it using the techniques at https://www.baeldung.com/linux/error-too-many-open-files
k
Copy code
{
  "Source": "192.168.57.78:34974",
  "Telemetry": {
    "node_id_str": "MRUTDNTLCCR001",
    "subscription_id_str": "telemetrycollection_INTERFACESTATS",
    "encoding_path": "Cisco-IOS-XR-pfi-im-cmd-oper:interfaces/interface-briefs/interface-brief",
    "collection_id": 363718,
    "collection_start_time": 1711514734786,
    "msg_timestamp": 1711514734786,
    "collection_end_time": 0
  },
  "Rows": [
    {
      "Timestamp": 1711514734793,
      "Keys": {
        "interface-name": "Bundle-Ether1"
      },
      "Content": {
        "actual-line-state": "im-state-up",
        "actual-state": "im-state-up",
        "bandwidth": 800000000,
        "bandwidth64-bit": 800000000,
        "encapsulation": "ether",
        "encapsulation-type-string": "ARPA",
        "interface": "Bundle-Ether1",
        "l2-transport": "false",
        "line-state": "im-state-up",
        "mtu": 9216,
        "state": "im-state-up",
        "sub-interface-mtu-overhead": 0,
        "type": "IFT_ETHERBUNDLE"
      }
    },
    {
      "Timestamp": 1711514734793,
      "Keys": {
        "interface-name": "Bundle-Ether1.1551"
      },
      "Content": {
        "actual-line-state": "im-state-up",
        "actual-state": "im-state-up",
        "bandwidth": 800000000,
        "bandwidth64-bit": 800000000,
        "encapsulation": "dot1q",
        "encapsulation-type-string": "802.1Q",
        "interface": "Bundle-Ether1.1551",
        "l2-transport": "false",
        "line-state": "im-state-up",
        "mtu": 9216,
        "parent-interface": "Bundle-Ether1",
        "state": "im-state-up",
        "sub-interface-mtu-overhead": 0,
        "type": "IFT_VLAN_SUBIF"
      }
    }
  ]
}
^this is the shortened version of a single payload...the "Rows" array can have 100s of objects I tried setting the max open files limit but seems like its not able to fix the issue is the code opening the file multiple times? any way to optimise the code here
g
hmm. it shouldn't be a problem if there's a lot of rows
can you tell what the open files are? maybe using 
lsof
k
^@Gian Merlino
g
i don't see any interesting open files here for 
java
processes, just a couple thousand files (pretty small number), mostly jars
perhaps it wasn't in the middle of doing anything when you captured the 
lsof
. it would be more interesting to see what the 
lsof
is like when it's almost about to say 'too many open files' (although not after, since after, it would have already closed them all and shut down)
alternatively, the task logs might be interesting
k
when running on a server i didnt face this problem...maybe it was specific to my macOS limits...thanks for the help although can you give some insight on this log and how to make ingestion more efficient -
Open in Slack
PreviousNext
Powered by