Druid: Support directly reading and writing Druid data from Spark

[JulianJaffePinterest]

Motivation

To paraphrase the design doc, Druid is an excellent OLAP tool that aims to serve real-time analytics. Apache Spark, on the other hand, is a de-facto standard in the industry for big data processing, which is mostly batch-oriented by nature. While Druid has a few ways to query and ingest data, most of them are not suitable to be used from within ETL workflows, and specifically - via Spark (a few use-cases are detailed in this slack thread).

For ingesting data, the primary concern is the inefficiency of needing to write data produced in a Spark application to an intermediate location, using additional resources to read the data back into memory and prepare it for ingestion, and then finally ingesting the data into Druid via a Druid indexing task. If a Spark application can write files directly to deep storage in the format needed by Druid, these wasteful intermediate steps can be avoided.

On the reading side, some users want to use Spark to perform more complicated or arbitrary operations on data stored in Druid than Druid supports, or to join Druid data against other data stores, or simply to create very large reports. While this could be done via the existing Druid query interfaces, most of these use cases are batch-oriented, and so are both computationally expensive to produce and unlikely to be repeated with any frequency. To avoid degrading the performance of a Druid cluster for interactive workloads or triggering thrash as segments page in and out of memory, Spark applications could instead read the necessary data directly from the backing segment files.

Proposed changes

There are 3 primary ways to read and write data in Spark: the original RDD APIs, the DataSource V1 API, and the DataSource V2 API. A brief overview of some of the pros and cons of each is below:

RDD APIs

• Pros
  
  
  
  • Most flexible
  
  
• Cons
  
  
  
  • Unable to take full advantage of most Spark improvements over the years such as the Catalyst optimizer, predicate pushdown and input pruning on the read side
  
  
  • No built-in support for transactional writes and clean up
  
  
  • Difficult for users to tune properly
  
  
  • Doesn't use standard Spark data APIs (spark.read.format("druid") and df.write.format("druid"))
  
  

DataSource V1

• Pros
  
  
  
  • Stable API (supported across all 2.x releases)
  
  
• Cons
  
  
  
  • In the process of being supplanted by DataSource API V2
  
  
  • Still coupled to legacy Spark internals (SQLContext, etc.)
  
  
  • Doesn't handle writing very well
  
  
  • Difficult to push down predicates
  
  

DataSource V2

• Pros
  
  
  
  • Current focus of Spark development efforts
  
  
  • Supports predicate pushdown, better column pruning, transactional writes, columnar reads, streaming reads and writes, and more
  
  
• Cons
  
  
  
  • Evolving API: We can support Spark 2.4, which was released in November 2018, but not earlier versions. We will need to make relatively minor code changes to support Spark 3.x, which has a circulating release candidate but no clear release data and will need time to see widespread adoption.
  
  
  • Little partitioning information is available to the writer
  
  

This presentation goes into more detail if desired.

Because DataSource V2 is more powerful and full-featured and is the current focus of Spark development efforts, we should target this API for our development efforts. Moreover, since Spark 2.4 has been released for a year and a half and Spark traditionally sees slow uptake across major versions, Spark 2.4 will likely continue to be the dominant deployed version for some time, lessening the argument for designing against the upcoming 3.x modifications to the API.

Direct Reader

The direct Spark reader will have two modes of operation:

1. In the first and most common case, users will provide a data source name, one or more intervals (via Spark filters), and optionally various connection and configuration parameters. We will issue segment metadata queries to determine the schema of data source, query the metadata store to obtain the load specs for the active segments covering the desired intervals for the given data source, and then read each segment into a partition.
   
   
   
   1. Ideally, we could construct the schema purely from queries to the INFORMATION_SCHEMA.COLUMNS and not have to issue potentially expensive segment metadata queries over large intervals. However, the INFORMATION_SCHEMA tables don't contain information on the specific complex serde used to encode a metric or on whether or not a particular column has entries with multiple values (see #9707).
   
   
   2. Similarly, the sys tables exposed through the SQL query interface do not contain the load specs for the given segments, so segment locations will need to be queried through the coordinator API or the metadata server. Since we need to interact with the metadata server for the writer anyway and querying the metadata server is much more performant than querying the coordinator API for segment loadspecs, we will read segment locations on deep storage via SQL queries to the metadata server.
   
   
2. In the second case, we will allow users to provide a list of segments directly and bypass querying the metadata store. This will support cases where users have alternative ways of determining segment locations.

Direct Writer

Perhaps the biggest disadvantage of the two DataSource APIs vis-à-vis the older RDD APIs is the lack of control over the partitioning of the input data. This gives users much greater control over how data is partitioned, but also allows them to easily make suboptimal choices without meaning too. Druid requires each segment to have extensive information about broader partitioning choices in order to take advantage of most of its features (e.g. a segment must know how many other segments there are in an interval to support atomic updates, and must know _which_ segment of n it is to support contiguity checks and minor compaction, etc.). Without this information, Druid loses some flexibility and consistency guarantees.

Frustratingly, Spark shares very little information with its writers (only a partition's id, as well as task and epoch ids). This information is insufficient on its own to construct most Druid shard specs. To work around this, we can allow users to optionally provide additional partition information as options to the writers while falling back to the knowledge-less case if they don't. This will allow users who require certain Druid features to achieve them without unduly burdening users who don't.

At a high level, the Druid writer will work by constructing one or more incremental indices per partition (one per segment interval with rows in the partition unless a user provided more information) and then merging these together into segment files, publishing them to deep storage, and reporting the resulting data segments to the driver. If all partitions successfully write their segments, Spark will update the metadata store directly and the Druid coordinator will begin the loading process the next time it runs. If one or more partition write fails, Spark will delete all temporary files associated with the write attempt on its executors and delete the already written segments from deep storage. In this way, the writer should avoid polluting deep storage with segments from failed writes, and users won't be left with partially updated data sources.

Registries to shadow extensions

Because reading and writing data will happen on Spark clusters instead of Druid clusters, we can't take full advantage of the extensibility and dependency injection of Druid (e.g. we can't support custom metadata stores or complex metrics we don't know about the same way Druid does). To support custom extensions, we can use a registry pattern instead. By default, we can register the core implementations of various extension points and expose public APIs that allow users to register their own as well. Where we need use certain features in code, such as when we interact with deep storage or construct shard specs, we can pattern match against the registered functions for the given task. As with passing partitioning information to the writer, this pattern allows users who need more complex functionality to take advantage of it while "just working" for users who don't.

Rationale

The basic motivation for direct readers and writers in Spark is discussed above, as are the pros and cons of the various Spark APIs for reading data into and writing data out of Spark applications. Fuller discussions of the benefits and drawbacks of various technical implementations in Spark as well as existing third-party alternatives can be found in the design doc.

Operational impact

There will be no operational impacts on existing Druid clusters. This proposal would add readers and writers that can be called from within a Spark job.

Test plan

Because the reader and writer are symmetric, we can add integration tests that write segments to ephemeral local storage and read them back and verify correctness. Beyond these integration tests and unit tests, the biggest challenge will be testing all the various deep storage and metadata server possibilities. Here we will have to rely to some degree on the existing test support for the underlying extensions.

Future work

The primary goal of this proposal is to add readers and writers with all core features supported. The Spark DataSource V2 API allows us to envision support for many future enhancements, even if they're not part of the initial scope. Some highlights include:

• Columnar reads: Because one way Druid data is stored on disk is columnar, we can take advantage of Spark's support for columnar reads to greatly increase the read speed of many workloads.
• Structured streaming support for reading: The DataSource API V2 optionally supports reading updating data into structured streams. We can use this to support continuous reading from Druid data sources into Spark streaming jobs. The writer can also support outputting data from continually updating tables, but it probably makes more sense for users to simply emit records to Kafka or Kinesis and allow Druid indexing tasks to handle the ingestion since other users of Druid won't be able to query data being produced in Spark until it's been written to deep storage and loaded on historicals.
• Statistics: If we do use segment metadata queries to extract schema information for the reader, we can optionally pull segment statistics as well and use those to plan more efficient reads.

Annex

Spark read/write to Druid - design doc

[JulianJaffePinterest]

I haven't had the time I'd hoped to have over the past month to finish this up, so the code stands at about the same place: code is complete and working as expected, but the necessary pieces to make this actually useful (documentation) is still in progress. The outstanding work is to sanitize the writer configs, document everything, and figure out how to test this with other deep storage and metadata server technologies (we run this with segments stored on S3 and use a MySQL instance for our metadata server). I'm posting a pointer to the code here to save people having to dig through the design doc and to hopefully spur myself to carve out more time to resolve these issue.

[mangrrua]

Hello @JulianJaffePinterest

Do you still work on this connector? We may improve this connector, but for the first version your source code seems good.

[averma111]

@mangrrua and @JulianJaffePinterest I am very thankful to your guys for the work of spark connector to druid. Can I package this jar and make it working with current druid 0.18.1 version?

[averma111]

Also as per the design document https://github.com/metamx/druid-spark-batch is option for writing into Druid. but the build is failing. Can I get some help.

[soumyajose0784]

@JulianJaffePinterest , we were able start druid servuces with the spark extension built from github code. Could you please share other configurations and ingestion spec sample for data ingestion from spark

[xsqian]

@mangrrua and @JulianJaffePinterest I am very thankful to your guys for the work of spark connector to druid. Can I package this jar and make it working with current druid 0.18.1 version?

like to get some response on this question as well. Thanks!

[averma111]

@xsqian May get some response on this,

[bowenli86]

Hi folks, any update on this?

[JulianJaffePinterest]

No update from my side on this. I haven't had time to work on this and it appears that there isn't community appetite to _support_ direct Spark readers and writers (there is certainly community desire for such a feature though!). I haven't left this in a great state - the code works, and I suppose if you wanted to work backwards from the DruidDataSourceOptionKeys you could even make it work for you but the documentation is mostly in code comments instead of a useful readme. Even the usual first step of just working backwards from the tests doesn't help too much here since the whole point of these connectors is to hide the various APIs behind the Spark DataSourceV2 API - everything interesting happens in the options passed along in the .options() call (e.g. spark.read.format('druid').options(Map(DruidDataSourceOptionKeys.brokerHostKey -> '<my broker hostname>', ...))) so the key piece for usability is the documentation. The best pointers I can give you there are that the reader is easier to get working, and can mostly be read off from the DruidDataSourceOptionsKeys. The writer is trickier, as discussed in the proposal and in the code. Basically, in order to effectively use the output in Druid you'll need to use a custom partitioner in Spark and pass along a map to the writer to work around the limited information Spark passes across. This pattern is pretty anti-user and so one of the aims of this proposal was to start a discussion on ways to improve the situation, but that hasn't happened yet.

@averma111, @mangrrua, and any others interested in working on this proposal/packaging the code/etc., my code is licensed under the same license as the Druid project, as confirmed in the license headers on each file. I would love to see the community pick up where I left off.

[JulianJaffePinterest]

I had a little free time so I pulled in the most recent changes to mainline Druid and started a readme. There are a few minor signature changes that I blindly migrated to, but down the road we should actually adopt the new approaches - they will be more performant. I also added a simple readme with tables of the various config options and some sample code.

[JulianJaffePinterest]

Had some more time, so I added a round-trip example to the test suites that should help demonstrate how to use these connectors. The major outstanding piece is porting the partitioner from the druid-spark-batch repo, updating it, and adding some scala magic to make it easier to glue the partitioning and the writing together.

[JulianJaffePinterest]

A little more time 😃. I've implemented a solution for the partitioning problem mentioned above - the writer now "rationalizes" shard specs just before committing them, resulting in contiguous and complete time chunks. This means that linear and numbered shard specs need no special information from users. I've also added scaffolding and a sample partitioner to illustrate how providing partition mapping metadata can be done for more complex shard specs such as hashed or single dimension. I also added more documentation around partitioning. Once the deep storage config serialization is fixed up, this should be drop-in ready (it's pretty clean and easy to use at the moment, modulo pushing segments to deep storage which still needs some custom work).

[JulianJaffePinterest]

I cleaned up the docs a little more and added a few more tests. I don't have access to Azure or GCS blob storage to actually implement default segment writer registration functions and not every class the constructors require are serializable, so I haven't implemented working versions. While I do have access to S3, I use custom SegmentPushers and SegmentKillers and so the same caveats apply to the shipped s3 segment writer functions. Local and HDFS deep storage segment writers work out of the box, and I'm hopeful that people who use the various cloud providers' blob storages for deep storage will contribute actual working default implementations as they build them 🤞.

With the admittedly big exception of some of the deep storage writers, this is pretty much done. I think this process has pretty conclusively demonstrated that there isn't sufficient community support for maintaining Spark connectors as part of mainline Druid, so I'll leave this issue open but the next step will likely be to move these connectors to a standalone repository and ask for a pointer to be added to the Druid docs.

[JulianJaffePinterest]

Fixed up the deep storage writers for S3 and GCS, added support for password providers in metadata server connections, and improved the documentation.

[b-slim]

@JulianJaffePinterest Let me start by saying this is great work and would love to share with you some of the experience we had when building a similar connector between Hive and Druid.
The main point I want to share with you is why we picked the Hive Repo as the best place for such a connector.

• The major APIs used of the integration are the Hive Splits and Partitions I think In your case it is DatasourceV2 thus by having the code within Hive Repo we had a better compile-time dependency check and faster iteration (no need to public release to get things working with the bleeding-edge master) while most of the Druid APIs are RestFull and easy to get around without public releases of Druid.
• From a testing perspective we found it is way simple to run the tests within the Hive Integration test since most of the work and heavy lifting is done by Hive (in your case will be spark) again this was a big win for us.
• From a feature set perspective we found ourselves doing some extra work in Hive Partition logic or Planning logic to enable some bloom filter pushdown or well-balanced druid segments and again by having the connector leaving within Hive repo made things way simpler.

Those are the 2 cents that I want to share and should be considered as opinion and not review

[JulianJaffePinterest]

@b-slim Your points are well taken. There are a few differences between how the Spark and Hive projects run their codebases, and in how this extension is implemented, that I think change the calculus a little bit.

• For this connector, the Spark API used is public while the Druid APIs used are internal.
• The Spark project is much more reticent to pull in DataSourceV2 etc. implementations than the Hive project is to pull in storage handlers.

You're definitely right that there are some awkward edges in trying to fit this extension into the main Druid codebase, and it may not make sense to do so at the end of the day.