User-friendly Hadoop-based re-indexing/compaction

[gianm]

#1374 is an implementation of being able to reindex Druid segments using the Hadoop indexer. There should be some user friendly way of doing this on an ongoing basis (for cleaning up realtime-ingested segments).

I think this could be done as a "HadoopReindexTask" that takes a really simple spec, just "dataSource", "intervals", and tuning stuff like a partitionsSpec and whatnot. It would then scan the timeline for the provided "intervals", find sections where there are segments that are unbalanced (just by looking at their metadata, not by actually opening them) and then reindex those sections. The idea is that if segments in those intervals are fine the way they are, the task should do nothing. The idea is also that if you run the task twice, it shouldn't do anything the second time.

It could be useful to have some override that forces the reindexing to always happen. This could be useful if the automatic thing doesn't make the decision that you wanted it to make.

I think to be really user friendly this will depend on storing aggregatorFactories in segment metadata (#1514), since otherwise you'd have to provide them to the HadoopReindexTask.

If this were implemented then it could be automated by just regularly submitting HadoopReindexTasks with intervals = 0000 to (now - X) where X is a few days or whatever.

[himanshug]

@guobingkun can you pls look into this.

[guobingkun]

@himanshug sure, will do.

[xvrl]

@gianm would this essentially be a "HadoopMergeTask"?

[gianm]

@xvrl sorta, except hopefully better in a few ways,

1. you can provide intervals rather than segments
2. it's idempotent
3. ideally you don't have to provide the aggregator specs to the task (this depends on Store aggregatorFactories in segment metadata #1514)

[himanshug]

@xvrl @gian
as I am talking to @guobingkun , I am imagining, "HadoopMergeTask" will be a simple wrapper around "HadoopIndexTask" that will be more like a convenient way(with some auto filling such as that of granularity, aggregators etc) of calling "HadoopIndexTask" with dataSource pathSpec in it .

[guobingkun]

I have been thinking letting Coordinator find unbalanced segment sections and submit HadoopReindexTask for those unbalanced sections.

Basically, we can create a runnable that Coordinator periodically executes.

At start of this runnable, for each dataSource, Coordinator will build a timeline of segments,
then for each segment's shards, we tag it by either S or O

S: the size of shard is below optimal size
O: the size of shard >= optimal size

Optimal size should be configured externally, we could probably just reuse Coordinator's "mergeBytesLimit".

Here is a timeline of segments for a specific dataSource (Note: The segment granularity for each segment could be arbitrary)

SegID:    1   2   3   4   5   6   7   8   9  10  11  12   13
Shard0: |_O_|_O_|_O_|_O_|_S_|_S_|_S_|_O_|_S_|_O_|_S_|_O_|_S_|__|__|__|__|__|__|__|
Shard1:         |_S_|                   |_S_|   |_S_|_S_|
Shard2:         |_S_|                   |_S_|   |_S_|

Coordinator will periodically scan this timeline, and submit HadoopReindexTask for unbalanced sections. The end result we want to achieve is that there are no shards in the timeline that are tagged by S.

Here is a relatively concrete algorithm that Coordinator could use for this runnable,

currTotalSize = 0
currInterval = null
shouldMergeNow = false

for each segment
  if currInterval == null
    currInterval = segment.getInterval()
  else 
    currInterval = currInterval.withEnd(segment.getEnd())
  for each shard of current segment
    if shard.getSize() < optimal size
      shouldMergeNow = true
    currTotalSize += shard.getSize()

  if currTotalSize >= optimal size 
    if shouldMergeNow 
      // Note: I still need to think of a routine that determines the segment granularity for 
      // HadoopReindexTask so that the segments generated by it have evenly distributed sizes.
      submitHadoopReindexTask(dataSource, currInterval) 

    // at this point, we are certain segments in currInterval are balanced
    currTotalSize = 0
    currInterval = null
    shouldMergeNow = false
  else {
    // do nothing. since the current segment doesn't have enough total size to make a segment 
    // with size >= optimal size, we'll seek to merge it with the next segment in the timeline
  }

if currInterval != null (this check is same as if currTotalSize > 0)
  // To discuss: if the last chunk doesn't have enough size, what should we do about it?
  // Some options, 
  // 1. We don't do anything with it, simply wait for new segments coming in, and the 
  // next Coordinator's run will do the right thing.
  // 2. Find a way to merge it with a previous optimal segment
  // 3. ?

Basically, the algorithm iterates each segment in the timeline, and for each shard in that segment, we check if its size is below optimal size, if yes, we should definitely try to merge that shard with others. However, it's possible that the total size of shards in that segment is not big enough to make an optimal size segment, so we should seek to merge the current segment with the next one in the timeline.

This algorithm should handle the cases below,

Case 1:
Segment 3 has three shards, two of which have sub-optimal sizes. In this case, we should reindex the interval that segment 3 covers, such that all its shards have optimal size.

Case 2:
Segments 5, 6, 7 are all small, so we continue iterating until we find Segment 8, now the current total size is >= optimal size, we can submit a reindex task that cover 5, 6, 7, 8.

Case 3:
Similar to Case 2, Segment 9 has three shards, but the total size of its shards are still less than optimal size, so merge it with Segment 10.

Case 5: The total size of all the segments in the history are not big enough, there is nothing we can do in this case, we just have to wait for new segments coming in such that we have enough size.

Case 6: Segment 13 is left over, should we merge it with previous segment or wait for new segments coming in?

Case 7: ?

[guobingkun]

Tacked by #1998

[stale]

This issue has been marked as stale due to 280 days of inactivity. It will be closed in 2 weeks if no further activity occurs. If this issue is still relevant, please simply write any comment. Even if closed, you can still revive the issue at any time or discuss it on the dev@druid.apache.org list. Thank you for your contributions.

[stale]

This issue has been closed due to lack of activity. If you think that is incorrect, or the issue requires additional review, you can revive the issue at any time.