jnewhouse

Yep, pretty much. Right now filesystem^ sources are finite, scanning the target path at operator startup time and processing all matching files. This processing is done by opening an asynchronous reader, courtesy of the object_store crate.

^We call these Filesystem Sources/Sinks to match terminology present in other streaming systems, but I'm not in love with it.

Hi there! We actually already have a built-in Nexmark source. It's pretty useful for developing new capabilities, and available as a source out of the box.

Just read through the DBSP docs and it looks like it is working in a similar space. The biggest differences in my mind are around distribution and reliability. Arroyo works across a cluster of machines and has built in fault tolerance, while for DBSP that's still just planned for the future.

(I'm the co-creator of Arroyo, for context)

I don't think protobuf was around for public use when we came up with this format, which began around 2005. We use Protobuf internally, and some of our columns are actually byte[]'s containing protobuf data. We now support Parquet and are doing more work with other big data tools, but we've had a hard time matching the performance of our custom stuff.

It natively supports a limited set of Columns. Basically boxed primitives, java.util.Date, joda.time.DateTime, and arrays and double arrays of both boxed and unboxed versions of the preceding. The list of Columns being used is used to read and write to a byte buffer. The byte buffer is almost entirely the field's data, with one or two bytes describing how the subsequent field is encoded. Nested objects aren't handled out of the box, but there is the capability to define a UserRowField that allows for serialization/deserialization to bytes of any Serializable class. This gets used for our SQL map-reduce function a lot. The downside is that you need to have the UserRowField implementation in your classpath in order to read the Row, which is not generally the case.

Those switches all occurred at the pipeline level, leaving the map-reduce platform untouched. Switching our base logs to something like Parquet, Thrift or Protobuf would be a much larger project. We do support writing and reading Parquet to allow us to interface with other big data systems.

I think what I'd heard about was likely a poorly implemented use of Avro. I haven't actually worked with it.

We started developing rowfiles around 2005. Thrift wasn't open sourced until 2007. I couldn't find a date for protobuf's release, but I don't think it was standard outside of google at that time. We use protobufs internally, and have a number of Rows whose field values are byte[]s containing protobufs. One big thing our rowfiles gives us is fast indexing. The only other big data format I know of that gives that is Kudu, which uses the same indexing scheme.

If you want more details, we were packing a Row class into a base64 encoded string using an ObjectOutputStream. This is a fine thing for small scale serialization but sucks at scale, because of the reasons mentioned in the post. Sorry we don't have code examples, but it's unclear how useful it'd be given that no one else uses our file format. If you want a bit more detail on how the format works. Each metadata contains a list of typed columns to define the schema of a given part. Our map-reduce framework has a bunch of internal logic that tries to justify the written Row class with the one the Mapper class is asking for. This allows us to do things like ingest different versions of a row with in the context of a single job. I think questions of serialization at the scale are generally interesting, although ymmv. I know of one company using Avro, which doesn't let you cleanly update or track schema. They've ended up storing every schema in an HBase table and reserving the first 8 bytes to do a lookup into this table to know the row's schema.

A standard database table isn't large enough to handle our large datasets. For example, the Hercules dataset was over 2 petabytes and even after optimization is almost 1 petabyte. Big data systems like Spark, Impala, Presto, etc. are designed to make the data look like a table, even though it is spread out into many files in a distributed filesystem. This is what we do. It's pretty common to reimplement some database features onto these big data file formats. In our case we have very fast indexes that let us quickly fetch data, similar to an index in a postgresql table.